#!/usr/bin/env python3 # Author: Tom Sapletta ยท https://tom.sapletta.com # Part of the ifURI solution. """Flatten remaining nested files under month folders via URI. This complements ``flatten_invoices.py``. It scans the node filesystem through ``fs://host/dir/query/list`` and moves any still-nested files under month folders to flat month-level paths: ``YYYY.MM/YYYY.MM.DD-category-original.ext`` No OCR is required. The date is extracted from the file path/name when possible; otherwise the day becomes ``00``. """ from __future__ import annotations import argparse import csv import json import re from pathlib import Path, PurePosixPath from typing import Any from flatten_invoices import ( DEFAULT_NODE_URL, NodeClient, add_collision_suffix, extract_date, route_value, slug, trim_component, ) MONTH_RE = re.compile(r"^20\d{2}[.-]\d{2}$") def list_dir(client: NodeClient, uri: str, path: str) -> list[dict[str, Any]]: data = route_value(client.run(uri, {"path": path})) if not data.get("ok"): raise RuntimeError(f"list failed for {path}: {data.get('error') or data}") return list(data.get("entries") or []) def split_words(raw: str) -> list[str]: return [item.strip() for item in raw.replace(";", ",").split(",") if item.strip()] def discover_months(client: NodeClient, list_uri: str, root_path: str, months: str) -> list[str]: explicit = split_words(months) if explicit: return explicit entries = list_dir(client, list_uri, root_path) found = [entry["name"] for entry in entries if entry.get("type") == "dir" and MONTH_RE.match(str(entry["name"]))] return sorted(found) def child_path(parent: str, name: str) -> str: return str(PurePosixPath(parent) / name) if parent not in {"", "."} else name def walk_files(client: NodeClient, list_uri: str, start: str, exclude_parts: set[str]) -> list[str]: files: list[str] = [] stack = [start] while stack: current = stack.pop() if set(PurePosixPath(current).parts).intersection(exclude_parts): continue for entry in list_dir(client, list_uri, current): path = child_path(current, str(entry["name"])) if entry.get("type") == "dir": stack.append(path) elif entry.get("type") == "file": files.append(path) return sorted(files) def top_level_files(client: NodeClient, list_uri: str, months: list[str]) -> set[str]: seen: set[str] = set() for month in months: for entry in list_dir(client, list_uri, month): if entry.get("type") == "file": seen.add(str(PurePosixPath(month) / str(entry["name"]))) return seen def label_from_path(path: str) -> str: parts = PurePosixPath(path).parts if "_by_supplier" in parts: index = parts.index("_by_supplier") if index + 1 < len(parts): return slug(parts[index + 1]) if len(parts) > 2: return slug(parts[1]) return "unknown" def build_plan(paths: list[str], seen: set[str], date_separator: str = ".") -> list[dict[str, str]]: plan: list[dict[str, str]] = [] for path in paths: parts = PurePosixPath(path).parts if len(parts) <= 2: continue month = parts[0] if not MONTH_RE.match(month): continue date = extract_date({"path": path, "preview": ""}, separator=date_separator) label = label_from_path(path) original = PurePosixPath(path).name.replace("/", "_").strip() filename = trim_component(f"{date}-{label}-{original}") target = str(PurePosixPath(month) / filename) base_target = target collision_index = 2 while target in seen: target = add_collision_suffix(base_target, collision_index) collision_index += 1 seen.add(target) plan.append({ "path": path, "target": target, "date": date, "label": label, "original": original, }) return plan def write_csv(path: Path, rows: list[dict[str, Any]]) -> None: fields = ["path", "target", "ok", "moved", "dryRun", "date", "label", "error"] path.parent.mkdir(parents=True, exist_ok=True) 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 flatten_tree_files(args: argparse.Namespace, client: NodeClient | None = None) -> dict[str, Any]: client = client or NodeClient(args.node_url, timeout=args.timeout) exclude_parts = set(split_words(args.exclude_parts)) months = discover_months(client, args.list_uri, args.root_path, args.months) seen = top_level_files(client, args.list_uri, months) paths: list[str] = [] for month in months: paths.extend(walk_files(client, args.list_uri, month, exclude_parts)) nested = [path for path in paths if len(PurePosixPath(path).parts) > 2] if args.max_files: nested = nested[: args.max_files] plan = build_plan(nested, seen=seen, date_separator=args.date_separator) results: list[dict[str, Any]] = [] for item in plan: envelope = client.run(args.move_uri, { "path": item["path"], "target_path": item["target"], "dry_run": not args.execute, "overwrite": args.overwrite, "make_parents": True, }) data = route_value(envelope) results.append({ **item, "ok": bool(data.get("ok")), "moved": bool(data.get("moved")), "dryRun": data.get("dryRun", not args.execute), "target": data.get("target", item["target"]), "error": data.get("error", ""), }) payload = { "ok": all(row["ok"] for row in results), "dryRun": not args.execute, "months": months, "candidate_count": len(plan), "ok_count": sum(1 for row in results if row["ok"]), "moved_count": sum(1 for row in results if row["moved"]), "results": results, } out_dir = Path(args.output_dir) out_dir.mkdir(parents=True, exist_ok=True) stem = "flat_tree_manifest" if args.execute else "flat_tree_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="Flatten remaining nested files under month folders via URI.") parser.add_argument("--node-url", default=DEFAULT_NODE_URL) parser.add_argument("--output-dir", default=".state") parser.add_argument("--root-path", default=".") parser.add_argument("--months", default="") parser.add_argument("--exclude-parts", default="no_invoice") parser.add_argument("--list-uri", default="fs://host/dir/query/list") parser.add_argument("--move-uri", default="fs://host/file/command/move") parser.add_argument("--date-separator", default=".", choices=[".", "-"]) parser.add_argument("--max-files", type=int, default=0) parser.add_argument("--timeout", type=int, default=120) parser.add_argument("--overwrite", action="store_true") parser.add_argument("--execute", action="store_true", help="actually move/rename files; default is dry-run") return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) result = flatten_tree_files(args) print(json.dumps({ "ok": result["ok"], "dryRun": result["dryRun"], "months": result["months"], "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())