58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import asdict
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.models import ApplicationRecord, ExternalCreateResult
|
|
from app.security import mask_secret_url
|
|
|
|
|
|
class ExternalTicketClient:
|
|
def __init__(self, create_url_with_key: str, timeout: float = 10.0) -> None:
|
|
self._url = create_url_with_key.strip()
|
|
self._timeout = timeout
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return bool(self._url)
|
|
|
|
@property
|
|
def safe_url(self) -> str:
|
|
return mask_secret_url(self._url)
|
|
|
|
async def create_ticket(self, application: ApplicationRecord) -> ExternalCreateResult:
|
|
if not self.enabled:
|
|
return ExternalCreateResult(ok=False, error="external_url_not_configured")
|
|
|
|
body: dict[str, Any] = {
|
|
"application_id": application.id,
|
|
"category": application.category,
|
|
"description": application.description,
|
|
"customer_name": application.customer_name,
|
|
"customer_id": application.customer_id,
|
|
"chat_id": application.chat_id,
|
|
"attachments_json": application.attachments_json,
|
|
"source": application.source,
|
|
"created_at": application.created_at,
|
|
}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self._timeout) as client:
|
|
response = await client.post(self._url, json=body)
|
|
if 200 <= response.status_code < 300:
|
|
payload = response.json() if response.content else {}
|
|
external_id = str(
|
|
payload.get("id")
|
|
or payload.get("ticket_id")
|
|
or payload.get("result", {}).get("id")
|
|
or ""
|
|
).strip() or None
|
|
return ExternalCreateResult(ok=True, external_id=external_id, raw_response=payload)
|
|
return ExternalCreateResult(
|
|
ok=False,
|
|
error=f"http_{response.status_code}: {response.text[:600]}",
|
|
)
|
|
except Exception as exc:
|
|
return ExternalCreateResult(ok=False, error=str(exc))
|
|
|