"""Gmail account management endpoints.""" import secrets from datetime import datetime, timezone from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile from pydantic import BaseModel from sqlalchemy import 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 GmailAccount, OAuthClientConfig, OAuthToken, User from app.services import gmail_service router = APIRouter(prefix="/gmail", tags=["gmail-accounts"]) class AccountUpdate(BaseModel): sending_method: str = "api" # api | smtp daily_limit: int = 200 campaign_limit: int = 200 def _account_out(account: GmailAccount) -> dict: t = account.oauth_token return { "id": account.id, "email": account.email, "display_name": account.display_name, "status": account.status, "sending_method": account.sending_method, "daily_limit": account.daily_limit, "campaign_limit": account.campaign_limit, "sends_today": account.sends_today, "last_success_at": account.last_success_at.isoformat() if account.last_success_at else None, "error_count": account.error_count, "last_error": (account.last_error or "")[:300], "cooldown_until": account.cooldown_until.isoformat() if account.cooldown_until else None, "has_refresh_token": bool(t and t.refresh_token_enc), "token_expires_at": t.expires_at.isoformat() if t and t.expires_at else None, "authorized": account.status == "authorized", } @router.get("/accounts") async def list_accounts(user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): rows = (await db.execute(select(GmailAccount).where( GmailAccount.user_id == user.id).order_by(GmailAccount.id))).scalars().all() return {"accounts": [_account_out(a) for a in rows], "count": len(rows)} @router.post("/accounts/connect") async def connect_start(user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): """Return the Google authorization URL for this user to open (real OAuth start).""" try: gmail_service.resolve_client_config() except gmail_service.GmailOauthError: raise HTTPException(400, "No Google OAuth client configuration imported yet. " "Use 'Import Gmail Configuration' first.") from app.api.routers.auth import google_start_impl return await google_start_impl(user) @router.post("/accounts/{account_id}/test") async def test_account(account_id: int, user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): """Real connection test: verify OAuth creds and Gmail API profile.""" account = await db.get(GmailAccount, account_id) if not account or account.user_id != user.id: raise HTTPException(404, "Account not found") token = account.oauth_token if not token: return {"ok": False, "detail": "No OAuth token stored for this account"} from app.db.session import SyncSession with SyncSession() as s: acc_sync = s.get(GmailAccount, account.id) tok_sync = s.get(OAuthToken, token.id) try: client = gmail_service.build_gmail_client_from_account(acc_sync, tok_sync) result = client.verify_connection() except Exception as e: result = {"ok": False, "detail": str(e)[:400]} if result.get("ok"): account.status = "authorized" await db.commit() else: account.status = "expired" if "auth" in (result.get("detail") or "").lower() else account.status account.last_error = (result.get("detail") or "")[:500] await db.commit() await log_audit(db, user.id, "gmail.test", "gmail_account", account.id, {"ok": result.get("ok")}, None) return result @router.post("/accounts/{account_id}/reauthorize") async def reauthorize(account_id: int, user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): account = await db.get(GmailAccount, account_id) if not account or account.user_id != user.id: raise HTTPException(404, "Account not found") try: client = gmail_service.resolve_client_config_for_account(account) except gmail_service.GmailOauthError as e: raise HTTPException(400, str(e)) state = gmail_service.generate_state() import json as _json import redis as _redis from app.core.config import settings try: r = _redis.from_url(settings.REDIS_URL, socket_timeout=2) r.setex(f"oauthstate:{state}", 600, _json.dumps({"user_id": user.id})) except Exception: raise HTTPException(503, "State store unavailable") url = gmail_service.build_authorization_url(client, state) return {"authorization_url": url} @router.post("/accounts/{account_id}/revoke") =====RECIPIENTS IMPORT===== "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), 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), 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), 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(list_id: int = Form(...), payload: ManualRecipient = None, user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): 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() 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("") 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), 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), 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()