"""Marketing module: email templates, single sends, campaigns, audit log.""" import re import asyncio from datetime import datetime, timezone from bson import ObjectId from bson.errors import InvalidId from fastapi import APIRouter, Depends, HTTPException from deps import get_db, require_roles from emailer import send_email from models import EmailTemplateIn, SendEmailIn, CampaignIn, MARKETING_ROLES router = APIRouter(prefix="/marketing", tags=["marketing"]) _gate = require_roles(*MARKETING_ROLES) def _oid(s): try: return ObjectId(s) except InvalidId: raise HTTPException(status_code=400, detail="ID inválido") def _ser_tpl(t: dict) -> dict: return { "id": str(t["_id"]), "name": t["name"], "subject": t["subject"], "body_html": t["body_html"], "description": t.get("description", ""), "created_at": t.get("created_at"), } def _ser_log(l: dict) -> dict: return { "id": str(l["_id"]), "to": l["to"], "subject": l["subject"], "status": l.get("status"), "error": l.get("error"), "sender_id": l.get("sender_id"), "sender_name": l.get("sender_name"), "campaign_id": l.get("campaign_id"), "crm_client_id": l.get("crm_client_id"), "created_at": l.get("created_at"), } # ---------- Templates ---------- @router.get("/templates") async def list_templates(_: dict = Depends(_gate)): db = get_db() cur = db.email_templates.find({}).sort("created_at", -1) return [_ser_tpl(t) async for t in cur] @router.post("/templates") async def create_template(payload: EmailTemplateIn, _: dict = Depends(_gate)): db = get_db() doc = {**payload.model_dump(), "created_at": datetime.now(timezone.utc).isoformat()} r = await db.email_templates.insert_one(doc) doc["_id"] = r.inserted_id return _ser_tpl(doc) @router.patch("/templates/{tid}") async def update_template(tid: str, payload: EmailTemplateIn, _: dict = Depends(_gate)): db = get_db() r = await db.email_templates.find_one_and_update({"_id": _oid(tid)}, {"$set": payload.model_dump()}, return_document=True) if not r: raise HTTPException(status_code=404, detail="Plantilla no encontrada") return _ser_tpl(r) @router.delete("/templates/{tid}") async def delete_template(tid: str, _: dict = Depends(_gate)): db = get_db() r = await db.email_templates.delete_one({"_id": _oid(tid)}) if r.deleted_count == 0: raise HTTPException(status_code=404, detail="Plantilla no encontrada") return {"ok": True} # ---------- Variable substitution ---------- def _render(text: str, ctx: dict) -> str: def sub(m): key = m.group(1).strip() return str(ctx.get(key, m.group(0))) return re.sub(r"\{\{\s*([\w_]+)\s*\}\}", sub, text or "") async def _log(db, sender, to, subject, status, error=None, campaign_id=None, crm_client_id=None): doc = { "to": to, "subject": subject, "status": status, "error": error, "sender_id": sender["id"], "sender_name": sender.get("name", ""), "campaign_id": campaign_id, "crm_client_id": crm_client_id, "created_at": datetime.now(timezone.utc).isoformat(), } await db.email_log.insert_one(doc) # ---------- Send single ---------- @router.post("/send") async def send_one(payload: SendEmailIn, user: dict = Depends(_gate)): db = get_db() ctx = {} if payload.crm_client_id: c = await db.crm_clients.find_one({"_id": _oid(payload.crm_client_id)}) if c: ctx = {"name": c["name"], "company": c.get("company", ""), "email": c.get("email", "")} subject = _render(payload.subject, ctx) body = _render(payload.body_html, ctx) result = await send_email(str(payload.to), subject, body, reply_to=None) status = "sent" if result.get("ok") else "failed" await _log(db, user, str(payload.to), subject, status, result.get("error"), None, payload.crm_client_id) if payload.crm_client_id: await db.crm_clients.update_one( {"_id": _oid(payload.crm_client_id)}, {"$set": {"last_contact_at": datetime.now(timezone.utc).isoformat()}}, ) return {"ok": result.get("ok", False), "error": result.get("error")} # ---------- Campaign (bulk) ---------- @router.post("/campaigns") async def run_campaign(payload: CampaignIn, user: dict = Depends(_gate)): db = get_db() audience: list[dict] = [] # CRM filter crm_filter: dict = {} if payload.audience_status: crm_filter["status"] = payload.audience_status if payload.audience_tags: crm_filter["tags"] = {"$in": payload.audience_tags} if crm_filter: async for c in db.crm_clients.find(crm_filter): if c.get("email"): audience.append({"id": str(c["_id"]), "email": c["email"], "name": c["name"], "company": c.get("company", "")}) # ad-hoc emails for em in (payload.audience_emails or []): audience.append({"id": None, "email": str(em), "name": "", "company": ""}) if not audience: raise HTTPException(status_code=400, detail="Audiencia vacía") # save campaign record camp_doc = { "name": payload.name, "subject": payload.subject, "body_html": payload.body_html, "audience_count": len(audience), "sent_count": 0, "failed_count": 0, "created_by": user["id"], "created_at": datetime.now(timezone.utc).isoformat(), } camp = await db.campaigns.insert_one(camp_doc) campaign_id = str(camp.inserted_id) sent = 0 failed = 0 async def _one(target): nonlocal sent, failed ctx = {"name": target.get("name") or "", "company": target.get("company") or "", "email": target["email"]} subj = _render(payload.subject, ctx) body = _render(payload.body_html, ctx) r = await send_email(target["email"], subj, body) ok = r.get("ok") if ok: sent += 1 else: failed += 1 await _log(db, user, target["email"], subj, "sent" if ok else "failed", r.get("error"), campaign_id, target.get("id")) # send with limited concurrency sem = asyncio.Semaphore(5) async def bounded(t): async with sem: await _one(t) await asyncio.gather(*[bounded(t) for t in audience]) await db.campaigns.update_one({"_id": camp.inserted_id}, {"$set": {"sent_count": sent, "failed_count": failed}}) return {"id": campaign_id, "audience_count": len(audience), "sent_count": sent, "failed_count": failed} @router.get("/campaigns") async def list_campaigns(_: dict = Depends(_gate)): db = get_db() cur = db.campaigns.find({}).sort("created_at", -1).limit(50) items = [] async for c in cur: items.append({ "id": str(c["_id"]), "name": c["name"], "subject": c["subject"], "audience_count": c.get("audience_count", 0), "sent_count": c.get("sent_count", 0), "failed_count": c.get("failed_count", 0), "created_at": c.get("created_at"), }) return items @router.get("/log") async def list_log(_: dict = Depends(_gate)): db = get_db() cur = db.email_log.find({}).sort("created_at", -1).limit(200) return [_ser_log(l) async for l in cur]