317 lines
13 KiB
Python
Executable File
317 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Probe Jasmin invoice print API variants without creating documents.
|
|
|
|
This script is read-only: it fetches an existing invoice and tries several
|
|
possible /print query-parameter variants to determine whether an A4 PDF can be
|
|
obtained through the API even when the default print endpoint returns a narrow
|
|
receipt-like page.
|
|
|
|
Use after ORC->FA conversion tests created a real invoice, e.g. FA.FA2026.137.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from sqlalchemy import text
|
|
|
|
from app.db import engine
|
|
from app.jasmin_client import JasminClient
|
|
|
|
OUT_PREFIX = "/tmp/clientflow_jasmin_invoice_print_api_probe"
|
|
POINT_TO_MM = 25.4 / 72.0
|
|
A4_PT = (595.0, 842.0)
|
|
|
|
|
|
def _s(value: Any) -> str:
|
|
return str(value or "").strip()
|
|
|
|
|
|
def _json_default(value: Any) -> str:
|
|
try:
|
|
return value.isoformat() # type: ignore[attr-defined]
|
|
except Exception:
|
|
return str(value)
|
|
|
|
|
|
def _looks_a4(w: float, h: float, tolerance_pt: float = 20.0) -> bool:
|
|
return any(
|
|
abs(a - A4_PT[0]) <= tolerance_pt and abs(b - A4_PT[1]) <= tolerance_pt
|
|
for a, b in ((w, h), (h, w))
|
|
)
|
|
|
|
|
|
def _extract_pdf_boxes(data: bytes) -> list[dict[str, Any]]:
|
|
latin = data.decode("latin-1", errors="ignore")
|
|
pattern = re.compile(r"/(MediaBox|CropBox)\s*\[\s*([-+0-9.]+)\s+([-+0-9.]+)\s+([-+0-9.]+)\s+([-+0-9.]+)\s*\]", re.I)
|
|
out: list[dict[str, Any]] = []
|
|
seen: set[tuple[str, float, float, float, float]] = set()
|
|
for m in pattern.finditer(latin):
|
|
kind = m.group(1)
|
|
x0, y0, x1, y1 = (float(m.group(i)) for i in range(2, 6))
|
|
key = (kind.lower(), x0, y0, x1, y1)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
w, h = abs(x1 - x0), abs(y1 - y0)
|
|
out.append({
|
|
"kind": kind,
|
|
"width_pt": round(w, 2),
|
|
"height_pt": round(h, 2),
|
|
"width_mm": round(w * POINT_TO_MM, 1),
|
|
"height_mm": round(h * POINT_TO_MM, 1),
|
|
"orientation": "landscape" if w > h else "portrait",
|
|
"looks_a4": _looks_a4(w, h),
|
|
})
|
|
return out
|
|
|
|
|
|
def _pdf_info(data: bytes, content_type: str) -> dict[str, Any]:
|
|
info: dict[str, Any] = {
|
|
"content_type": content_type,
|
|
"bytes": len(data),
|
|
"sha256_12": hashlib.sha256(data).hexdigest()[:12] if data else "",
|
|
"is_pdf_header": data[:5] == b"%PDF-",
|
|
"page_count_hint": None,
|
|
"boxes": [],
|
|
"format_warning": "",
|
|
}
|
|
if not data:
|
|
info["format_warning"] = "empty_response"
|
|
return info
|
|
if not info["is_pdf_header"]:
|
|
info["format_warning"] = "not_pdf_header"
|
|
info["body_preview"] = data[:500].decode("utf-8", errors="replace")
|
|
return info
|
|
latin = data.decode("latin-1", errors="ignore")
|
|
info["page_count_hint"] = len(re.findall(r"/Type\s*/Page(?!s)\b", latin)) or None
|
|
boxes = _extract_pdf_boxes(data)
|
|
info["boxes"] = boxes
|
|
if boxes and not any(b.get("looks_a4") for b in boxes if _s(b.get("kind")).lower() == "mediabox"):
|
|
info["format_warning"] = "non_a4_or_unexpected_mediabox"
|
|
return info
|
|
|
|
|
|
def _box_summary(info: dict[str, Any]) -> str:
|
|
boxes = info.get("boxes") or []
|
|
if not boxes:
|
|
return f"content_type={info.get('content_type')} bytes={info.get('bytes')} warning={info.get('format_warning')}"
|
|
b = boxes[0]
|
|
return f"{b.get('kind')} {b.get('width_mm')}x{b.get('height_mm')}mm a4={b.get('looks_a4')} warning={info.get('format_warning') or ''}"
|
|
|
|
|
|
def _selected_invoice_fields(raw: Any) -> dict[str, Any]:
|
|
if not isinstance(raw, dict):
|
|
return {"raw_type": type(raw).__name__, "raw_preview": _s(raw)[:500]}
|
|
rx = re.compile(r"print|report|layout|template|format|paper|serie|series|type|fiscal|operation", re.I)
|
|
interesting = {k: raw.get(k) for k in sorted(raw.keys()) if rx.search(k)}
|
|
return {
|
|
"id": raw.get("id") or raw.get("key"),
|
|
"naturalKey": raw.get("naturalKey"),
|
|
"documentType": raw.get("documentType") or raw.get("documentTypeKey"),
|
|
"documentTypeDescription": raw.get("documentTypeDescription"),
|
|
"serie": raw.get("serie") or raw.get("serieKey"),
|
|
"seriesNumber": raw.get("seriesNumber") or raw.get("number"),
|
|
"company": raw.get("company") or raw.get("companyKey"),
|
|
"buyerCustomerPartyName": raw.get("buyerCustomerPartyName"),
|
|
"payableAmount": raw.get("payableAmount"),
|
|
"taxExclusiveAmount": raw.get("taxExclusiveAmount"),
|
|
"printLayout": raw.get("printLayout"),
|
|
"printedReportName": raw.get("printedReportName"),
|
|
"isPrinted": raw.get("isPrinted"),
|
|
"isReprinted": raw.get("isReprinted"),
|
|
"fiscalDocumentType": raw.get("fiscalDocumentType"),
|
|
"interesting_fields": interesting,
|
|
}
|
|
|
|
|
|
def _query_local_invoice(value: str) -> dict[str, Any] | None:
|
|
with engine.begin() as conn:
|
|
row = conn.execute(text("""
|
|
SELECT id::text AS local_doc_id, opportunity_id::text, external_id, document_number,
|
|
document_type, serie, series_number, status, total_amount, payload, created_at
|
|
FROM commercial_documents
|
|
WHERE system = 'jasmin'
|
|
AND document_kind = 'invoice'
|
|
AND (document_number = :v OR external_id = :v OR id::text = :v)
|
|
ORDER BY created_at DESC
|
|
LIMIT 1
|
|
"""), {"v": value}).mappings().first()
|
|
return dict(row) if row else None
|
|
|
|
|
|
async def _raw_pdf_request(client: JasminClient, path: str, *, params: dict[str, Any] | None = None) -> tuple[int, bytes, str, dict[str, str]]:
|
|
token = await client.get_token()
|
|
url = f"{client.api_root}/{path.lstrip('/')}"
|
|
headers = {"Authorization": f"Bearer {token}", "Accept": "application/pdf"}
|
|
async with httpx.AsyncClient(timeout=client.timeout, follow_redirects=True) as hc:
|
|
response = await hc.get(url, params=params or {}, headers=headers)
|
|
return response.status_code, response.content, response.headers.get("content-type", ""), {
|
|
"content-disposition": response.headers.get("content-disposition", ""),
|
|
"location": response.headers.get("location", ""),
|
|
}
|
|
|
|
|
|
def _variant_candidates(args: argparse.Namespace) -> list[dict[str, Any]]:
|
|
report_names = []
|
|
for name in [args.report_name, "Sales_InvoiceReportTaxsummaryTaxExclusive", "Sales_InvoiceReportTaxsummaryTaxIncluded", "Sales_InvoiceReport", "Sales_InvoiceReportTaxsummary"]:
|
|
name = _s(name)
|
|
if name and name not in report_names:
|
|
report_names.append(name)
|
|
|
|
variants: list[dict[str, Any]] = [{"name": "baseline", "params": {}}]
|
|
param_names = [
|
|
"printedReportName", "reportName", "report", "reportKey", "reportId",
|
|
"printLayout", "layout", "layoutKey", "template", "templateKey",
|
|
"paper", "paperSize", "format", "pageSize",
|
|
]
|
|
for report in report_names:
|
|
for param in param_names:
|
|
if param in {"paper", "paperSize", "format", "pageSize"}:
|
|
continue
|
|
variants.append({"name": f"{param}={report}", "params": {param: report}})
|
|
for param in ("paper", "paperSize", "format", "pageSize"):
|
|
variants.append({"name": f"{param}=A4+reportName", "params": {param: "A4", "reportName": report}})
|
|
for param in ("paper", "paperSize", "format", "pageSize"):
|
|
variants.append({"name": f"{param}=A4", "params": {param: "A4"}})
|
|
return variants
|
|
|
|
|
|
def _write_outputs(result: dict[str, Any]) -> None:
|
|
json_path = Path(OUT_PREFIX + ".json")
|
|
md_path = Path(OUT_PREFIX + ".md")
|
|
csv_path = Path(OUT_PREFIX + ".csv")
|
|
json_path.write_text(json.dumps(result, ensure_ascii=False, indent=2, default=_json_default), encoding="utf-8")
|
|
|
|
rows = result.get("variants") or []
|
|
ok = [r for r in rows if not ((r.get("pdf") or {}).get("format_warning")) and (r.get("pdf") or {}).get("is_pdf_header")]
|
|
lines = [
|
|
"# Probe Jasmin invoice print API",
|
|
"",
|
|
f"- Invoice input: `{result.get('requested_invoice')}`",
|
|
f"- Invoice external ID: `{result.get('invoice_external_id')}`",
|
|
f"- Natural key: `{(result.get('remote_invoice') or {}).get('naturalKey') or ''}`",
|
|
f"- Variants tested: `{len(rows)}`",
|
|
f"- A4/OK variants: `{len(ok)}`",
|
|
"",
|
|
"## Resultado",
|
|
]
|
|
if ok:
|
|
lines.append("Encontrada pelo menos uma variante que devolve PDF A4/sem warning:")
|
|
for r in ok[:10]:
|
|
lines.append(f"- `{r.get('name')}` params=`{r.get('params')}` → `{_box_summary(r.get('pdf') or {})}`")
|
|
else:
|
|
lines.append("Nenhuma variante testada devolveu A4. O endpoint /print parece ignorar layout/report por query string, ou o layout FA2026 está configurado no Jasmin como formato estreito.")
|
|
lines.extend(["", "## Amostra de variantes", ""])
|
|
for r in rows[:25]:
|
|
lines.append(f"- `{r.get('name')}` status={r.get('status_code')} `{_box_summary(r.get('pdf') or {})}`")
|
|
md_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
with csv_path.open("w", encoding="utf-8", newline="") as fh:
|
|
writer = csv.DictWriter(fh, fieldnames=["name", "status_code", "content_type", "bytes", "is_pdf", "box", "warning", "params"])
|
|
writer.writeheader()
|
|
for r in rows:
|
|
p = r.get("pdf") or {}
|
|
writer.writerow({
|
|
"name": r.get("name"),
|
|
"status_code": r.get("status_code"),
|
|
"content_type": p.get("content_type"),
|
|
"bytes": p.get("bytes"),
|
|
"is_pdf": p.get("is_pdf_header"),
|
|
"box": _box_summary(p),
|
|
"warning": p.get("format_warning"),
|
|
"params": json.dumps(r.get("params") or {}, ensure_ascii=False),
|
|
})
|
|
print(f"JSON: {json_path}")
|
|
print(f"Markdown: {md_path}")
|
|
print(f"CSV: {csv_path}")
|
|
|
|
|
|
async def _main_async(args: argparse.Namespace) -> int:
|
|
client = JasminClient()
|
|
local = _query_local_invoice(args.invoice)
|
|
invoice_id = _s(args.invoice_external_id or (local or {}).get("external_id") or args.invoice)
|
|
remote: dict[str, Any] = {}
|
|
try:
|
|
remote_raw = await client.get_invoice(invoice_id)
|
|
remote = _selected_invoice_fields(remote_raw)
|
|
if remote.get("id"):
|
|
invoice_id = _s(remote.get("id"))
|
|
except Exception as exc:
|
|
remote = {"fetch_error": str(exc)}
|
|
|
|
save_dir = Path(args.save_pdf_dir) if args.save_pdf_dir else None
|
|
if save_dir:
|
|
save_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
variants: list[dict[str, Any]] = []
|
|
candidates = _variant_candidates(args)[: max(1, args.max_variants)]
|
|
for i, candidate in enumerate(candidates, start=1):
|
|
name = candidate["name"]
|
|
params = candidate["params"]
|
|
try:
|
|
status, content, content_type, headers_subset = await _raw_pdf_request(
|
|
client, f"/billing/invoices/{invoice_id}/print", params=params
|
|
)
|
|
pdf = _pdf_info(content, content_type)
|
|
if save_dir and status < 400 and pdf.get("is_pdf_header"):
|
|
safe = re.sub(r"[^A-Za-z0-9_.-]+", "_", f"{i:02d}_{name}").strip("_")
|
|
(save_dir / f"{safe}.pdf").write_bytes(content)
|
|
variants.append({
|
|
"name": name,
|
|
"params": params,
|
|
"status_code": status,
|
|
"headers_subset": headers_subset,
|
|
"pdf": pdf,
|
|
})
|
|
print(f"{i:02d}/{len(candidates)} {name}: status={status} {_box_summary(pdf)}")
|
|
except Exception as exc:
|
|
variants.append({"name": name, "params": params, "error": str(exc)})
|
|
print(f"{i:02d}/{len(candidates)} {name}: ERROR {exc}")
|
|
|
|
ok = [r for r in variants if (r.get("pdf") or {}).get("is_pdf_header") and not (r.get("pdf") or {}).get("format_warning")]
|
|
result = {
|
|
"requested_invoice": args.invoice,
|
|
"invoice_external_id": invoice_id,
|
|
"resolved_at": datetime.utcnow().isoformat() + "Z",
|
|
"local_invoice": local or {},
|
|
"remote_invoice": remote,
|
|
"variants": variants,
|
|
"recommendation": (
|
|
"Encontrada variante A4 via /print. Adaptar ClientFlow para usar esses parâmetros no download/envio da fatura."
|
|
if ok else
|
|
"Nenhuma variante /print testada devolveu A4. Corrigir layout/template da série FA2026 no Jasmin ou usar outro endpoint oficial de relatório; manter bloqueio de envio automático no ClientFlow."
|
|
),
|
|
}
|
|
print("SUMMARY")
|
|
print(f"variants={len(variants)} a4_ok={len(ok)}")
|
|
print(result["recommendation"])
|
|
_write_outputs(result)
|
|
return 0
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(description="Probe Jasmin invoice print endpoint variants without creating documents.")
|
|
p.add_argument("--invoice", required=True, help="Invoice document number/local id/external id, e.g. FA.FA2026.137")
|
|
p.add_argument("--invoice-external-id", default="", help="Explicit Jasmin invoice GUID; useful for invoices not yet synced locally")
|
|
p.add_argument("--report-name", default="", help="Preferred invoice report/layout name to test first")
|
|
p.add_argument("--save-pdf-dir", default="", help="Directory to save returned PDFs")
|
|
p.add_argument("--max-variants", type=int, default=80)
|
|
return p
|
|
|
|
|
|
def main() -> int:
|
|
return asyncio.run(_main_async(build_parser().parse_args()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|