Initial commit

backend/brand_style.py

@@ -0,0 +1,78 @@
+from typing import Dict
+import json
+from pathlib import Path
+
+
+class BrandStyle:
+    def __init__(self):
+        self.style = self._load_style()
+        self.templates = {
+            "email": self._load_template("email"),
+            "social": self._load_template("social"),
+            "general": self._load_template("general")
+        }
+
+    def _load_style(self) -> Dict:
+        path = Path("data/style_guidelines/style.json")
+        if path.exists():
+            return json.loads(path.read_text())
+        else:
+            # Create default style if doesn't exist
+            default_style = {
+                "tone": "professional",
+                "phrases": ["innovative", "results-driven", "empowering", "transformation"],
+                "avoid": ["cheap", "guarantee", "spam", "scam"]
+            }
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(json.dumps(default_style, indent=2))
+            return default_style
+
+    def _load_template(self, name: str) -> str:
+        path = Path(f"data/style_guidelines/{name}_template.txt")
+        if path.exists():
+            return path.read_text()
+        else:
+            # Create default templates if they don't exist
+            templates = {
+                "email": "Write a professional email that engages the reader and includes a clear call-to-action.",
+                "social": "Create an engaging social media post that captures attention and encourages interaction.",
+                "general": "Generate marketing copy that is compelling, authentic, and aligned with brand values."
+            }
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(templates.get(name, ""))
+            return templates.get(name, "")
+
+    def get_prompt(self, request) -> str:
+        # Handle both dict and Pydantic model objects
+        if hasattr(request, 'dict'):
+            # It's a Pydantic model
+            request_dict = request.dict()
+            tone = request.tone or self.style['tone']
+            content_type = request.content_type
+        else:
+            # It's already a dict
+            request_dict = request
+            tone = request_dict.get('tone', self.style['tone'])
+            content_type = request_dict.get('content_type', 'general')
+
+        return f"""
+You are a marketing assistant for Adriana James. Follow these brand guidelines:
+
+TONE: {tone}
+STYLE: Professional, authentic, and empowering
+
+PREFERRED PHRASES: {', '.join(self.style['phrases'])}
+AVOID USING: {', '.join(self.style['avoid'])}
+
+CONTENT TYPE: {content_type}
+TEMPLATE GUIDANCE: {self.templates.get(content_type, self.templates['general'])}
+
+Create marketing copy that:
+1. Reflects Adriana James' brand voice
+2. Is engaging and authentic
+3. Includes a clear value proposition
+4. Has a compelling call-to-action when appropriate
+5. Feels personal and relatable
+
+Remember: Focus on transformation, empowerment, and results while maintaining professionalism.
+"""

backend/config.py

@@ -0,0 +1,11 @@
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+class Config:
+    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+    COHERE_API_KEY = os.getenv("COHERE_API_KEY")
+    DATA_PATH = "data/past_campaigns/campaigns.json"
+    FAISS_INDEX_PATH = "data/vector_store.index"

backend/copywriter.py

@@ -0,0 +1,24 @@
+# backend/copywriter.py
+from .vector_store import VectorStore
+from .brand_style import BrandStyle
+import openai
+
+class Copywriter:
+    def __init__(self):
+        self.vector_store = VectorStore()
+        self.brand_style = BrandStyle()
+    
+    def generate_copy(self, request):
+        # Move the generation logic from main.py here
+        similar = self.vector_store.search(request.prompt, request.content_type)
+        
+        response = openai.ChatCompletion.create(
+            model="gpt-3.5-turbo",
+            messages=[
+                {"role": "system", "content": self.brand_style.get_prompt(request)},
+                {"role": "user", "content": f"Prompt: {request.prompt}\n\nSimilar examples:\n{similar}"}
+            ],
+            temperature=0.7
+        )
+        
+        return response.choices[0].message.content

backend/embeddings.py

@@ -0,0 +1,18 @@
+from config import Config
+import cohere
+import openai
+
+class Embeddings:
+    def __init__(self):
+        self.cohere = cohere.Client(Config.COHERE_API_KEY)
+    
+    def get_embedding(self, text: str, engine: str = "cohere"):
+        if engine == "cohere":
+            response = self.cohere.embed(texts=[text], model="small")
+            return response.embeddings[0]
+        else:  # OpenAI fallback
+            response = openai.Embedding.create(
+                input=[text],
+                model="text-embedding-ada-002"
+            )
+            return response.data[0].embedding

backend/main.py

@@ -0,0 +1,131 @@
+from fastapi import FastAPI, HTTPException, UploadFile, File
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+from typing import Optional, List
+import json
+from datetime import datetime
+from .vector_store import VectorStore
+from .brand_style import BrandStyle
+from .config import Config
+import openai
+import os
+
+# Initialize OpenAI
+openai.api_key = Config.OPENAI_API_KEY
+
+# Initialize
+app = FastAPI(title="Marketing Assistant AI", version="0.1.0")
+vector_store = VectorStore()
+brand_style = BrandStyle()
+
+# CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Models
+class CampaignRequest(BaseModel):
+    prompt: str
+    content_type: str = "general"
+    tone: Optional[str] = None
+
+class Campaign(BaseModel):
+    content: str
+    content_type: str
+    metadata: dict = {}
+
+# Routes
+@app.post("/generate")
+async def generate_copy(request: CampaignRequest):
+    """Generate marketing copy based on prompt and brand guidelines"""
+    try:
+        # Get similar content from vector store
+        similar = vector_store.search(request.prompt, request.content_type)
+        
+        # Format similar content for context
+        similar_content = ""
+        if similar:
+            similar_content = "\n\nSimilar past campaigns for reference:\n"
+            for i, campaign in enumerate(similar[:3], 1):
+                similar_content += f"{i}. {campaign.get('content', '')}\n"
+        
+        # Generate with OpenAI
+        system_prompt = brand_style.get_prompt(request)
+        user_prompt = f"Create marketing copy for: {request.prompt}{similar_content}"
+        
+        response = openai.ChatCompletion.create(
+            model="gpt-3.5-turbo",
+            messages=[
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_prompt}
+            ],
+            temperature=0.7,
+            max_tokens=500
+        )
+        
+        generated_copy = response.choices[0].message.content
+        
+        # Store the generated copy for future reference
+        new_campaign = {
+            "content": generated_copy,
+            "content_type": request.content_type,
+            "metadata": {
+                "prompt": request.prompt,
+                "tone": request.tone,
+                "generated_at": datetime.now().isoformat()
+            }
+        }
+        
+        # Add to vector store for future similarity searches
+        vector_store.add_campaign(new_campaign)
+        
+        return {"result": generated_copy}
+    
+    except Exception as e:
+        print(f"Error in generate_copy: {str(e)}")  # For debugging
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/add-campaign")
+async def add_campaign(campaign: Campaign):
+    """Add a new campaign to the vector store"""
+    try:
+        # Add timestamp to metadata
+        campaign_data = campaign.dict()
+        campaign_data["metadata"]["added_at"] = datetime.now().isoformat()
+        
+        vector_store.add_campaign(campaign_data)
+        return {"status": "success"}
+    except Exception as e:
+        print(f"Error in add_campaign: {str(e)}")  # For debugging
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/search")
+async def search_campaigns(query: str, limit: int = 5):
+    """Search for similar campaigns"""
+    try:
+        results = vector_store.search(query, k=limit)
+        return {"results": results}
+    except Exception as e:
+        print(f"Error in search_campaigns: {str(e)}")  # For debugging
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/")
+async def root():
+    """Health check endpoint"""
+    return {"message": "Marketing Assistant AI is running", "version": "0.1.0"}
+
+@app.get("/health")
+async def health_check():
+    """Detailed health check"""
+    return {
+        "status": "healthy",
+        "vector_store_size": len(vector_store.campaigns),
+        "timestamp": datetime.now().isoformat()
+    }
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

backend/requirements.txt

@@ -0,0 +1,7 @@
+fastapi
+uvicorn
+openai==0.28
+cohere
+sentence-transformers
+faiss-cpu
+python-dotenv

backend/vector_store.py

@@ -0,0 +1,37 @@
+import faiss
+import numpy as np
+from sentence_transformers import SentenceTransformer
+from .config import Config
+import json
+from pathlib import Path
+
+
+class VectorStore:
+    def __init__(self):
+        self.model = SentenceTransformer('all-MiniLM-L6-v2')
+        self.index = faiss.IndexFlatL2(384)
+        self.campaigns = []
+        self._load_data()
+
+    def _load_data(self):
+        if Path(Config.DATA_PATH).exists():
+            with open(Config.DATA_PATH) as f:
+                self.campaigns = json.load(f)
+                if self.campaigns:
+                    embeddings = self.model.encode([c["content"] for c in self.campaigns])
+                    self.index.add(embeddings)
+
+    def add_campaign(self, campaign: dict):
+        self.campaigns.append(campaign)
+        embedding = self.model.encode([campaign["content"]])
+        self.index.add(embedding)
+        self._save_data()
+
+    def search(self, query: str, content_type: str = None, k: int = 3):
+        query_embedding = self.model.encode([query])
+        distances, indices = self.index.search(query_embedding, k)
+        return [self.campaigns[i] for i in indices[0]]
+
+    def _save_data(self):
+        with open(Config.DATA_PATH, 'w') as f:
+            json.dump(self.campaigns, f)