"""Suppression list manager: manual add, CSV/TXT import, search, delete, Clear All, bulk check. Suppressed addresses are skipped at import AND at send time.""" import csv import io from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile from pydantic import BaseModel, EmailStr from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_current_user, get_db, require_csrf from app.core.security import log_audit from app.models import SuppressionEntry, User from app.services.recipient_service import is_valid_email router = APIRouter(prefix="/suppression", tags=["suppression"]) REASON_CHOICES = ("unsubscribed", "bounced", "complaint", "manual", "other") def _norm_reason(r: str) -> str: r = (r or "").strip().lower() if r == "complained": r = "complaint" return r if r in REASON_CHOICES else "manual" class SuppressionIn(BaseModel): email: EmailStr reason: str = "manual" source: str = "" class BulkEmails(BaseModel): emails: list[EmailStr] reason: str = "manual" class CheckEmails(BaseModel): emails: list[EmailStr] def _serialize(s: SuppressionEntry) -> dict: return { "id": s.id, "email": s.email, "reason": s.reason, "source": s.source, "campaign_id": s.campaign_id, "created_at": s.created_at.isoformat() if s.created_at else None, } @router.get("/list") async def list_suppression(search: str = Query(""), reason: str = Query(""), page: int = Query(1, ge=1), page_size: int = Query(100, ge=1, le=500), db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): q = select(SuppressionEntry).where(SuppressionEntry.user_id == user.id) if search: q = q.where(SuppressionEntry.email.ilike(f"%{search}%")) if reason: q = q.where(SuppressionEntry.reason == reason) total = (await db.execute( select(func.count()).select_from(q.subquery()))).scalar() or 0 rows = (await db.execute( q.order_by(SuppressionEntry.id.desc()).offset((page - 1) * page_size).limit(page_size) )).scalars().all() return {"entries": [_serialize(s) for s in rows], "total": total, "page": page, "page_size": page_size} @router.post("/add") async def add_suppression(payload: SuppressionIn, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): email = payload.email.strip().lower() payload.reason = _norm_reason(payload.reason) existing = await db.execute( select(SuppressionEntry).where(SuppressionEntry.email == email, SuppressionEntry.user_id == user.id)) if existing.first(): raise HTTPException(status_code=409, detail="Already suppressed") s = SuppressionEntry(user_id=user.id, email=email, reason=payload.reason, source=payload.source.strip()[:200]) db.add(s) await db.commit() await log_audit(db, user.id, "suppression.add", "suppression", s.id, {"reason": s.reason}) return _serialize(s) @router.post("/import") async def import_suppression(file: UploadFile = None, reason: str = Query("manual"), db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): """Import suppressed emails from CSV/TXT (one email per line or email,reason).""" if file is None: raise HTTPException(status_code=400, detail="No file uploaded") raw = await file.read() if len(raw) > 2 * 1024 * 1024: raise HTTPException(status_code=413, detail="File exceeds 2MB") try: text = raw.decode("utf-8-sig") except UnicodeDecodeError: raise HTTPException(status_code=400, detail="File must be UTF-8 text") if text.lstrip().startswith(("{", "[")): raise HTTPException(status_code=400, detail="Only CSV or TXT files are accepted") entries, invalid = [], [] lines = text.splitlines() looks_csv = any("," in ln for ln in lines[:50]) if looks_csv: for row in csv.reader(io.StringIO(text)): if not row or not (row[0] or "").strip(): continue email = row[0].strip().lower() r = (row[1].strip() if len(row) > 1 else reason) or reason if email.lower().startswith(("email", "address")): continue if is_valid_email(email): entries.append((email, _norm_reason(r))) else: invalid.append(email) else: for line in lines: email = line.strip().lower() if not email or email.lower().startswith(("email", "address", "#")): continue if is_valid_email(email): entries.append((email, reason)) else: invalid.append(email) if not entries: raise HTTPException(status_code=400, detail=f"No valid emails found. Invalid: {invalid[:10]}") existing = set((await db.execute( select(SuppressionEntry.email).where(SuppressionEntry.user_id == user.id))).scalars()) added = 0 for email, r in entries: if email in existing: continue db.add(SuppressionEntry(user_id=user.id, email=email, reason=r, source="import")) existing.add(email) added += 1 await db.commit() await log_audit(db, user.id, "suppression.import", "suppression", None, {"added": added, "invalid": len(invalid)}) return {"added": added, "duplicates": len(entries) - added, "invalid": invalid[:20], "total_valid": len(entries)} @router.post("/bulk") async def bulk_add_suppression(payload: BulkEmails, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): payload.reason = _norm_reason(payload.reason) existing = set((await db.execute( select(SuppressionEntry.email).where(SuppressionEntry.user_id == user.id))).scalars()) added = 0 for e in payload.emails: email = e.strip().lower() if email in existing: continue db.add(SuppressionEntry(user_id=user.id, email=email, reason=payload.reason)) existing.add(email) added += 1 await db.commit() await log_audit(db, user.id, "suppression.bulk", "suppression", None, {"added": added}) return {"added": added, "duplicates": len(payload.emails) - added} @router.post("/check") async def check_suppression(payload: CheckEmails, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): """Return which of the given emails are currently suppressed.""" emails = [e.strip().lower() for e in payload.emails] rows = (await db.execute( select(SuppressionEntry.email).where(SuppressionEntry.user_id == user.id, SuppressionEntry.email.in_(emails)))).scalars().all() suppressed = set(rows) return {"suppressed": {e: (e in suppressed) for e in emails}} @router.post("/{sid}/delete") async def delete_suppression(sid: int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): s = await db.get(SuppressionEntry, sid) if not s or s.user_id != user.id: raise HTTPException(status_code=404, detail="Entry not found") email = s.email await db.delete(s) await db.commit() await log_audit(db, user.id, "suppression.delete", "suppression", sid, {"email": email}) return {"deleted": sid} @router.post("/clear-all") async def clear_all_suppression(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): rows = (await db.execute( select(SuppressionEntry).where(SuppressionEntry.user_id == user.id))).scalars().all() for s in rows: await db.delete(s) await db.commit() await log_audit(db, user.id, "suppression.clear_all", "suppression", None, {"deleted": len(rows)}) return {"deleted": len(rows)}