#!/usr/bin/env python3 # Author: Tom Sapletta · https://tom.sapletta.com # Part of the ifURI solution. """Move confidently non-invoice documents into no_invoice via URI. Input is the CSV created by ``invoice_audit.py``. The script is conservative: - OCR must be OK. - Path/preview must NOT contain invoice markers. - Path/preview must contain a non-invoice marker such as regulations, price list, form, terms, instructions. - Files are moved by ``fs://host/file/command/move_to_dir`` under IFURI_FS_ROOT. """ from __future__ import annotations import argparse import csv import json import urllib.error import urllib.request from pathlib import Path from typing import Any DEFAULT_NODE_URL = "http://192.168.188.201:8765" INVOICE_MARKERS = [ "faktura", "invoice", "receipt", "rachunek", "paragon", "vat", "ksef", "nota księgowa", "nota ksiegowa", "tax invoice", ] NON_INVOICE_MARKERS = [ "regulamin", "cennik", "formularz", "odstąpienia", "odstapienia", "warunki", "terms", "conditions", "privacy", "polityka", "instrukcja", "manual", "akceptacja", "świadczenia usług", "swiadczenia uslug", "bezpieczny", "ulotka", ] def bool_value(value: str | bool) -> bool: if isinstance(value, bool): return value return str(value).strip().lower() in {"true", "1", "yes", "tak"} def classify_no_invoice(row: dict[str, str]) -> tuple[bool, str]: if not bool_value(row.get("ok", "")): return False, "ocr_not_ok" haystack = f"{row.get('path', '')}\n{row.get('preview', '')}".lower() invoice_hit = next((marker for marker in INVOICE_MARKERS if marker in haystack), "") if invoice_hit: return False, f"invoice_marker:{invoice_hit}" non_invoice_hit = next((marker for marker in NON_INVOICE_MARKERS if marker in haystack), "") if not non_invoice_hit: return False, "no_non_invoice_marker" return True, f"non_invoice_marker:{non_invoice_hit}" class NodeClient: def __init__(self, base_url: str, timeout: int = 120): self.base_url = base_url.rstrip("/") self.timeout = timeout def run(self, uri: str, payload: dict[str, Any]) -> dict[str, Any]: body = json.dumps({"uri": uri, "payload": payload}).encode("utf-8") request = urllib.request.Request( self.base_url + "/run", data=body, headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(request, timeout=self.timeout) as response: return json.loads(response.read().decode("utf-8") or "{}") except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", "replace") if exc.fp else "" try: return json.loads(raw or "{}") except json.JSONDecodeError: return {"ok": False, "status": exc.code, "error": raw} def route_value(envelope: dict[str, Any]) -> dict[str, Any]: result = envelope.get("result") or {} if isinstance(result, dict) and isinstance(result.get("value"), dict): return result["value"] return result if isinstance(result, dict) else {} def load_candidates(csv_path: Path) -> list[dict[str, str]]: rows: list[dict[str, str]] = [] with csv_path.open(encoding="utf-8", newline="") as handle: for row in csv.DictReader(handle): keep, reason = classify_no_invoice(row) if keep and not str(row.get("path", "")).startswith("no_invoice/"): row["move_reason"] = reason rows.append(row) return rows def write_csv(path: Path, rows: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) fields = ["path", "target", "ok", "moved", "dryRun", "move_reason", "error"] with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() for row in rows: writer.writerow({field: row.get(field, "") for field in fields}) def move_candidates(args: argparse.Namespace, client: NodeClient | None = None) -> dict[str, Any]: client = client or NodeClient(args.node_url, timeout=args.timeout) candidates = load_candidates(Path(args.input_csv)) results: list[dict[str, Any]] = [] for row in candidates: envelope = client.run(args.move_uri, { "path": row["path"], "target_dir": args.target_dir, "preserve_relative": True, "dry_run": not args.execute, "overwrite": False, }) data = route_value(envelope) results.append({ "path": row["path"], "target": data.get("target", ""), "ok": bool(data.get("ok")), "moved": bool(data.get("moved")), "dryRun": data.get("dryRun", not args.execute), "move_reason": row.get("move_reason", ""), "error": data.get("error", envelope.get("error", "")), }) moved_count = sum(1 for row in results if row["moved"]) ok_count = sum(1 for row in results if row["ok"]) payload = { "ok": all(row["ok"] for row in results), "dryRun": not args.execute, "candidate_count": len(candidates), "ok_count": ok_count, "moved_count": moved_count, "target_dir": args.target_dir, "results": results, } out_dir = Path(args.output_dir) out_dir.mkdir(parents=True, exist_ok=True) stem = "no_invoice_move_manifest" if args.execute else "no_invoice_move_plan" json_path = out_dir / f"{stem}.json" csv_path = out_dir / f"{stem}.csv" json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") write_csv(csv_path, results) payload["reports"] = {"json": str(json_path), "csv": str(csv_path)} return payload def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Move confirmed non-invoice files to no_invoice via URI.") parser.add_argument("--node-url", default=DEFAULT_NODE_URL) parser.add_argument("--input-csv", default=".state/invoice_audit_files.csv") parser.add_argument("--output-dir", default=".state") parser.add_argument("--target-dir", default="no_invoice") parser.add_argument("--move-uri", default="fs://host/file/command/move_to_dir") parser.add_argument("--timeout", type=int, default=120) parser.add_argument("--execute", action="store_true", help="actually move files; default is dry-run") return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) result = move_candidates(args) print(json.dumps({ "ok": result["ok"], "dryRun": result["dryRun"], "candidate_count": result["candidate_count"], "ok_count": result["ok_count"], "moved_count": result["moved_count"], "reports": result["reports"], }, ensure_ascii=False, indent=2)) return 0 if result["ok"] else 1 if __name__ == "__main__": raise SystemExit(main())