2:start/pause/resume/cancel, SSE live progress stream, per-recipient monitor, CSV export."""
69:def _ser(c: Campaign) -> dict:
102:async def _get_own(db: AsyncSession, user: User, cid: int) -> Campaign:
109:async def _validate_refs(db: AsyncSession, user: User, data: CampaignIn):
187:async def list_campaigns(status: str = Query(""),
206:async def create_campaign(payload: CampaignIn, db: AsyncSession = Depends(get_db),
222:async def get_campaign(cid: int, user: User = Depends(get_current_user),
228:async def patch_campaign(cid: int, payload: CampaignIn,
250:async def delete_campaign(cid: int, db: AsyncSession = Depends(get_db),
263:async def clear_all_campaigns(db: AsyncSession = Depends(get_db),
286:async def preview_campaign(cid: int, sample_size: int = Query(0),
375:async def start(cid: int, db: AsyncSession = Depends(get_db),
407:async def pause(cid: int, db: AsyncSession = Depends(get_db),
419:async def resume(cid: int, db: AsyncSession = Depends(get_db),
433:async def cancel(cid: int, db: AsyncSession = Depends(get_db),
445:# Monitor: progress, SSE stream, recipient table
448:async def _progress_payload(db: AsyncSession, c: Campaign) -> dict:
475:async def get_progress(cid: int, user: User = Depends(get_current_user),
481:@router.get("/{cid}/stream")
482:async def campaign_stream(cid: int, request: Request,
485:    """SSE live progress stream. Manual implementation (no sse-starlette): a Redis
493:    async def event_gen():
540:    return StreamingResponse(event_gen(), media_type="text/event-stream",
547:async def campaign_recipients(cid: int, status: str = Query(""),
564:    async def names():
597:async def export_campaign(cid: int, db: AsyncSession = Depends(get_db),
605:    def generate():
635:async def wizard_options(db: AsyncSession = Depends(get_db),
652:    async def valid_count(list_id: int) -> int:
679:async def validate_step(payload: dict, db: AsyncSession = Depends(get_db),