"""DueBuddy API — рабочий пример на Python. Разбор счетов в поля. ЗАПУСК (ничего устанавливать не нужно, только Python 3.8+): # один счёт python api_client.py invoice.pdf # целая папка со счетами (PDF, JPG, PNG) python api_client.py ~/invoices/ # сохранить результат в файл python api_client.py ~/invoices/ --json result.json --csv result.csv КЛЮЧ берётся из переменной окружения DUEBUDDY_API_KEY, либо впишите его в файл ключа ниже. Так безопаснее: ключ не попадает в текст программы, которую вы кому-то показываете. export DUEBUDDY_API_KEY="ваш_ключ" python api_client.py invoice.pdf ЧТО ПОЛУЧИТЕ: номер счёта, даты, суммы, поставщика, покупателя, налоговый номер, IBAN. Пустое поле — это честный ответ: значение не подтвердилось, и система не стала его выдумывать. """ from __future__ import annotations import argparse import csv import json import os import pathlib import sys import time import urllib.error import urllib.request import uuid BASE_URL = "https://app.duebuddy.work" TIMEOUT_SECONDS = 180 POLL_EVERY_SECONDS = 3 SUPPORTED = {".pdf", ".jpg", ".jpeg", ".png"} # Поля, которые показываем в сводке (в файл попадают все) KEY_FIELDS = ("invoice_no", "invoice_date", "due_date", "currency", "seller_name", "seller_vat_id", "seller_iban", "gross_amount") def _key() -> str: """Ключ из переменной окружения — так он не хранится в тексте программы.""" key = os.environ.get("DUEBUDDY_API_KEY", "").strip() if not key: print("Не задан ключ. Выполните: export DUEBUDDY_API_KEY=\"ваш_ключ\"", file=sys.stderr) raise SystemExit(2) return key def _mime(name: str) -> str: ext = pathlib.Path(name).suffix.lower() if ext == ".png": return "image/png" if ext in (".jpg", ".jpeg"): return "image/jpeg" return "application/pdf" def send_invoice(path: pathlib.Path, key: str) -> int: """Отправить файл. Возвращает номер задания.""" boundary = uuid.uuid4().hex body = ( f'--{boundary}\r\n' f'Content-Disposition: form-data; name="file"; filename="{path.name}"\r\n' f'Content-Type: {_mime(path.name)}\r\n\r\n' ).encode() + path.read_bytes() + f"\r\n--{boundary}--\r\n".encode() request = urllib.request.Request( f"{BASE_URL}/v1/parse", data=body, method="POST", headers={"X-Api-Key": key, "Content-Type": f"multipart/form-data; boundary={boundary}"}) with urllib.request.urlopen(request, timeout=120) as response: return int(json.load(response)["job_id"]) def wait_result(job_id: int, key: str) -> dict: """Дождаться результата разбора.""" deadline = time.time() + TIMEOUT_SECONDS while time.time() < deadline: time.sleep(POLL_EVERY_SECONDS) request = urllib.request.Request(f"{BASE_URL}/v1/parse/{job_id}", headers={"X-Api-Key": key}) with urllib.request.urlopen(request, timeout=60) as response: result = json.load(response) if result.get("status") in ("done", "failed", "error"): return result return {"status": "timeout", "job_id": job_id} def parse_one(path: pathlib.Path, key: str) -> dict: """Разобрать один счёт. Никогда не падает — ошибку возвращает как результат.""" try: job_id = send_invoice(path, key) result = wait_result(job_id, key) fields = result.get("fields") or {} if isinstance(fields, list): # ответ может прийти списком пар fields = {item.get("key"): item.get("value") for item in fields} return {"file": path.name, "status": result.get("status"), "method": result.get("method"), "charged": result.get("charged"), "fields": {k: v for k, v in fields.items() if str(v or "").strip()}} except urllib.error.HTTPError as exc: hints = {401: "ключ неверный или отозван", 402: "нет активной подписки", 413: "файл слишком большой", 415: "формат не поддерживается", 429: "слишком много запросов, подождите"} return {"file": path.name, "status": "error", "error": f"HTTP {exc.code}: {hints.get(exc.code, exc.reason)}"} except Exception as exc: # noqa: BLE001 return {"file": path.name, "status": "error", "error": f"{type(exc).__name__}: {exc}"} def collect(target: pathlib.Path) -> list: """Собрать файлы: один файл или все поддерживаемые в папке.""" if target.is_file(): return [target] return sorted(p for p in target.rglob("*") if p.suffix.lower() in SUPPORTED) def main() -> int: parser = argparse.ArgumentParser( description="Разбор счетов через DueBuddy API: отправляет файлы и печатает найденные поля.") parser.add_argument("target", help="файл счёта или папка со счетами") parser.add_argument("--json", dest="json_out", help="сохранить все поля в JSON-файл") parser.add_argument("--csv", dest="csv_out", help="сохранить все поля в CSV (для таблиц)") args = parser.parse_args() key = _key() target = pathlib.Path(args.target).expanduser() if not target.exists(): print(f"Не найдено: {target}", file=sys.stderr) return 2 files = collect(target) if not files: print("В папке нет файлов PDF/JPG/PNG.", file=sys.stderr) return 2 print(f"Счетов к разбору: {len(files)}\n") results = [] for i, path in enumerate(files, 1): result = parse_one(path, key) results.append(result) if result["status"] == "done": f = result["fields"] print(f"[{i}/{len(files)}] {path.name} — {result.get('method')}") for name in KEY_FIELDS: if f.get(name): print(f" {name:16} {f[name]}") else: print(f"[{i}/{len(files)}] {path.name} — ОШИБКА: {result.get('error', result['status'])}") done = [r for r in results if r["status"] == "done"] print(f"\nРазобрано: {len(done)} из {len(files)}") if args.json_out: pathlib.Path(args.json_out).write_text( json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8") print(f"JSON сохранён: {args.json_out}") if args.csv_out: columns = sorted({k for r in done for k in r["fields"]}) with open(args.csv_out, "w", newline="", encoding="utf-8-sig") as fh: writer = csv.writer(fh, delimiter=";") writer.writerow(["file"] + columns) for r in done: writer.writerow([r["file"]] + [r["fields"].get(c, "") for c in columns]) print(f"CSV сохранён: {args.csv_out}") return 0 if __name__ == "__main__": raise SystemExit(main())