import logging import re from typing import Dict, Any, Optional from pydantic import BaseModel, Field from langchain_google_genai import ChatGoogleGenerativeAI from langchain.agents import create_agent from pymongo import MongoClient from datetime import datetime from string import Template from src.config import gemini_settings, mongo_settings from email_validator import validate_email logger = logging.getLogger(__name__) class EmailBatchOutput(BaseModel): """Structured output for the generated email outreach sequence.""" main_email_subject: str = Field(description="Subject line for the main email (short, curiosity-driven)") main_email_html: str = Field(description="HTML version of the main email") main_email_plain: str = Field(description="Plain text version of the main email") follow_up_1_subject: str = Field(description="Subject line for Follow-up 1") follow_up_1_body: str = Field(description="Follow-up 1 (Day 3): Gentle reminder + added value") follow_up_2_subject: str = Field(description="Subject line for Follow-up 2") follow_up_2_body: str = Field(description="Follow-up 2 (Day 7): Share case study / result") follow_up_3_subject: str = Field(description="Subject line for Follow-up 3") follow_up_3_body: str = Field(description="Follow-up 3 (Day 14): Breakup email (polite, non-pushy)") class EmailBatchGenerator: def __init__(self): self.model = ChatGoogleGenerativeAI( model="gemini-3-flash-preview", temperature=0.4, google_api_key=gemini_settings.api_key ) self.agent = create_agent( model=self.model, response_format=EmailBatchOutput ) self.mongo_client = MongoClient(mongo_settings.uri) self.db = self.mongo_client[mongo_settings.db] self.collection = self.db["generated_emails"] self.prompt_collection = self.db["ai_prompts"] self.system_instruction = """ You are an expert B2B cold email copywriter for a performance-driven digital marketing agency. Your task is to generate a high-converting cold email outreach sequence targeting local businesses. IMPORTANT RULES: - All emails are sent on behalf of **Omji Pandey, Director at Henceforth Solutions**. - Always include the sender signature as: Omji Pandey Director Henceforth Solutions - Never use placeholders such as [Name], [Business Name], [City], [Website], etc. - Write the email so it can be **sent directly to the client without any manual editing**. - The tone should be professional, conversational, and concise. - Focus on value, curiosity, and a soft call-to-action. - Avoid sounding spammy or overly salesy. ---------------------------------------- INPUT VARIABLES: ---------------------------------------- Business Category: $category Address: $address Target Audience: Local business owners in the above category Services Offered: - Local SEO (Google Maps ranking, GMB optimization, reviews growth) - SEO (on-page, technical, backlinks) - PPC Ads (Google Ads, Meta Ads) - Social Media Marketing (content + paid campaigns) - Website optimization (speed, conversion rate) Pain Points (auto-assume if not provided): - Low visibility on Google - Not ranking in top 3 map results - Poor website conversion - Not running ads or wasting ad spend - Weak or inconsistent social media presence RAG CONTEXT (Case studies / proof): $rag_context ---------------------------------------- EMAIL REQUIREMENTS: ---------------------------------------- Generate a cold email containing: 1. Subject line (short, curiosity-driven) 2. Hook (personalized observation or insight) 3. Value proposition (how Henceforth Solutions helps) 4. Social proof if available from RAG context 5. Soft CTA (ask if they are open to a quick chat or audit) Keep the email between **80–150 words**. Make it feel human-written, not AI-generated. """ self.user_instruction = """ Generate a personalized, non-spammy, highly relevant outreach email that feels human-written and tailored to the business. The email should: - Be concise (120–180 words) - Focus on value, not generic selling - Include subtle personalization placeholders - Avoid spammy phrases - Sound like a real person, not AI ---------------------------------------- OUTPUT COMPONENTS: ---------------------------------------- 1. MAIN EMAIL (Subject + HTML + Plain Text) - Provide a unique, non-salesy subject line. 2. FOLLOW-UP EMAILS (1, 2, 3) - For each follow-up, provide a separate, relevant SUBJECT and BODY. - Follow-up subjects should feel like a continuation (e.g., "Re:...", "Quick question regarding...", etc.) or offer a new angle. STRUCTURE: - Personalized opener referencing business - Observation (based on insight or common issue) - Value proposition (how we help) - Social proof (from RAG context) - Soft CTA (low friction) Use: business title: $title business website: $website business type: $business_type business tech stack: $tech_stack Seo problems: $identified_problems recommended actions: $recommended_actions 3. FOLLOW-UP EMAILS (3 total) Follow-up 1 (Day 3): Gentle reminder + added value Follow-up 2 (Day 7): Share case study / result Follow-up 3 (Day 14): Breakup email (polite, non-pushy) ---------------------------------------- WRITING STYLE: ---------------------------------------- - Conversational, human tone - No jargon overload - No long paragraphs - No "Dear Sir/Madam" - No hard selling - Avoid buzzwords like "synergy", "cutting-edge" - Keep sentences short and crisp ---------------------------------------- PERSONALIZATION RULES: ---------------------------------------- If insight is available: → Use it naturally in first 2 lines If no insight: → Use category + city based assumption Examples: - "noticed your clinic isn’t ranking in top results" - "saw your website could load faster" - "looks like you’re not running Google Ads" ---------------------------------------- VALUE ANGLES TO COVER (IMPORTANT): ---------------------------------------- The email must subtly touch at least 2–3 of these: - Local SEO → "get more calls from Google Maps" - SEO → "rank higher on search" - PPC → "generate instant leads" - Social Media → "improve brand visibility & engagement" Do NOT list all services mechanically — weave them naturally. ---------------------------------------- CTA RULES: ---------------------------------------- - Keep it soft and low commitment Examples: - “Would you be open to a quick audit?” - “Happy to share a few ideas if useful” - “Can I send you a quick analysis?” ---------------------------------------- ANTI-SPAM RULES: ---------------------------------------- - No ALL CAPS - No excessive exclamation marks - No "limited offer" - No "guaranteed results" - Avoid overly promotional tone """ self.rag_context = "Helped 50+ local businesses increase search traffic by 150% in 3 months." def _get_resolved_prompts(self, user_id: str, run_id: str) -> Dict[str, str]: """Fetch custom instructions from DB, or fallback to defaults.""" defaults = { "system": self.system_instruction, "user": self.user_instruction, "rag": self.rag_context } if not user_id: return defaults # 1. Check for specific run_id if run_id: specific = self.prompt_collection.find_one({"user_id": user_id, "run_id": run_id}) if specific: return { "system": specific.get("system_instruction") or defaults["system"], "user": specific.get("user_instruction") or defaults["user"], "rag": specific.get("rag_context") or defaults["rag"] } # 2. Check for universal prompt universal = self.prompt_collection.find_one({"user_id": user_id, "is_universal": True}) if universal: return { "system": universal.get("system_instruction") or defaults["system"], "user": universal.get("user_instruction") or defaults["user"], "rag": universal.get("rag_context") or defaults["rag"] } # 3. Fallback return defaults def generate_emails_for_lead(self, lead) -> Optional[Dict[str, Any]]: if not lead or not lead.email: logger.info("Lead has no email, skipping generation.") return None # Build prompt variables category = lead.category or "Local Business" address = lead.address or "Your area" rag_context = "Helped 50+ local businesses increase search traffic by 150% in 3 months." # Placeholder if we don't have real rag # For the user prompt variables title = lead.title or category website = lead.website or "your website" business_type = lead.business_type or category tech_stack = lead.tech_stack or [] identified_problems = lead.identified_problems or [] recommended_actions = lead.recommended_actions or [] user_id = getattr(lead, 'user_id', None) run_id = getattr(lead, 'run_id', None) resolved = self._get_resolved_prompts(user_id, run_id) system_prompt = Template(resolved["system"]).safe_substitute( category=category, address=address, rag_context=resolved["rag"] ) user_prompt = Template(resolved["user"]).safe_substitute( title=title, website=website, business_type=business_type, tech_stack=tech_stack, identified_problems=identified_problems, recommended_actions=recommended_actions ) try: logger.info(f"Generating emails for {title} ({lead.email})...") result = self.agent.invoke({ "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ] }) generated_content = result["structured_response"].dict() email_batch_doc = { "run_id": lead.run_id, "user_id": getattr(lead, 'user_id', None), "job_id": getattr(lead, 'job_id', None), "place_id": getattr(lead, 'place_id', None), "lead_email": lead.email, "title": title, "generated_emails": generated_content, "approval_status": "pending", "email_status": { "main": {"status": "pending", "sent_at": None}, "follow_up_1": {"status": "pending", "sent_at": None}, "follow_up_2": {"status": "pending", "sent_at": None}, "follow_up_3": {"status": "pending", "sent_at": None} }, "batch_version": 1, "metadata": { "website": website, "category": category }, "created_at": datetime.now(), "updated_at": datetime.now() } inserted = self.collection.insert_one(email_batch_doc) logger.info(f"Successfully saved email batch with id {inserted.inserted_id}") email_batch_doc["_id"] = str(inserted.inserted_id) return email_batch_doc except Exception as e: logger.error(f"Failed to generate/save email batch for {lead.title}: {e}") return None