"""Recipients: manual add, CSV/TXT import with validation & preview, lists management.""" import csv import io import secrets from typing import List, Optional from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile from pydantic import BaseModel, EmailStr from sqlalchemy import delete as sa_delete, select, func from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_current_user, get_current_user_csrf, get_db from app.core.config import settings from app.core.security import log_audit from app.models import Recipient, RecipientList, SuppressionEntry, User from app.services import recipient_service router = APIRouter(prefix="/recipients", tags=["recipients"]) class ManualRecipient(BaseModel): email: EmailStr first_name: str = "" last_name: str = "" company_name: str = "" company_phone: str = "" customer_email: Optional[EmailStr] = None custom_fields: dict = {} class ManualIn(BaseModel): list_id: int recipient: ManualRecipient class ListCreate(BaseModel): name: str description: str = "" def _rec_out(r: Recipient) -> dict: return {"id": r.id, "email": r.email, "first_name": r.first_name, "last_name": r.last_name, "company_name": r.company_name, "company_phone": r.company_phone, "customer_email": r.customer_email, "custom_fields": r.custom_fields or {}, "status": r.status, "is_valid": r.is_valid, "validation_error": r.validation_error} async def _suppression_and_bounces(db: AsyncSession) -> tuple: sup = set((await db.execute(select(SuppressionEntry.email))).scalars().all()) return sup, set() @router.get("/lists") async def list_lists(user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): rows = (await db.execute(select(RecipientList).where( RecipientList.user_id == user.id).order_by(RecipientList.id.desc()))).scalars().all() return {"lists": [{"id": l.id, "name": l.name, "description": l.description, "source": l.source, "row_count": l.row_count, "created_at": l.created_at.isoformat()} for l in rows]} @router.post("/lists") async def create_list(payload: ListCreate, user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db)): rl = RecipientList(user_id=user.id, name=payload.name, description=payload.description) db.add(rl) await db.commit() return {"id": rl.id, "name": rl.name} @router.post("/import-preview") async def import_preview(file: UploadFile = File(...), user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db)): """Parse + validate an upload WITHOUT saving. Shows the full validation report.""" content = await file.read() if len(content) > settings.MAX_UPLOAD_MB * 1024 * 1024: raise HTTPException(413, f"File too large (max {settings.MAX_UPLOAD_MB} MB)") fname = (file.filename or "").lower() if fname.endswith(".json"): raise HTTPException(400, "JSON recipient files are not accepted. Use CSV or TXT.") if not (fname.endswith(".csv") or fname.endswith(".txt")): raise HTTPException(400, "Only .csv and .txt files are accepted for recipients") if fname.endswith(".txt"): lines = recipient_service.parse_txt(content.decode("utf-8", errors="replace")) records = [] for ln in lines: parts = [p.strip() for p in ln.split(",")] rec = {"email": parts[0]} if len(parts) > 1: rec["first_name"] = parts[1] if len(parts) > 2: rec["last_name"] = parts[2] records.append(rec) headers = ["email", "first_name", "last_name"] parse_errors = [] else: records, headers, parse_errors = recipient_service.parse_csv( content.decode("utf-8", errors="replace")) sup, bounced = await _suppression_and_bounces(db) existing = set((await db.execute(select(Recipient.email))).scalars().all()) rpt = recipient_service.validate_rows(records, sup, bounced, existing, settings.MAX_RECIPIENTS_PER_CAMPAIGN) return {"filename": file.filename, "headers": headers, "summary": rpt.summary(), "valid_sample": rpt.valid[:25], "invalid_sample": rpt.invalid[:25], "duplicates_sample": rpt.duplicates[:10], "suppressed_sample": rpt.suppressed[:10], "parse_errors": parse_errors[:50], "parsed_records": len(records)} @router.post("/import") async def import_commit(list_id: int = Form(...), file: UploadFile = File(...), user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db), request: Request = None): """Commit an import into the given list (same validation as preview).""" rl = await db.get(RecipientList, list_id) if not rl or rl.user_id != user.id: raise HTTPException(404, "Recipient list not found") content = await file.read() if len(content) > settings.MAX_UPLOAD_MB * 1024 * 1024: raise HTTPException(413, "File too large") fname = (file.filename or "").lower() if fname.endswith(".json"): raise HTTPException(400, "JSON recipient files are not accepted. Use CSV or TXT.") if not (fname.endswith(".csv") or fname.endswith(".txt")): raise HTTPException(400, "Only .csv and .txt files are accepted") if fname.endswith(".txt"): lines = recipient_service.parse_txt(content.decode("utf-8", errors="replace")) records = [] for ln in lines: parts = [p.strip() for p in ln.split(",")] rec = {"email": parts[0]} if len(parts) > 1: rec["first_name"] = parts[1] if len(parts) > 2: rec["last_name"] = parts[2] records.append(rec) else: records, _, _ = recipient_service.parse_csv(content.decode("utf-8", errors="replace")) sup, bounced = await _suppression_and_bounces(db) existing = set((await db.execute(select(Recipient.email).where( Recipient.list_id == list_id))).scalars().all()) rpt = recipient_service.validate_rows(records, sup, bounced, existing, settings.MAX_RECIPIENTS_PER_CAMPAIGN) for v in rpt.valid: db.add(Recipient(list_id=list_id, email=v["email"], first_name=v["first_name"], last_name=v["last_name"], company_name=v["company_name"], company_phone=v["company_phone"], customer_email=v["customer_email"] or v["email"], custom_fields=v.get("custom_fields", {}), unsub_token=recipient_service.make_unsub_token())) rl.row_count += len(rpt.valid) rl.source = "txt" if fname.endswith(".txt") else "csv" await db.commit() await log_audit(db, user.id, "recipients.import", "recipient_list", list_id, rpt.summary(), request.client.host if request else None) return {"ok": True, **rpt.summary(), "list_id": list_id} @router.post("/manual") async def add_manual(payload_in: ManualIn, user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db), request: Request = None): payload, list_id = payload_in.recipient, payload_in.list_id rl = await db.get(RecipientList, list_id) if not rl or rl.user_id != user.id: raise HTTPException(404, "Recipient list not found") sup, bounced = await _suppression_and_bounces(db) existing = set((await db.execute(select(Recipient.email).where( Recipient.list_id == list_id))).scalars().all()) rpt = recipient_service.validate_rows( [payload.model_dump(exclude={"custom_fields"}) | {"custom_fields": payload.custom_fields}], sup, bounced, existing) if rpt.valid: v = rpt.valid[0] rec = Recipient(list_id=list_id, email=v["email"], first_name=v["first_name"], last_name=v["last_name"], company_name=v["company_name"], company_phone=v["company_phone"], customer_email=v["customer_email"] or v["email"], custom_fields=payload.custom_fields, unsub_token=recipient_service.make_unsub_token()) db.add(rec) rl.row_count += 1 rl.source = "manual" await db.commit() await log_audit(db, user.id, "recipients.manual_add", "recipient", rec.id, {"email": v["email"]}, request.client.host if request else None) return {"ok": True, "recipient": _rec_out(rec)} return {"ok": False, "summary": rpt.summary(), "invalid": rpt.invalid[:3], "duplicates": rpt.duplicates[:3], "suppressed": rpt.suppressed[:3]} @router.get("/list") async def list_recipients(list_id: int = 0, q: str = "", page: int = 1, limit: int = 50, user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): limit = min(max(1, limit), 500) stmt = select(Recipient).join(RecipientList).where(RecipientList.user_id == user.id) if list_id: stmt = stmt.where(Recipient.list_id == list_id) if q: stmt = stmt.where(Recipient.email.ilike(f"%{q}%")) total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar() rows = (await db.execute(stmt.order_by(Recipient.id).offset((page - 1) * limit).limit(limit))).scalars().all() return {"total": total, "page": page, "recipients": [_rec_out(r) for r in rows]} @router.delete("/{recipient_id}") async def delete_recipient(recipient_id: int, user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db)): rec = await db.get(Recipient, recipient_id) if not rec: raise HTTPException(404, "Not found") rl = await db.get(RecipientList, rec.list_id) if not rl or rl.user_id != user.id: raise HTTPException(404, "Not found") await db.delete(rec) rl.row_count = max(0, rl.row_count - 1) await db.commit() return {"ok": True} @router.delete("/lists/{list_id}") async def delete_list(list_id: int, user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db)): rl = await db.get(RecipientList, list_id) if not rl or rl.user_id != user.id: raise HTTPException(404, "Not found") await db.execute(sa_delete(Recipient).where(Recipient.list_id == list_id)) await db.delete(rl) await db.commit() return {"ok": True} @router.post("/lists/{list_id}/clear") async def clear_list(list_id: int, user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db)): rl = await db.get(RecipientList, list_id) if not rl or rl.user_id != user.id: raise HTTPException(404, "Not found") await db.execute(sa_delete(Recipient).where(Recipient.list_id == list_id)) rl.row_count = 0 await db.commit() return {"ok": True, "list_id": list_id}