"""Outreach queue: tasks, the human approval flow, and reply tracking.""" from __future__ import annotations from datetime import date from fastapi import APIRouter, Depends, HTTPException, Query, Response, status from sqlalchemy import func, select from sqlalchemy.orm import Session from app.database import get_db from app.models import Campaign, Lead, OutreachTask, Reply from app.models.enums import TaskStatus from app.schemas.base import Message, Page from app.schemas.outreach import ( BulkTaskAction, GenerateTasksRequest, OutreachTaskCreate, OutreachTaskRead, OutreachTaskUpdate, ReplyCreate, ReplyRead, TaskActionRequest, TaskLeadSummary, ) from app.services import outreach_service from app.utils.csv_utils import write_csv router = APIRouter(prefix="/outreach", tags=["outreach"]) TASK_EXPORT_COLUMNS = [ ("taskId", "id"), ("taskType", "task_type"), ("status", "status"), ("scheduledDate", "scheduled_date"), ("leadName", "lead_name"), ("jobTitle", "job_title"), ("company", "company"), ("profileUrl", "profile_url"), ("score", "score"), ("message", "message"), ("reason", "reason"), ] def get_task_or_404(db: Session, task_id: int) -> OutreachTask: task = db.get(OutreachTask, task_id) if task is None: raise HTTPException(status_code=404, detail=f"Task {task_id} not found") return task def serialize(task: OutreachTask) -> OutreachTaskRead: payload = OutreachTaskRead.model_validate(task) payload.final_message = task.final_message payload.campaign_name = task.campaign.name if task.campaign else None if task.lead is not None: payload.lead = TaskLeadSummary.model_validate(task.lead) return payload @router.get("/tasks", response_model=Page[OutreachTaskRead]) def list_tasks( db: Session = Depends(get_db), campaign_id: int | None = Query(default=None, alias="campaignId"), lead_id: int | None = Query(default=None, alias="leadId"), status_filter: list[str] | None = Query(default=None, alias="status"), task_type: list[str] | None = Query(default=None, alias="taskType"), due: str | None = Query(default=None, pattern="^(today|overdue|upcoming)$"), search: str | None = None, sort: str = "scheduled", page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=500, alias="pageSize"), ) -> Page[OutreachTaskRead]: query = select(OutreachTask).join(Lead, OutreachTask.lead_id == Lead.id) if campaign_id is not None: query = query.where(OutreachTask.campaign_id == campaign_id) if lead_id is not None: query = query.where(OutreachTask.lead_id == lead_id) if status_filter: query = query.where(OutreachTask.status.in_(status_filter)) if task_type: query = query.where(OutreachTask.task_type.in_(task_type)) if due == "today": query = query.where(OutreachTask.scheduled_date == date.today()) elif due == "overdue": query = query.where(OutreachTask.scheduled_date < date.today()) elif due == "upcoming": query = query.where(OutreachTask.scheduled_date > date.today()) if search: needle = f"%{search.strip()}%" query = query.where(Lead.full_name.ilike(needle) | Lead.company.ilike(needle)) if sort == "score": query = query.order_by(Lead.score.desc(), OutreachTask.scheduled_date) elif sort == "newest": query = query.order_by(OutreachTask.id.desc()) else: query = query.order_by( OutreachTask.scheduled_date, OutreachTask.sequence_index, OutreachTask.id ) total = int(db.scalar(select(func.count()).select_from(query.subquery())) or 0) rows = db.scalars(query.offset((page - 1) * page_size).limit(page_size)).all() return Page[OutreachTaskRead]( items=[serialize(task) for task in rows], total=total, page=page, page_size=page_size ) @router.post("/tasks", response_model=OutreachTaskRead, status_code=status.HTTP_201_CREATED) def create_task(payload: OutreachTaskCreate, db: Session = Depends(get_db)) -> OutreachTaskRead: lead = db.get(Lead, payload.lead_id) if lead is None: raise HTTPException(status_code=404, detail=f"Lead {payload.lead_id} not found") campaign = db.get(Campaign, payload.campaign_id) if payload.campaign_id else lead.campaign task = outreach_service.create_task( db, lead, campaign, payload.task_type.value, payload.scheduled_date or date.today(), reason=payload.reason, generate=payload.generate_message, ) db.commit() db.refresh(task) return serialize(task) @router.post("/tasks/generate", response_model=dict) def generate_tasks(payload: GenerateTasksRequest, db: Session = Depends(get_db)) -> dict: campaign = db.get(Campaign, payload.campaign_id) if campaign is None: raise HTTPException(status_code=404, detail=f"Campaign {payload.campaign_id} not found") return outreach_service.generate_tasks_for_campaign(db, campaign, payload.only_qualified) @router.get("/tasks/export") def export_tasks( db: Session = Depends(get_db), campaign_id: int | None = Query(default=None, alias="campaignId"), status_filter: list[str] | None = Query(default=None, alias="status"), ) -> Response: query = select(OutreachTask).order_by(OutreachTask.scheduled_date, OutreachTask.id) if campaign_id is not None: query = query.where(OutreachTask.campaign_id == campaign_id) if status_filter: query = query.where(OutreachTask.status.in_(status_filter)) rows = [] for task in db.scalars(query).all(): rows.append( { "id": task.id, "task_type": task.task_type, "status": task.status, "scheduled_date": task.scheduled_date.isoformat(), "lead_name": task.lead.full_name if task.lead else "", "job_title": task.lead.job_title if task.lead else "", "company": task.lead.company if task.lead else "", "profile_url": task.lead.profile_url if task.lead else "", "score": task.lead.score if task.lead else 0, "message": task.final_message or "", "reason": task.reason or "", } ) return Response( content=write_csv(rows, TASK_EXPORT_COLUMNS), media_type="text/csv", headers={"Content-Disposition": 'attachment; filename="leadpilot-outreach-queue.csv"'}, ) @router.post("/tasks/bulk", response_model=dict) def bulk_task_action(payload: BulkTaskAction, db: Session = Depends(get_db)) -> dict: handlers = { "approve": outreach_service.approve_task, "complete": outreach_service.complete_task, "skip": outreach_service.skip_task, "reject": outreach_service.reject_task, } handler = handlers.get(payload.action) if handler is None: raise HTTPException(status_code=422, detail=f"Unknown action '{payload.action}'") updated = 0 for task_id in payload.task_ids: task = db.get(OutreachTask, task_id) if task is None: continue if payload.action == "approve": handler(db, task) else: handler(db, task, payload.notes) updated += 1 db.commit() return {"updated": updated, "action": payload.action} @router.get("/tasks/{task_id}", response_model=OutreachTaskRead) def get_task(task_id: int, db: Session = Depends(get_db)) -> OutreachTaskRead: return serialize(get_task_or_404(db, task_id)) @router.put("/tasks/{task_id}", response_model=OutreachTaskRead) def update_task( task_id: int, payload: OutreachTaskUpdate, db: Session = Depends(get_db) ) -> OutreachTaskRead: task = get_task_or_404(db, task_id) for key, value in payload.model_dump(exclude_unset=True).items(): if value is not None: setattr(task, key, value.value if hasattr(value, "value") else value) db.commit() db.refresh(task) return serialize(task) @router.delete("/tasks/{task_id}", response_model=Message) def delete_task(task_id: int, db: Session = Depends(get_db)) -> Message: task = get_task_or_404(db, task_id) db.delete(task) db.commit() return Message(detail=f"Deleted task {task_id}") @router.post("/{task_id}/approve", response_model=OutreachTaskRead) def approve_task( task_id: int, payload: TaskActionRequest | None = None, db: Session = Depends(get_db) ) -> OutreachTaskRead: task = get_task_or_404(db, task_id) if payload and payload.message: task.edited_message = payload.message outreach_service.approve_task(db, task) db.commit() db.refresh(task) return serialize(task) @router.post("/{task_id}/complete", response_model=OutreachTaskRead) def complete_task( task_id: int, payload: TaskActionRequest | None = None, db: Session = Depends(get_db) ) -> OutreachTaskRead: """Called after the user performed the action on LinkedIn themselves.""" task = get_task_or_404(db, task_id) outreach_service.complete_task(db, task, payload.notes if payload else None) db.commit() db.refresh(task) return serialize(task) @router.post("/{task_id}/skip", response_model=OutreachTaskRead) def skip_task( task_id: int, payload: TaskActionRequest | None = None, db: Session = Depends(get_db) ) -> OutreachTaskRead: task = get_task_or_404(db, task_id) outreach_service.skip_task(db, task, payload.notes if payload else None) db.commit() db.refresh(task) return serialize(task) @router.post("/{task_id}/reject", response_model=OutreachTaskRead) def reject_task( task_id: int, payload: TaskActionRequest | None = None, db: Session = Depends(get_db) ) -> OutreachTaskRead: task = get_task_or_404(db, task_id) outreach_service.reject_task(db, task, payload.notes if payload else None) db.commit() db.refresh(task) return serialize(task) @router.post("/{task_id}/fail", response_model=OutreachTaskRead) def fail_task( task_id: int, payload: TaskActionRequest | None = None, db: Session = Depends(get_db) ) -> OutreachTaskRead: task = get_task_or_404(db, task_id) outreach_service.fail_task(db, task, payload.notes if payload else None) db.commit() db.refresh(task) return serialize(task) # --------------------------------------------------------------------------- # Replies # --------------------------------------------------------------------------- replies_router = APIRouter(prefix="/replies", tags=["replies"]) @replies_router.get("", response_model=list[ReplyRead]) def list_replies( db: Session = Depends(get_db), campaign_id: int | None = Query(default=None, alias="campaignId"), lead_id: int | None = Query(default=None, alias="leadId"), limit: int = 100, ) -> list[ReplyRead]: query = select(Reply).order_by(Reply.id.desc()).limit(limit) if campaign_id is not None: query = query.where(Reply.campaign_id == campaign_id) if lead_id is not None: query = query.where(Reply.lead_id == lead_id) results = [] for reply in db.scalars(query).all(): payload = ReplyRead.model_validate(reply) payload.lead_name = reply.lead.full_name if reply.lead else None results.append(payload) return results @replies_router.post("", response_model=ReplyRead, status_code=status.HTTP_201_CREATED) def create_reply(payload: ReplyCreate, db: Session = Depends(get_db)) -> ReplyRead: """Record a reply. Any still-open follow-up for that lead is stopped.""" lead = db.get(Lead, payload.lead_id) if lead is None: raise HTTPException(status_code=404, detail=f"Lead {payload.lead_id} not found") reply = outreach_service.record_reply( db, lead, payload.reply_type.value, payload.content, payload.task_id ) db.commit() db.refresh(reply) result = ReplyRead.model_validate(reply) result.lead_name = lead.full_name return result @replies_router.delete("/{reply_id}", response_model=Message) def delete_reply(reply_id: int, db: Session = Depends(get_db)) -> Message: reply = db.get(Reply, reply_id) if reply is None: raise HTTPException(status_code=404, detail=f"Reply {reply_id} not found") db.delete(reply) db.commit() return Message(detail=f"Deleted reply {reply_id}")