From b74180e59556fa02b9b597f1727ab2dc0ae5c4dc Mon Sep 17 00:00:00 2001 From: boladeE Date: Mon, 21 Apr 2025 15:31:44 +0100 Subject: [PATCH] Refactor BrandStyleManager and main.py to integrate database functionality for campaign management. Replace JSON file handling with database calls for loading, saving, updating, and deleting campaigns. Update sample campaign loading to retrieve data from the database instead of JSON files. --- backend/brand_style.py | 37 +++++-------------- backend/database.py | 80 +++++++++++++++++++++++++++++++++++++++++ backend/main.py | 50 +++++--------------------- data/campaigns.db | Bin 0 -> 12288 bytes 4 files changed, 98 insertions(+), 69 deletions(-) create mode 100644 backend/database.py create mode 100644 data/campaigns.db diff --git a/backend/brand_style.py b/backend/brand_style.py index 6ef38d0..21d8bde 100644 --- a/backend/brand_style.py +++ b/backend/brand_style.py @@ -6,6 +6,7 @@ from PyPDF2 import PdfReader from embeddings import CohereEmbeddings from vector_store import VectorStore from config import settings +from database import Database class BrandStyleManager: def __init__(self, embeddings: CohereEmbeddings, vector_store: VectorStore): @@ -13,6 +14,7 @@ class BrandStyleManager: self.embeddings = embeddings self.vector_store = vector_store self.brand_voice = self._load_brand_voice() + self.db = Database() self.sample_campaigns = self._load_sample_campaigns() def _load_brand_voice(self) -> Dict[str, Any]: @@ -24,14 +26,8 @@ class BrandStyleManager: return {} def _load_sample_campaigns(self) -> List[Dict[str, Any]]: - """Load sample campaigns from JSON.""" - - file_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "data", "past_campaigns", "sample_campaigns.json")) - if os.path.exists(file_path): - with open(file_path, 'r', encoding='utf-8') as f: - data = json.load(f) - return data.get("campaigns", []) - return [] + """Load sample campaigns from the database.""" + return self.db.get_all_campaigns() def _extract_text_from_pdf(self, pdf_path: str) -> str: """Extract text from a PDF file.""" @@ -39,9 +35,7 @@ class BrandStyleManager: try: reader = PdfReader(pdf_path) for page in reader.pages: - page_text = page.extract_text() - if page_text: - text += page_text + "\n\n" + text += page.extract_text() + "\n" except Exception as e: print(f"Error extracting text from PDF: {e}") return text @@ -83,23 +77,10 @@ class BrandStyleManager: else: print("No content found to add to vector store") - def get_relevant_context(self, prompt: str, k: int = 5) -> List[Dict]: - """Get relevant context for a given prompt from book excerpts.""" - # Generate embedding for the prompt - prompt_embedding = self.embeddings.generate_embedding(prompt) - - # Search for similar content in book excerpts - results = self.vector_store.search(prompt_embedding, k=k) - - # Optionally rerank results - if results: - texts = [result["text"] for result in results] - reranked = self.embeddings.rerank_results(prompt, texts, top_n=k) - # Convert reranked results to the expected format - return [{"text": text} for text in reranked] - - # If no results, return empty list - return [] + def get_relevant_context(self, query: str, top_k: int = 3) -> List[Dict[str, str]]: + """Get relevant context from the vector store based on the query.""" + query_embedding = self.embeddings.generate_embedding(query) + return self.vector_store.search(query_embedding, k=top_k) def get_brand_voice(self) -> Dict[str, Any]: """Get brand voice guidelines.""" diff --git a/backend/database.py b/backend/database.py new file mode 100644 index 0000000..15a6672 --- /dev/null +++ b/backend/database.py @@ -0,0 +1,80 @@ +import sqlite3 +import os +from typing import List, Dict, Any +import datetime + +class Database: + def __init__(self): + data_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "data")) + self.db_path = os.path.join(data_dir, "campaigns.db") + self._init_db() + + def _init_db(self): + """Initialize the database with required tables.""" + with sqlite3.connect(self.db_path) as conn: + cursor = conn.cursor() + cursor.execute(''' + CREATE TABLE IF NOT EXISTS campaigns ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + prompt TEXT NOT NULL, + content TEXT NOT NULL, + timestamp TEXT NOT NULL + ) + ''') + conn.commit() + + def get_all_campaigns(self) -> List[Dict[str, Any]]: + """Retrieve all campaigns from the database.""" + with sqlite3.connect(self.db_path) as conn: + cursor = conn.cursor() + cursor.execute('SELECT prompt, content, timestamp FROM campaigns ORDER BY timestamp DESC') + rows = cursor.fetchall() + return [ + { + "prompt": row[0], + "content": row[1], + "timestamp": row[2] + } + for row in rows + ] + + def add_campaign(self, prompt: str, content: str) -> None: + """Add a new campaign to the database.""" + timestamp = datetime.datetime.now().isoformat() + with sqlite3.connect(self.db_path) as conn: + cursor = conn.cursor() + cursor.execute( + 'INSERT INTO campaigns (prompt, content, timestamp) VALUES (?, ?, ?)', + (prompt, content, timestamp) + ) + conn.commit() + + def update_campaign(self, index: int, content: str) -> bool: + """Update an existing campaign's content.""" + with sqlite3.connect(self.db_path) as conn: + cursor = conn.cursor() + cursor.execute('SELECT id FROM campaigns ORDER BY timestamp DESC LIMIT 1 OFFSET ?', (index,)) + result = cursor.fetchone() + if result: + campaign_id = result[0] + timestamp = datetime.datetime.now().isoformat() + cursor.execute( + 'UPDATE campaigns SET content = ?, timestamp = ? WHERE id = ?', + (content, timestamp, campaign_id) + ) + conn.commit() + return True + return False + + def delete_campaign(self, index: int) -> bool: + """Delete a campaign by its index.""" + with sqlite3.connect(self.db_path) as conn: + cursor = conn.cursor() + cursor.execute('SELECT id FROM campaigns ORDER BY timestamp DESC LIMIT 1 OFFSET ?', (index,)) + result = cursor.fetchone() + if result: + campaign_id = result[0] + cursor.execute('DELETE FROM campaigns WHERE id = ?', (campaign_id,)) + conn.commit() + return True + return False \ No newline at end of file diff --git a/backend/main.py b/backend/main.py index e208165..9fcd261 100644 --- a/backend/main.py +++ b/backend/main.py @@ -4,30 +4,17 @@ from typing import Optional, List, Dict, Any from copywriter import generate_marketing_copy from brand_style import BrandStyleManager from config import settings +from database import Database import os import json import datetime app = Flask(__name__) -# Initialize brand style manager +# Initialize brand style manager and database data_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "data")) campaign_prompt = [] - -def load_campaigns(): - campaigns_file = os.path.join(data_dir, "past_campaigns", "campaigns.json") - if os.path.exists(campaigns_file): - with open(campaigns_file, 'r', encoding='utf-8') as f: - data = json.load(f) - campaigns = data.get("campaigns", []) - return campaigns - return [] - -def save_campaigns(campaigns): - campaigns_file = os.path.join(data_dir, "past_campaigns", "campaigns.json") - os.makedirs(os.path.dirname(campaigns_file), exist_ok=True) - with open(campaigns_file, 'w', encoding='utf-8') as f: - json.dump({"campaigns": campaigns}, f, indent=4) +db = Database() @app.route('/', methods=['GET', 'POST']) def root(): @@ -41,7 +28,7 @@ def root(): @app.route('/campaigns') def view_campaigns(): - campaigns = load_campaigns() + campaigns = db.get_all_campaigns() return render_template('campaigns.html', campaigns=campaigns) @app.route('/save-edit', methods=['POST']) @@ -49,40 +36,21 @@ def save_edit(): edited_copy = request.form.get('editedCopy') global campaign_prompt prompt = campaign_prompt[-1] - campaigns = load_campaigns() - new_campaign = { - "prompt": prompt, - "content": edited_copy, - "timestamp": datetime.datetime.now().isoformat() - } - campaigns.append(new_campaign) - save_campaigns(campaigns) - + db.add_campaign(prompt, edited_copy) return render_template('index.html', generated_copy="Campaign saved successfully") @app.route('/update-campaign', methods=['POST']) def update_campaign(): index = int(request.form.get('index')) edited_copy = request.form.get('editedCopy') - - campaigns = load_campaigns() - if 0 <= index < len(campaigns): - campaigns[index]['content'] = edited_copy - campaigns[index]['timestamp'] = datetime.datetime.now().isoformat() - save_campaigns(campaigns) - + db.update_campaign(index, edited_copy) return redirect(url_for('view_campaigns')) @app.route('/delete-campaign', methods=['POST']) def delete_campaign(): index = int(request.form.get('index')) - - campaigns = load_campaigns() - if 0 <= index < len(campaigns): - campaigns.pop(index) - save_campaigns(campaigns) - + db.delete_campaign(index) return redirect(url_for('view_campaigns')) -if __name__ == "__main__": - app.run(host='localhost', port=8000, debug=True) \ No newline at end of file +if __name__ == '__main__': + app.run(debug=True) \ No newline at end of file diff --git a/data/campaigns.db b/data/campaigns.db new file mode 100644 index 0000000000000000000000000000000000000000..ffb8b90d9f9789ec3dbdc8307b870b3ab6582cca GIT binary patch literal 12288 zcmeI2PjB2r7{f@Hp5ekL;Op=i`VqKs;~CpowIvmts~Kez?|5e3_xU~VSea}deshvqrm51ou(TI^ z7<4+pr$j*ztl_na*JbPA!u4G?_x``yTCnxQpBospa{Z6s`rr8Gn(Gb&2_q>r4bzuM}sd0=X7>{ ze0p>~rmqKMI=UD=I38l9)4_1myK-isRbE&c4Za=G@Bv;ICnr~r43)Gj|JP8P=4>o9 zy}a(f61e5g<~Z~R@4fSS`2LP3;0bsFo`5Id33vjYfG6;#5P0@e=UQ<1>qpPZ`sZ6m zMaIG~#VlD1d@^QK&NfM9-0fb-OocNVt5VbFWtP#Ia=(wKA|oLq8fhVo`yj0uaj!I~ zWMjHrqK);9bxfl~%$aP$6zO!%z4~&VnpDYNy@b0}2;(}QKmBb|S`wP6!pte+Rh01@ zS;ff4b!swJvy5$G$SMjoyM)~urVMc|GWE=e4eL1{(xU`ISi8=Tu_{hF*>a&tn{1CJ zWtv5RYLv|w8K-P0l~l?I zi&#v*jA=NV)}ocm%232yWYty!tK0pOH9vp)yP-raY8|1FWW|ih9fP#WNKG3EI6zGy z7R->L6dJYx?m@%7Pu=cuBxmRjKKhk%XFD3QVVO7 zOr3A6)xsuqqHgt5VJxU(VqVHLL~tb$8-~;7@;|?0{xylv$)F+=trlBq+3<^%$eg7Rm}9Gyy3R+lDKl$Y`;K#0q)ZDj zb9R>nxx%d|5*4|P6(d>H(M+mETMmr_D3lN-6hldCZd=J)G(0p$ykGtW}I8-A-8u9o1&nCF{6fmcLVoOt48swh$E%3LLqk zWl^59h$?5T+dPm5z%Mv)2rC|1p)E!@rIQMtPcsHfKmbk!BqkSla-d`i7mSuwjYY8^ z;}&wmzp=2v-aIC@Xp}G>7WGC*ZJXFgHDg*>y=D(R9=k`x5P$IaZg+ckzrTH_zq>Qq z*+1CZKG?gnwR`uI{kuEv{NMTDuK-`Zc>