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)