#!/usr/bin/env python3 """Probe Jasmin quotation -> invoice API using a specific quotation. Safe by default: it only reads local DB/Jasmin, fetches PDFs, and prints a recommended experiment plan. It will only create an invoice in Jasmin when the operator passes --execute-convert plus --confirm-document matching the quotation. Use case: understand whether fromQuotation can be parameterized so ClientFlow can convert ORC -> FA without using the Jasmin UI and still obtain an A4 invoice PDF. """ from __future__ import annotations import argparse import asyncio import csv import hashlib import json import os import re from datetime import date, datetime from pathlib import Path from typing import Any, Iterable import httpx from sqlalchemy import text from app.config import settings from app.db import engine from app.jasmin_client import JasminClient, JasminError A4_PORTRAIT_PT = (595.0, 842.0) POINT_TO_MM = 25.4 / 72.0 OUT_PREFIX = "/tmp/clientflow_jasmin_quotation_invoice_api_probe" 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 _payload(value: Any) -> dict[str, Any]: if isinstance(value, dict): return value if not value: return {} try: return json.loads(value) except Exception: return {} def _first(data: dict[str, Any], keys: Iterable[str]) -> Any: for key in keys: if data.get(key) not in (None, ""): return data.get(key) return None def _money(value: Any) -> str: if isinstance(value, dict): for k in ("amount", "value", "baseAmount", "reportingAmount"): if value.get(k) is not None: return _s(value.get(k)) return _s(value) def _interesting_fields(raw: dict[str, Any]) -> dict[str, Any]: rx = re.compile(r"print|report|layout|template|format|paper|serie|series|type|fiscal|operation", re.I) return {k: raw.get(k) for k in sorted(raw.keys()) if rx.search(k)} def _selected_doc_fields(raw: Any) -> dict[str, Any]: if not isinstance(raw, dict): return {"raw_type": type(raw).__name__, "raw_preview": _s(raw)[:500]} return { "id": _first(raw, ["id", "key"]), "naturalKey": raw.get("naturalKey"), "documentNumber": raw.get("documentNumber"), "documentType": _first(raw, ["documentType", "documentTypeKey"]), "documentTypeDescription": raw.get("documentTypeDescription"), "serie": _first(raw, ["serie", "serieKey"]), "serieDescription": raw.get("serieDescription"), "seriesNumber": _first(raw, ["seriesNumber", "number"]), "company": _first(raw, ["company", "companyKey"]), "documentDate": raw.get("documentDate"), "postingDate": raw.get("postingDate"), "currency": _first(raw, ["currency", "currencyKey"]), "totalAmount": _money(raw.get("totalAmount")), "payableAmount": _money(raw.get("payableAmount")), "taxExclusiveAmount": _money(raw.get("taxExclusiveAmount")), "buyerCustomerParty": _first(raw, ["buyerCustomerParty", "buyerCustomerPartyKey"]), "buyerCustomerPartyName": raw.get("buyerCustomerPartyName"), "paymentTerm": _first(raw, ["paymentTerm", "paymentTermKey"]), "paymentMethod": _first(raw, ["paymentMethod", "paymentMethodKey"]), "priceList": _first(raw, ["priceList", "priceListKey"]), "printLayout": _first(raw, ["printLayout", "report", "reportName", "reportKey", "documentReport", "printTemplate"]), "printedReportName": raw.get("printedReportName"), "isPrinted": raw.get("isPrinted"), "isReprinted": raw.get("isReprinted"), "operationType": raw.get("operationType"), "fiscalDocumentType": raw.get("fiscalDocumentType"), "line_count": len(raw.get("documentLines") or raw.get("lines") or []) if isinstance(raw.get("documentLines") or raw.get("lines") or [], list) else None, "interesting_fields": _interesting_fields(raw), } def _looks_a4(w: float, h: float, tolerance_pt: float = 20.0) -> bool: pairs = [(w, h), (h, w)] return any(abs(a - A4_PORTRAIT_PT[0]) <= tolerance_pt and abs(b - A4_PORTRAIT_PT[1]) <= tolerance_pt for a, b in pairs) def _extract_pdf_boxes(data: bytes) -> list[dict[str, Any]]: latin = data.decode("latin-1", errors="ignore") boxes: list[dict[str, Any]] = [] pattern = re.compile(r"/(MediaBox|CropBox)\s*\[\s*([-+0-9.]+)\s+([-+0-9.]+)\s+([-+0-9.]+)\s+([-+0-9.]+)\s*\]", re.I) 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 = abs(x1 - x0) h = abs(y1 - y0) boxes.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 boxes 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[:300].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(bool(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 _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") q = result.get("quotation") or {} qp = result.get("quotation_pdf") or {} linked = result.get("linked_invoice") or {} cp = result.get("conversion_probe") or {} lines = [ "# Probe Jasmin ORC → FA via API", "", f"- Quotation: `{q.get('local_number') or q.get('remote', {}).get('naturalKey') or result.get('requested_quotation')}`", f"- External ID: `{q.get('external_id') or q.get('remote', {}).get('id')}`", f"- Quotation PDF: `{_pdf_box_summary(qp)}`", f"- Linked invoice: `{linked.get('invoice_number') or '—'}`", f"- API fromQuotation OPTIONS/POST surface: `{cp.get('summary') or 'ver JSON'}`", "", "## Recomendação", result.get("recommendation") or "Sem recomendação automática.", "", "## Body candidates", "```json", json.dumps(result.get("body_candidates") or {}, ensure_ascii=False, indent=2, default=_json_default), "```", ] 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=[ "quotation", "quotation_external_id", "quotation_pdf", "linked_invoice", "linked_invoice_pdf", "recommendation_short", ]) writer.writeheader() writer.writerow({ "quotation": q.get("local_number") or result.get("requested_quotation"), "quotation_external_id": q.get("external_id") or q.get("remote", {}).get("id"), "quotation_pdf": _pdf_box_summary(qp), "linked_invoice": linked.get("invoice_number") or "", "linked_invoice_pdf": _pdf_box_summary(linked.get("invoice_pdf") or {}), "recommendation_short": _s(result.get("recommendation")).split("\n")[0][:300], }) print(f"JSON: {json_path}") print(f"Markdown: {md_path}") print(f"CSV: {csv_path}") def _pdf_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')}" box = boxes[0] return f"{box.get('kind')} {box.get('width_mm')}x{box.get('height_mm')}mm a4={box.get('looks_a4')} warning={info.get('format_warning') or ''}" def _parse_orc_number(value: str) -> dict[str, Any]: m = re.match(r"^([A-Z]+)\.([A-Z]+\d{4})\.(\d+)$", value.strip(), re.I) if not m: return {} return {"document_type": m.group(1).upper(), "serie": m.group(2).upper(), "series_number": int(m.group(3))} def _query_local_quotation(number_or_id: str) -> dict[str, Any] | None: with engine.begin() as conn: row = conn.execute(text(""" SELECT q.id::text AS local_doc_id, q.opportunity_id::text AS opportunity_id, q.external_id, q.document_number, q.document_type, q.serie, q.series_number, q.status, q.total_amount, q.payload, q.created_at, o.title AS opportunity_title, o.customer_name AS opportunity_customer_name, c.name AS fiscal_customer_name, c.tax_id AS fiscal_customer_tax_id, inv.id::text AS linked_invoice_doc_id, inv.external_id AS linked_invoice_external_id, inv.document_number AS linked_invoice_number FROM commercial_documents q LEFT JOIN commercial_documents inv ON inv.parent_document_id = q.id AND inv.document_kind = 'invoice' LEFT JOIN opportunities o ON o.id = q.opportunity_id LEFT JOIN customers c ON c.id = q.customer_id WHERE q.system = 'jasmin' AND q.document_kind = 'quotation' AND (q.document_number = :key OR q.external_id = :key OR q.id::text = :key) ORDER BY q.created_at DESC LIMIT 1 """), {"key": number_or_id}).mappings().first() return dict(row) if row else None async def _resolve_remote_quotation(client: JasminClient, key: str, local: dict[str, Any] | None) -> tuple[str, dict[str, Any], dict[str, Any]]: if local and _s(local.get("external_id")): qid = _s(local.get("external_id")) return qid, await client.get_quotation(qid), {"source": "local_external_id"} parsed = _parse_orc_number(key) filters = [f"naturalKey eq '{key}'"] if parsed: filters.append( "documentType eq '{document_type}' and serie eq '{serie}' and seriesNumber eq {series_number}".format(**parsed) ) errors: list[str] = [] for flt in filters: try: data = await client.list_quotations(top=5, filter=flt, orderby="documentDate desc") values = data.get("value") if isinstance(data, dict) else [] if values: remote = values[0] qid = _s(remote.get("id") or remote.get("key")) if qid: return qid, await client.get_quotation(qid), {"source": "odata", "filter": flt, "odata_hit": _selected_doc_fields(remote)} except Exception as exc: errors.append(f"{flt}: {exc}") raise JasminError(f"Não consegui resolver orçamento remoto {key}. Erros OData: {errors}") async def _raw_request_with_meta(client: JasminClient, method: str, path: str, *, params: dict[str, Any] | None = None, json_body: Any = None, accept: str = "application/json") -> dict[str, Any]: token = await client.get_token() url = f"{client.api_root}/{path.lstrip('/')}" headers = {"Authorization": f"Bearer {token}", "Accept": accept} if json_body is not None: headers["Content-Type"] = "application/json" async with httpx.AsyncClient(timeout=client.timeout, follow_redirects=True) as http: resp = await http.request(method.upper(), url, params=params, json=json_body, headers=headers) content_type = resp.headers.get("content-type", "") body: Any try: body = resp.json() except Exception: body = resp.text[:2000] return { "method": method.upper(), "path": path, "params": params or {}, "status_code": resp.status_code, "content_type": content_type, "headers_subset": {k: v for k, v in resp.headers.items() if k.lower() in {"allow", "content-type", "location"}}, "body_preview": body, } def _body_candidates(remote_quote: dict[str, Any], args: argparse.Namespace) -> dict[str, Any]: company = _first(remote_quote, ["company", "companyKey"]) or settings.jasmin_company_key document_date = args.document_date or date.today().isoformat() invoice_type = args.invoice_type invoice_serie = args.invoice_serie base = { "documentType": invoice_type, "serie": invoice_serie, "company": company, "documentDate": document_date, "postingDate": document_date, } inherited_print: dict[str, Any] = {} for key in ("printLayout", "printedReportName", "printAllDiscounts", "printAllPaymentMethods"): if remote_quote.get(key) not in (None, ""): inherited_print[key] = remote_quote.get(key) if args.print_layout: inherited_print["printLayout"] = args.print_layout if args.printed_report_name: inherited_print["printedReportName"] = args.printed_report_name return { "current_empty": {}, "explicit_invoice_type_serie": base, "explicit_invoice_type_serie_plus_print_fields": {**base, **inherited_print}, "notes": [ "Não executar todos os candidatos no mesmo orçamento: cada POST pode criar uma fatura real.", "Se o endpoint ignorar estes campos, a correção tem de ser no print endpoint ou na configuração Jasmin da série FA.", ], } def _body_for_mode(candidates: dict[str, Any], args: argparse.Namespace) -> dict[str, Any]: if args.body_json: parsed = json.loads(args.body_json) if not isinstance(parsed, dict): raise SystemExit("--body-json tem de ser um objeto JSON") return parsed key_map = { "empty": "current_empty", "explicit-fa": "explicit_invoice_type_serie", "explicit-fa-print": "explicit_invoice_type_serie_plus_print_fields", } return dict(candidates.get(key_map[args.body_mode]) or {}) async def _fetch_and_save_pdf(client: JasminClient, kind: str, external_id: str, save_dir: Path | None, label: str) -> dict[str, Any]: if kind == "quotation": data, content_type = await client.print_quotation_pdf(external_id) elif kind == "invoice": data, content_type = await client.print_invoice_pdf(external_id) else: raise ValueError(kind) if save_dir: save_dir.mkdir(parents=True, exist_ok=True) safe = re.sub(r"[^A-Za-z0-9_.-]+", "_", label).strip("_") or external_id (save_dir / f"{safe}.pdf").write_bytes(data) return _pdf_info(data, content_type) async def _main_async(args: argparse.Namespace) -> int: client = JasminClient() local = _query_local_quotation(args.quotation) qid, remote_quote, resolve_meta = await _resolve_remote_quotation(client, args.quotation, local) save_dir = Path(args.save_pdf_dir) if args.save_pdf_dir else None body_candidates = _body_candidates(remote_quote, args) result: dict[str, Any] = { "requested_quotation": args.quotation, "resolved_at": datetime.utcnow().isoformat() + "Z", "quotation": { "local_number": _s(local.get("document_number")) if local else "", "local_doc_id": _s(local.get("local_doc_id")) if local else "", "opportunity_id": _s(local.get("opportunity_id")) if local else "", "external_id": qid, "resolve_meta": resolve_meta, "remote": _selected_doc_fields(remote_quote), "local_payload_keys": sorted(_payload(local.get("payload") if local else None).keys()), }, "quotation_pdf": await _fetch_and_save_pdf(client, "quotation", qid, save_dir, f"{args.quotation}_quotation"), "linked_invoice": {}, "api_surface": {}, "body_candidates": body_candidates, "conversion_probe": {"executed": False}, } if local and _s(local.get("linked_invoice_external_id")): inv_id = _s(local.get("linked_invoice_external_id")) remote_invoice = await client.get_invoice(inv_id) result["linked_invoice"] = { "invoice_number": _s(local.get("linked_invoice_number")), "external_id": inv_id, "remote": _selected_doc_fields(remote_invoice), "invoice_pdf": await _fetch_and_save_pdf(client, "invoice", inv_id, save_dir, f"{local.get('linked_invoice_number') or inv_id}_invoice"), } if args.probe_options: surface: dict[str, Any] = {} for method, path in [ ("OPTIONS", f"/billing/invoices/fromQuotation/{qid}"), ("GET", f"/billing/invoices/fromQuotation/{qid}"), ]: try: surface[f"{method} {path}"] = await _raw_request_with_meta(client, method, path) except Exception as exc: surface[f"{method} {path}"] = {"error": str(exc)} result["api_surface"] = surface if args.execute_convert: if args.confirm_document != args.quotation: raise SystemExit("Segurança: para criar fatura real, passa --confirm-document exatamente igual a --quotation") body = _body_for_mode(body_candidates, args) path = f"/billing/invoices/fromQuotation/{qid}" conv = await _raw_request_with_meta(client, "POST", path, json_body=body) conv["body_sent"] = body conv["warning"] = "Este POST pode ter criado uma fatura fiscal real no Jasmin. Importar/sincronizar no ClientFlow se for documento válido." new_id = _s(conv.get("body_preview")).strip('"') if conv.get("status_code", 0) < 400 else "" if isinstance(conv.get("body_preview"), str): new_id = _s(conv.get("body_preview")).strip('"') elif isinstance(conv.get("body_preview"), dict): new_id = _s(conv["body_preview"].get("id") or conv["body_preview"].get("key") or conv["body_preview"].get("value")) conv["created_invoice_external_id_guess"] = new_id if new_id: try: new_invoice = await client.get_invoice(new_id) conv["created_invoice_remote"] = _selected_doc_fields(new_invoice) natural = _s(new_invoice.get("naturalKey") or new_invoice.get("documentNumber") or new_id) conv["created_invoice_pdf"] = await _fetch_and_save_pdf(client, "invoice", new_id, save_dir, f"{natural}_invoice") except Exception as exc: conv["created_invoice_fetch_error"] = str(exc) result["conversion_probe"] = {"executed": True, **conv} result["recommendation"] = _recommend(result) _print_summary(result) _write_outputs(result) return 0 def _recommend(result: dict[str, Any]) -> str: qpdf = result.get("quotation_pdf") or {} linked = result.get("linked_invoice") or {} ipdf = linked.get("invoice_pdf") or {} conv = result.get("conversion_probe") or {} cpdf = conv.get("created_invoice_pdf") or {} if cpdf: if not cpdf.get("format_warning"): return "A variante executada criou fatura com PDF A4. Adaptar ClientFlow para usar esse body/estratégia e sincronizar a fatura criada." return "A variante executada também devolveu PDF não-A4. O body de conversão não resolveu; testar print endpoint/layout ou corrigir configuração Jasmin da série FA." if qpdf and not qpdf.get("format_warning") and ipdf and ipdf.get("format_warning"): return ( "O ORC imprime A4, mas a FA ligada imprime em formato estranho. Antes de assumir erro de configuração, " "usar este script com --execute-convert num orçamento de teste/real autorizado e body-mode explicit-fa-print. " "Se continuar não-A4, a correção é no endpoint de print/layout da FA ou na configuração da série FA2026." ) return ( "Probe read-only concluído. Verifica campos interesting_fields e body_candidates. " "Só executar conversão com --execute-convert se este ORC puder gerar uma fatura real." ) def _print_summary(result: dict[str, Any]) -> None: q = result.get("quotation") or {} print(f"QUOTATION={result.get('requested_quotation')} external_id={q.get('external_id')}") print(f"quotation_pdf={_pdf_box_summary(result.get('quotation_pdf') or {})}") linked = result.get("linked_invoice") or {} if linked: print(f"linked_invoice={linked.get('invoice_number')} external_id={linked.get('external_id')}") print(f"linked_invoice_pdf={_pdf_box_summary(linked.get('invoice_pdf') or {})}") else: print("linked_invoice=none") conv = result.get("conversion_probe") or {} if conv.get("executed"): print(f"conversion_status={conv.get('status_code')} created_invoice={conv.get('created_invoice_external_id_guess')}") print(f"created_invoice_pdf={_pdf_box_summary(conv.get('created_invoice_pdf') or {})}") else: print("conversion_executed=False") print("RECOMMENDATION:") print(result.get("recommendation") or "") def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description="Probe Jasmin ORC->FA conversion API safely.") p.add_argument("--quotation", required=True, help="Quotation document number/external id, e.g. ORC.ORC2026.185") p.add_argument("--save-pdf-dir", default="", help="Directory where fetched PDFs are saved for inspection") p.add_argument("--probe-options", action=argparse.BooleanOptionalAction, default=True, help="Probe safe OPTIONS/GET surface for fromQuotation endpoint") p.add_argument("--invoice-type", default=os.getenv("JASMIN_INVOICE_TYPE", "FA")) p.add_argument("--invoice-serie", default=os.getenv("JASMIN_INVOICE_SERIE", "FA2026")) p.add_argument("--document-date", default="", help="YYYY-MM-DD used in body candidates; defaults to today") p.add_argument("--print-layout", default=os.getenv("JASMIN_INVOICE_PRINT_LAYOUT", "")) p.add_argument("--printed-report-name", default=os.getenv("JASMIN_INVOICE_PRINTED_REPORT_NAME", "")) p.add_argument("--execute-convert", action="store_true", help="DANGEROUS: performs POST fromQuotation and can create a real invoice") p.add_argument("--confirm-document", default="", help="Required with --execute-convert; must exactly match --quotation") p.add_argument("--body-mode", choices=["empty", "explicit-fa", "explicit-fa-print"], default="explicit-fa-print") p.add_argument("--body-json", default="", help="Custom JSON object body for conversion; overrides --body-mode") return p def main() -> int: args = build_parser().parse_args() return asyncio.run(_main_async(args)) if __name__ == "__main__": raise SystemExit(main())