@router.post("/accounts/{account_id}/revoke") async def disconnect_account(account_id: int, user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), request: Request = None): """Disconnect: revoke token at Google and delete stored credentials.""" 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 revoked_at_google = False if token: from app.services.gmail_service import revoke_token revoked_at_google = revoke_token(token.refresh_token or token.access_token) account.status = "revoked" account.is_active = False account.revoked_by_user = True account.revoked_at = datetime.now(timezone.utc) if token: await db.delete(token) await db.commit() await log_audit(db, user.id, "gmail.disconnect", "gmail_account", account.id, {"email": account.email, "revoked_at_google": revoked_at_google}, request.client.host if request else None) return {"ok": True, "revoked_at_google": revoked_at_google} @router.delete("/accounts/{account_id}") async def delete_account(account_id: int, user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), request: Request = None): return await disconnect_account(account_id, user, db, request) @router.post("/accounts/clear-all") async def clear_all_accounts(user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), request: Request = None): """Clear All: revoke & remove all locally saved Gmail connections (audit preserved).""" rows = (await db.execute(select(GmailAccount).where( GmailAccount.user_id == user.id))).scalars().all() revoked = 0 for account in rows: token = account.oauth_token if token: if gmail_service.revoke_token(token.refresh_token or token.access_token): revoked += 1 account.status = "revoked" account.is_active = False account.revoked_by_user = True account.revoked_at = datetime.now(timezone.utc) if token: await db.delete(token) await db.commit() await log_audit(db, user.id, "gmail.clear_all", "gmail_account", None, {"count": len(rows), "revoked_at_google": revoked}, request.client.host if request else None) return {"ok": True, "removed": len(rows), "revoked_at_google": revoked} @router.post("/accounts/import-config") async def import_client_config(file: UploadFile = File(...), user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), request: Request = None): """Import & validate a Google OAuth client_secret JSON (never echoed back).""" from app.core.config import settings content = await file.read() if len(content) > 1024 * 512: raise HTTPException(413, "File too large (max 512 KB)") if not file.filename.lower().endswith(".json"): raise HTTPException(400, "Only .json configuration files are accepted") try: parsed = gmail_service.parse_client_json(content) except gmail_service.GmailOauthError as e: raise HTTPException(400, str(e)) row = OAuthClientConfig( name=parsed["project_id"] or file.filename[:100], user_id=user.id, client_id=parsed["client_id"], client_secret=parsed["client_secret"], # encrypted via property setter project_id=parsed["project_id"], redirect_uri=f"{settings.BASE_URL}/auth/google/callback", ) db.add(row) await db.commit() await log_audit(db, user.id, "oauth_config.imported", "oauth_client_config", row.id, {"project": parsed["project_id"]}, request.client.host if request else None) return {"ok": True, "id": row.id, "project_id": parsed["project_id"], "client_id_preview": parsed["client_id"][:12] + "…", "redirect_uri": row.redirect_uri} @router.patch("/accounts/{account_id}") async def update_account(account_id: int, payload: AccountUpdate, 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") if payload.sending_method in ("api", "smtp"): account.sending_method = payload.sending_method account.daily_limit = max(1, payload.daily_limit) account.campaign_limit = max(1, payload.campaign_limit) await db.commit() return _account_out(account) =====PHONES===== """Company phone numbers manager: manual add, CSV/TXT import, activation toggle, usage counts, delete, Clear All. Feeds %CompanyPhone% variable in campaigns.""" import csv import io import re from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile from pydantic import BaseModel, field_validator 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 CompanyPhoneNumber, User router = APIRouter(prefix="/phones", tags=["phones"]) PHONE_RE = re.compile(r"^\+?[0-9][0-9 ()\-.]{4,30}$") class PhoneIn(BaseModel): phone: str label: str = "" @field_validator("phone") @classmethod def valid_phone(cls, v): v = v.strip() if not PHONE_RE.match(v): raise ValueError("Invalid phone number format") return v class PhonePatch(BaseModel): label: Optional[str] = None is_active: Optional[bool] = None def _serialize(p: CompanyPhoneNumber) -> dict: return { "id": p.id, "label": p.label, "phone": p.phone, "is_active": bool(p.is_active), "use_count": p.use_count, "created_at": p.created_at.isoformat() if p.created_at else None, } async def _get_own(db: AsyncSession, user: User, pid: int) -> CompanyPhoneNumber: p = await db.get(CompanyPhoneNumber, pid) if not p or p.user_id != user.id: raise HTTPException(status_code=404, detail="Phone number not found") return p @router.get("/list") async def list_phones(active_only: bool = Query(False), search: str = Query(""), page: int = Query(1, ge=1), page_size: int = Query(200, ge=1, le=500), db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): q = select(CompanyPhoneNumber).where(CompanyPhoneNumber.user_id == user.id) if active_only: q = q.where(CompanyPhoneNumber.is_active == True) # noqa: E712 if search: q = q.where(CompanyPhoneNumber.phone.ilike(f"%{search}%") | CompanyPhoneNumber.label.ilike(f"%{search}%")) total = (await db.execute( select(func.count()).select_from(q.subquery()))).scalar() or 0 rows = (await db.execute( q.order_by(CompanyPhoneNumber.id.desc()).offset((page - 1) * page_size).limit(page_size) )).scalars().all() return {"phones": [_serialize(p) for p in rows], "total": total, "page": page, "page_size": page_size} @router.post("/add") async def add_phone(payload: PhoneIn, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): """Add one phone number manually.""" dup = await db.execute( select(CompanyPhoneNumber).where(CompanyPhoneNumber.user_id == user.id, CompanyPhoneNumber.phone == payload.phone)) if dup.first(): raise HTTPException(status_code=409, detail="Phone number already exists") p = CompanyPhoneNumber(user_id=user.id, phone=payload.phone, label=payload.label.strip()) db.add(p) await db.commit() await log_audit(db, user.id, "phone.add", "company_phone", p.id, {"label": p.label}) return _serialize(p) @router.post("/import") async def import_phones(file: UploadFile = None, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): """Import phone numbers from CSV (phone[,label]) or TXT (one per line). Supports bulk import of 10+ numbers at once.""" 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") # reject JSON-style input explicitly (CSV/TXT only per spec) stripped = text.lstrip() if stripped.startswith("{") or stripped.startswith("["): raise HTTPException(status_code=400, detail="Only CSV or TXT files are accepted") entries: list[dict] = [] invalid: list[str] = [] if "," in "\n".join(text.splitlines()[:50]) or file.filename.lower().endswith(".csv"): reader = csv.reader(io.StringIO(text)) for row in reader: if not row or not any(c.strip() for c in row): continue phone = (row[0] or "").strip() label = (row[1] or "").strip() if len(row) > 1 else "" if not phone or phone.lower().startswith(("phone", "number", "#")): continue # header if PHONE_RE.match(phone): entries.append({"phone": phone, "label": label}) else: invalid.append(phone) else: for line in text.splitlines(): phone = line.strip() if not phone or phone.lower().startswith(("phone", "number", "#")): continue if PHONE_RE.match(phone): entries.append({"phone": phone, "label": ""}) else: invalid.append(phone) if not entries: raise HTTPException(status_code=400, detail=f"No valid phone numbers found. Invalid: {invalid[:10]}") existing = set((await db.execute( select(CompanyPhoneNumber.phone).where(CompanyPhoneNumber.user_id == user.id))).scalars()) added, duplicates = 0, 0 for e in entries: if e["phone"] in existing: duplicates += 1 continue db.add(CompanyPhoneNumber(user_id=user.id, phone=e["phone"], label=e["label"])) existing.add(e["phone"]) added += 1 await db.commit() await log_audit(db, user.id, "phone.import", "company_phone", None, {"added": added, "duplicates": duplicates, "invalid": len(invalid)}) return {"added": added, "duplicates": duplicates, "invalid": invalid[:20], "total_valid": len(entries)} @router.post("/{pid}/patch") async def patch_phone(pid: int, payload: PhonePatch, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): p = await _get_own(db, user, pid) if payload.label is not None: p.label = payload.label.strip()[:150] if payload.is_active is not None: p.is_active = bool(payload.is_active) await db.commit() await log_audit(db, user.id, "phone.patch", "company_phone", pid, {"is_active": p.is_active}) return _serialize(p) @router.post("/{pid}/delete") async def delete_phone(pid: int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): p = await _get_own(db, user, pid) phone_value = p.phone await db.delete(p) await db.commit() await log_audit(db, user.id, "phone.delete", "company_phone", pid, {"phone": phone_value}) return {"deleted": pid} @router.post("/clear-all") async def clear_all_phones(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): rows = (await db.execute( select(CompanyPhoneNumber).where(CompanyPhoneNumber.user_id == user.id))).scalars().all() for p in rows: await db.delete(p) await db.commit() await log_audit(db, user.id, "phone.clear_all", "company_phone", None, {"deleted": len(rows)}) return {"deleted": len(rows)} =====TESTSEND===== """Send Test Email: REAL send through a connected Gmail account with provider-confirmed results (message ID from Gmail), full variable substitution and optional attachments. This is the compliance-friendly way to verify end-to-end delivery before a campaign.""" import asyncio import base64 from datetime import datetime, timezone from typing import Optional from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, EmailStr, Field 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.config import settings as cfg from app.core.security import log_audit from app.models import EmailTemplate, GmailAccount, OAuthToken, User from app.services import message_builder from app.services.campaign_engine import get_setting from app.services.gmail_service import build_gmail_client_from_account from app.services.variable_engine import GeneratedValues, VariableEngine from app.db.session import SyncSession router = APIRouter(prefix="/test-send", tags=["test-send"]) class TestSendIn(BaseModel): account_id: int to_email: EmailStr subject: str = Field(min_length=1, max_length=500) template_id: Optional[int] = None body_html: str = "" body_text: str = "" include_pdf: bool = False use_random_variables: bool = True @router.post("/send") async def send_test(payload: TestSendIn, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): """REAL test send through Gmail API or SMTP XOAUTH2 using an authorized account. Returns the provider's message ID — never a fake success.""" import time t0 = time.monotonic() account = await db.get(GmailAccount, payload.account_id) if not account or account.user_id != user.id: raise HTTPException(status_code=404, detail="Gmail account not found") if account.status != "authorized" or not account.is_active: raise HTTPException(status_code=409, detail=f"Account is {account.status}; re-authorize first") if account.cooldown_until and account.cooldown_until > datetime.now(timezone.utc): raise HTTPException(status_code=409, detail="Account is in cooldown") # resolve body: template (with real variable substitution) or raw body template = None if payload.template_id: template = await db.get(EmailTemplate, payload.template_id) if not template or template.user_id != user.id: raise HTTPException(status_code=404, detail="Template not found") body_html = template.body_html if template else payload.body_html body_text = template.body_text if template else payload.body_text if not body_html and not body_text: raise HTTPException(status_code=400, detail="Provide a template or message body") problems = message_builder.validate_html(body_html) if body_html else {} if problems.get("errors"): raise HTTPException(status_code=400, detail=f"HTML rejected: {'; '.join(problems['errors'])}") # fake recipient record for variable substitution (test only) class _FakeRecipient: pass fr = _FakeRecipient() fr.email = str(payload.to_email) fr.first_name = "Test" fr.last_name = "Recipient" fr.company_name = await get_setting(db, "company_name", "Acme Corp") or "Acme Corp" fr.company_phone = "555-0100"