import logging from typing import Optional from datetime import datetime from bson import ObjectId from pymongo import MongoClient import sqlalchemy as sa from src.config import mongo_settings, db_settings from src.database import engine from src.outreach.email_generator import EmailBatchGenerator from src.emailCampaigns.service import WoodpeckerService from src.emailCampaigns.models import AddProspectsV1Request, ProspectV1, CampaignIdWrapper from src.outreach.models import ( EmailBatchResponse, EmailBatchListResponse, RunIdSummary, RunIdListResponse, EmailPatchRequest, RegenerateResponse, DeleteResponse, BatchActionResponse, LeadResponse, LeadListResponse, PromptRequest, PromptResponse, PromptListResponse, ResolvedPromptResponse, ) logger = logging.getLogger(__name__) class OutreachService: """Service layer for CRUD operations on the generated_emails MongoDB collection.""" def __init__(self): self.mongo_client = MongoClient(mongo_settings.uri) self.db = self.mongo_client[mongo_settings.db] self.collection = self.db["generated_emails"] self.sync_collection = self.db["campaign_runs"] self.prompt_collection = self.db["ai_prompts"] self.email_generator = EmailBatchGenerator() self.woodpecker_service = WoodpeckerService() # ────────────────────────────────────────────── # Helpers # ────────────────────────────────────────────── @staticmethod def _serialize_doc(doc: dict) -> dict: """Convert MongoDB _id (ObjectId) to string for Pydantic serialization.""" if doc and "_id" in doc: doc["_id"] = str(doc["_id"]) return doc # ────────────────────────────────────────────── # READ — Run IDs # ────────────────────────────────────────────── def get_all_run_ids(self, user_id: str) -> RunIdListResponse: """Return all distinct run_ids with their email counts and latest timestamp.""" pipeline = [ {"$match": {"user_id": user_id}}, { "$group": { "_id": "$run_id", "email_count": {"$sum": 1}, "latest_created_at": {"$max": "$created_at"}, } }, {"$sort": {"latest_created_at": -1}}, ] results = list(self.collection.aggregate(pipeline)) runs = [ RunIdSummary( run_id=r["_id"], email_count=r["email_count"], latest_created_at=r.get("latest_created_at"), ) for r in results ] return RunIdListResponse(total_runs=len(runs), runs=runs) # ────────────────────────────────────────────── # READ — Emails by run_id (paginated) # ────────────────────────────────────────────── def get_emails_by_run_id( self, run_id: str, user_id: str, page: int = 1, limit: int = 20, approval_status: Optional[str] = None, ) -> EmailBatchListResponse: """Fetch paginated email documents for a given run_id.""" query: dict = {"run_id": run_id, "user_id": user_id} if approval_status: query["approval_status"] = approval_status total = self.collection.count_documents(query) skip = (page - 1) * limit docs = list( self.collection.find(query) .sort("created_at", -1) .skip(skip) .limit(limit) ) items = [EmailBatchResponse(**self._serialize_doc(d)) for d in docs] return EmailBatchListResponse( total=total, page=page, limit=limit, items=items, ) # ────────────────────────────────────────────── # READ — Leads from PostgreSQL (paginated) # ────────────────────────────────────────────── def get_leads_by_run_id( self, run_id: str, user_id: str, page: int = 1, limit: int = 20, ) -> LeadListResponse: """Fetch paginated enriched leads from PostgreSQL for a given run_id.""" count_query = sa.text("SELECT COUNT(*) FROM leads_insights WHERE run_id = :run_id AND user_id = :user_id") items_query = sa.text(""" SELECT * FROM leads_insights WHERE run_id = :run_id AND user_id = :user_id ORDER BY created_at DESC LIMIT :limit OFFSET :offset """) offset = (page - 1) * limit with engine.connect() as conn: total = conn.execute(count_query, {"run_id": run_id, "user_id": user_id}).scalar() or 0 result = conn.execute(items_query, {"run_id": run_id, "user_id": user_id, "limit": limit, "offset": offset}) rows = result.mappings().all() items = [LeadResponse(**dict(row)) for row in rows] return LeadListResponse( total=total, page=page, limit=limit, items=items, ) def get_lead_stats_by_run(self, run_id: str, user_id: str): """Fetch counts of lead statuses for a specific run.""" query = sa.text(""" SELECT COUNT(*) as total_leads, COUNT(*) FILTER (WHERE is_email_generated = TRUE) as generated, COUNT(*) FILTER (WHERE email_generation_status = 'processing') as processing, COUNT(*) FILTER (WHERE email_generation_status = 'failed') as failed FROM leads_insights WHERE run_id = :run_id AND user_id = :user_id """) with engine.connect() as conn: row = conn.execute(query, {"run_id": run_id, "user_id": user_id}).mappings().first() return { "run_id": run_id, "total": row["total_leads"], "generated": row["generated"], "processing": row["processing"], "failed": row["failed"], "pending": row["total_leads"] - (row["generated"] + row["processing"] + row["failed"]) } # ────────────────────────────────────────────── # READ — Single email by _id # ────────────────────────────────────────────── def get_email_by_id(self, email_id: str, user_id: str) -> Optional[EmailBatchResponse]: """Fetch a single email document by its MongoDB _id.""" try: doc = self.collection.find_one({"_id": ObjectId(email_id), "user_id": user_id}) except Exception: return None if not doc: return None return EmailBatchResponse(**self._serialize_doc(doc)) # ────────────────────────────────────────────── # PATCH — Partial update # ────────────────────────────────────────────── def patch_email(self, email_id: str, patch: EmailPatchRequest, user_id: str) -> Optional[EmailBatchResponse]: """Partially update an email document. Only non-None fields in the patch are applied.""" update_fields: dict = {} if patch.approval_status is not None: update_fields["approval_status"] = patch.approval_status if patch.generated_emails is not None: update_fields["generated_emails"] = patch.generated_emails if patch.email_status is not None: # Note: This replaces the entire email_status object. # Could be upgraded to nested updates if needed. update_fields["email_status"] = patch.email_status if patch.metadata is not None: update_fields["metadata"] = patch.metadata if not update_fields: # Nothing to update — just return the existing doc return self.get_email_by_id(email_id, user_id) update_fields["updated_at"] = datetime.now() try: result = self.collection.find_one_and_update( {"_id": ObjectId(email_id), "user_id": user_id}, {"$set": update_fields}, return_document=True, # return the updated doc ) except Exception: return None if not result: return None return EmailBatchResponse(**self._serialize_doc(result)) # ────────────────────────────────────────────── # STATUS — Approve / Decline # ────────────────────────────────────────────── def update_email_status(self, email_id: str, status: str, user_id: str) -> Optional[EmailBatchResponse]: """Update the approval status of a single email.""" return self.patch_email(email_id, EmailPatchRequest(approval_status=status), user_id) def update_run_status(self, run_id: str, status: str, user_id: str) -> BatchActionResponse: """Update the approval status of all emails in a run.""" result = self.collection.update_many( {"run_id": run_id, "user_id": user_id}, {"$set": {"approval_status": status, "updated_at": datetime.now()}} ) return BatchActionResponse( modified_count=result.modified_count, message=f"Successfully set status to '{status}' for {result.modified_count} emails in run {run_id}" ) # ────────────────────────────────────────────── # DELETE — Single email # ────────────────────────────────────────────── def delete_email(self, email_id: str, user_id: str) -> DeleteResponse: """Delete a single email document by _id.""" try: result = self.collection.delete_one({"_id": ObjectId(email_id), "user_id": user_id}) except Exception: return DeleteResponse(deleted_count=0, message="Invalid email ID format") if result.deleted_count == 0: return DeleteResponse(deleted_count=0, message="Email not found") return DeleteResponse(deleted_count=1, message="Email deleted successfully") # ────────────────────────────────────────────── # DELETE — All emails for a run_id # ────────────────────────────────────────────── def delete_emails_by_run_id(self, run_id: str, user_id: str) -> DeleteResponse: """Delete all email documents for a given run_id.""" result = self.collection.delete_many({"run_id": run_id, "user_id": user_id}) if result.deleted_count == 0: return DeleteResponse(deleted_count=0, message=f"No emails found for run_id: {run_id}") return DeleteResponse( deleted_count=result.deleted_count, message=f"Deleted {result.deleted_count} emails for run_id: {run_id}", ) # ────────────────────────────────────────────── # REGENERATE — Re-create emails for a run_id # ────────────────────────────────────────────── def regenerate_emails_for_run(self, run_id: str, user_id: str) -> RegenerateResponse: """ Resets email generation status for a run: 1. Resets 'is_email_generated' to FALSE in PostgreSQL. 2. Deletes existing emails in MongoDB for this run_id. """ # Step 1: Update PostgreSQL to mark emails as not generated # We target rows that were previously successful and generated update_query = sa.text(""" UPDATE leads_insights SET is_email_generated = FALSE WHERE run_id = :run_id AND user_id = :user_id AND email_generation_status = 'approved' AND is_email_generated = TRUE AND valid_email_exists = TRUE """) with engine.begin() as conn: # Using begin() for automatic transaction commit result = conn.execute(update_query, {"run_id": run_id, "user_id": user_id}) count_updated = result.rowcount if count_updated == 0: return RegenerateResponse( run_id=run_id, total_leads=0, succeeded=0, failed=0, skipped=0, ) self.collection.delete_many({"run_id": run_id, "user_id": user_id}) return RegenerateResponse( run_id=run_id, total_leads=count_updated, succeeded=count_updated, failed=0, skipped=0, ) # ────────────────────────────────────────────── # WOODPECKER SYNC # ────────────────────────────────────────────── async def sync_approved_emails_to_woodpecker(self, run_id: str, user_id: str) -> dict: """ Sync all 'approved' emails for a run that haven't been synced yet to Woodpecker. """ # 1. Get the run to find the Woodpecker Campaign ID run = self.sync_collection.find_one({"run_id": run_id, "user_id": user_id}) if not run: return {"success": False, "message": "Run not found"} wp_campaign_id = run.get("woodpecker_campaign_id") if not wp_campaign_id: return {"success": False, "message": "No Woodpecker campaign linked to this run"} pipeline = [ { "$match": { "run_id": run_id, "user_id": user_id, "approval_status": "approved", "woodpecker_synced": {"$ne": True} } }, { "$group": { "_id": "$lead_email", "doc": {"$first": "$$ROOT"} } }, { "$replaceRoot": {"newRoot": "$doc"} } ] emails = list(self.collection.aggregate(pipeline)) if not emails: return {"success": False, "message": "No new approved emails to sync"} # 3. Format prospects for Woodpecker prospects = [] for doc in emails: gen_emails = doc.get("generated_emails", {}) # Extract first name from title or use a default title = doc.get("title", "") first_name = title.split()[0] if title else "there" p = ProspectV1( email=doc["lead_email"], first_name=first_name, snippet1=gen_emails.get("main_email_html", ""), snippet2=gen_emails.get("main_email_subject", ""), custom_id=str(doc["_id"]) ) prospects.append(p) # 4. Send to Woodpecker request = AddProspectsV1Request( campaign=CampaignIdWrapper(campaign_id=int(wp_campaign_id)), prospects=prospects ) try: result = await self.woodpecker_service.add_prospects_to_campaign_v1(request) # 5. Mark as synced in MongoDB email_ids = [doc["_id"] for doc in emails] self.collection.update_many( {"_id": {"$in": email_ids}}, {"$set": { "woodpecker_synced": True, "woodpecker_synced_at": datetime.now(), "updated_at": datetime.now() }} ) return { "success": True, "synced_count": len(prospects), "woodpecker_response": result } except Exception as e: logger.error(f"Failed to sync to Woodpecker: {e}") return {"success": False, "error": str(e)} def get_sync_status(self, run_id: str, user_id: str): pipeline = [ { "$match": { "run_id": run_id, "user_id": user_id } }, { "$project": { "campaign_name": 1, "status": 1, "job_ids_count": { "$size": { "$ifNull": ["$job_ids", []] } }, "synced_job_ids_count": { "$size": { "$ifNull": ["$synced_job_ids", []] } }, "unsynced_ids": 1 } }, { "$addFields": { "remaining_count": { "$size": { "$ifNull": ["$unsynced_ids", []] } }, "is_fully_synced": { "$eq": ["$job_ids_count", "$synced_job_ids_count"] }, "_id": { "$toString": "$_id" } } } ] try: # Convert cursor to list to check for empty results result = list(self.sync_collection.aggregate(pipeline)) if not result: return {"success": False, "message": "Run ID not found"} return { "success": True, "data": result[0] # Return the first (and likely only) result } except Exception as e: return {"success": False, "error": str(e)} # ────────────────────────────────────────────── # PROMPTS — CRUD # ────────────────────────────────────────────── def list_prompts(self, user_id: str) -> PromptListResponse: """List all custom prompts for a user.""" docs = list(self.prompt_collection.find({"user_id": user_id})) items = [PromptResponse(**self._serialize_doc(d)) for d in docs] return PromptListResponse(total=len(items), items=items) def upsert_prompt(self, user_id: str, prompt: PromptRequest) -> PromptResponse: """Create or update a prompt document.""" query = {"user_id": user_id, "run_id": prompt.run_id} if prompt.run_id is None: query["is_universal"] = True doc = { "user_id": user_id, "run_id": prompt.run_id, "system_instruction": prompt.system_instruction, "user_instruction": prompt.user_instruction, "rag_context": prompt.rag_context, "is_universal": prompt.is_universal, "updated_at": datetime.now() } result = self.prompt_collection.find_one_and_update( query, {"$set": doc, "$setOnInsert": {"created_at": datetime.now()}}, upsert=True, return_document=True ) return PromptResponse(**self._serialize_doc(result)) def delete_prompt(self, prompt_id: str, user_id: str) -> DeleteResponse: """Delete a custom prompt by its _id.""" try: result = self.prompt_collection.delete_one({"_id": ObjectId(prompt_id), "user_id": user_id}) except Exception: return DeleteResponse(deleted_count=0, message="Invalid prompt ID format") if result.deleted_count == 0: return DeleteResponse(deleted_count=0, message="Prompt not found") return DeleteResponse(deleted_count=1, message="Prompt deleted successfully") def get_resolved_prompt(self, user_id: str, run_id: str) -> ResolvedPromptResponse: """Expose the internal generator's resolution logic to the API.""" resolved = self.email_generator._get_resolved_prompts(user_id, run_id) return ResolvedPromptResponse( system_instruction=resolved["system"], user_instruction=resolved["user"], rag_context=resolved["rag"] ) # ────────────────────────────────────────────── # Lightweight lead adapter # ────────────────────────────────────────────── class _LeadProxy: """Thin adapter that exposes DB row fields as attributes expected by EmailBatchGenerator.""" def __init__(self, data: dict): self._data = data def __getattr__(self, name): if name.startswith("_"): raise AttributeError(name) return self._data.get(name) def _row_to_lead_obj(row: dict, run_id: str) -> _LeadProxy: """Convert a PostgreSQL row dict into a lead-like object for the email generator.""" row["run_id"] = run_id return _LeadProxy(row)