import base64
import html
import os
from datetime import datetime, timezone
from typing import Any, Dict, Optional
import httpx
from fastapi import Depends, FastAPI, Header, HTTPException
GOTENBERG_URL = os.getenv("ONEOK_GOTENBERG_URL", "http://oneok-gotenberg:3000").rstrip("/")
API_KEY = os.getenv("ONEOK_ADAPTER_API_KEY", "").strip()
app = FastAPI(title="1OK Docs Adapter", version="1.0.0")
def _auth(authorization: Optional[str] = Header(default=None)) -> None:
if not API_KEY:
return
expected = f"Bearer {API_KEY}"
if authorization != expected:
raise HTTPException(status_code=401, detail="Unauthorized")
def _render_html(title: str, data: Dict[str, Any]) -> str:
ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
rows = []
for k, v in data.items():
rows.append(
f"| {html.escape(str(k))} | "
f"{html.escape(str(v))} |
"
)
table = "\n".join(rows) if rows else "| No data |
"
return f"""
{html.escape(title)}
Generated: {html.escape(ts)}
"""
async def _render_pdf(file_name: str, html_text: str) -> Dict[str, Any]:
endpoints = [
f"{GOTENBERG_URL}/forms/chromium/convert/html",
f"{GOTENBERG_URL}/forms/libreoffice/convert",
]
async with httpx.AsyncClient(timeout=60.0) as client:
files = {file_name: (file_name, html_text.encode("utf-8"), "text/html")}
last_error = None
pdf = b""
for url in endpoints:
resp = await client.post(url, files=files)
if resp.status_code < 400:
pdf = resp.content
break
last_error = f"{url} -> {resp.status_code} {resp.text[:200]}"
if not pdf:
raise HTTPException(status_code=502, detail=f"Gotenberg error: {last_error}")
return {
"file_name": file_name.replace(".html", ".pdf"),
"mime": "application/pdf",
"size_bytes": len(pdf),
"pdf_base64": base64.b64encode(pdf).decode("utf-8"),
}
@app.get("/health")
def health() -> Dict[str, Any]:
return {"status": "ok", "service": "oneok-docs-adapter"}
@app.post("/docs/render_quote_pdf")
async def render_quote_pdf(body: Dict[str, Any], _: None = Depends(_auth)) -> Dict[str, Any]:
quote_id = body.get("quote_id")
payload = body.get("quote_payload") or {}
if not quote_id and not payload:
raise HTTPException(status_code=400, detail="quote_id or quote_payload required")
if quote_id and "quote_id" not in payload:
payload = dict(payload)
payload["quote_id"] = quote_id
html_doc = _render_html("Комерційна пропозиція (1OK)", payload)
return await _render_pdf("quote.html", html_doc)
@app.post("/docs/render_invoice_pdf")
async def render_invoice_pdf(body: Dict[str, Any], _: None = Depends(_auth)) -> Dict[str, Any]:
payload = body.get("invoice_payload")
if not isinstance(payload, dict):
raise HTTPException(status_code=400, detail="invoice_payload required")
html_doc = _render_html("Рахунок (1OK)", payload)
return await _render_pdf("invoice.html", html_doc)