app/services/callback.py

import asyncio
import logging

import httpx

log = logging.getLogger(__name__)


def _auth_headers(task_auth_token: str | None) -> dict[str, str]:
    if task_auth_token is None:
        return {}
    return {"Authorization": task_auth_token}


async def notify_finish(
    finish_url: str,
    task_auth_token: str | None,
    doc_id: str,
    success: bool,
    message: str | None = None,
    project_id: str | None = None,
) -> None:
    """
    POST to backend finish URL with status and optional Authorization header.
    Best-effort; logs and optionally retries once on 5xx. Does not raise.
    """
    if not finish_url:
        return
    body = {"status": "complete" if success else "failure", "doc_id": doc_id}
    if project_id is not None:
        body["project_id"] = project_id
    if not success and message:
        body["message"] = message
    headers = _auth_headers(task_auth_token)
    log.info("Symbios finish request url=%s body=%s", finish_url, body)
    async with httpx.AsyncClient(timeout=30.0) as client:
        last_err = None
        for attempt in range(2):
            try:
                resp = await client.post(finish_url, json=body, headers=headers)
                log.info("Symbios finish response url=%s status_code=%s body=%s", finish_url, resp.status_code, resp.text[:500] if resp.text else "")
                if resp.status_code == 200:
                    log.info("Notify finish OK doc_id=%s status=%s", doc_id, body["status"])
                    return
                if 500 <= resp.status_code < 600:
                    log.warning(
                        "Symbios finish 5xx doc_id=%s attempt=%d status_code=%s response_body=%s",
                        doc_id,
                        attempt + 1,
                        resp.status_code,
                        resp.text,
                    )
                    if attempt == 0:
                        await asyncio.sleep(2)
                        continue
                    return
                log.warning(
                    "Notify finish unexpected response doc_id=%s status_code=%s body=%s",
                    doc_id,
                    resp.status_code,
                    resp.text[:200],
                )
                return
            except Exception as e:
                last_err = e
                log.warning("Notify finish request failed doc_id=%s error=%s", doc_id, e, exc_info=True)
                if attempt == 0:
                    await asyncio.sleep(2)
                    continue
        log.exception("Notify finish failed doc_id=%s last_err=%s", doc_id, last_err)


async def notify_update(
    update_url: str,
    task_auth_token: str | None,
    doc_id: str,
    finished: int,
    total: int,
    message: str,
    project_id: str | None = None,
) -> None:
    """
    POST to backend update URL with progress (doc_id, finished, total, message).
    Best-effort; does not raise. Used to report progress as each pipeline stage completes.
    """
    if not update_url:
        return
    body: dict[str, str | int] = {
        "doc_id": doc_id,
        "finished": finished,
        "total": total,
        "message": message,
    }
    if project_id is not None:
        body["project_id"] = project_id
    headers = _auth_headers(task_auth_token)
    log.info("Symbios update request url=%s token=%s body=%s", update_url, task_auth_token, body)
    async with httpx.AsyncClient(timeout=15.0) as client:
        try:
            resp = await client.post(update_url, json=body, headers=headers)
            log.info("Symbios update response url=%s status_code=%s body=%s", update_url, resp.status_code, resp.text[:500] if resp.text else "")
            if resp.status_code != 200:
                log.warning(
                    "Notify update doc_id=%s finished=%s/%s status_code=%s response_body=%s",
                    doc_id,
                    finished,
                    total,
                    resp.status_code,
                    resp.text,
                )
            if 500 <= resp.status_code < 600:
                log.warning(
                    "Symbios update 5xx doc_id=%s status_code=%s response_body=%s",
                    doc_id,
                    resp.status_code,
                    resp.text,
                )
        except Exception as e:
            log.warning("Notify update failed doc_id=%s: %s", doc_id, e, exc_info=True)


def notify_update_sync(
    update_url: str,
    task_auth_token: str | None,
    doc_id: str,
    finished: int,
    total: int,
    message: str,
    project_id: str | None = None,
) -> None:
    """
    Synchronous POST to backend update URL. Use from sync context (e.g. docling
    progress_callback). Best-effort; does not raise.
    """
    if not update_url:
        return
    body: dict[str, str | int] = {
        "doc_id": doc_id,
        "finished": finished,
        "total": total,
        "message": message,
    }
    if project_id is not None:
        body["project_id"] = project_id
    headers = _auth_headers(task_auth_token)
    log.info("Symbios update (sync) request url=%s body=%s", update_url, body)
    try:
        with httpx.Client(timeout=15.0) as client:
            resp = client.post(update_url, json=body, headers=headers)
            log.info("Symbios update (sync) response url=%s status_code=%s body=%s", update_url, resp.status_code, resp.text[:500] if resp.text else "")
            if resp.status_code != 200:
                log.warning(
                    "Notify update (sync) doc_id=%s finished=%s/%s status_code=%s response_body=%s",
                    doc_id,
                    finished,
                    total,
                    resp.status_code,
                    resp.text,
                )
            if 500 <= resp.status_code < 600:
                log.warning(
                    "Symbios update (sync) 5xx doc_id=%s status_code=%s response_body=%s",
                    doc_id,
                    resp.status_code,
                    resp.text,
                )
    except Exception as e:
        log.warning("Notify update (sync) failed doc_id=%s: %s", doc_id, e, exc_info=True)