Files
fn_registry/python/functions/infra/vault_profile_dispatch.py
T
egutierrez a802f59f55 chore: auto-commit (95 archivos)
- cmd/fn/doctor.go
- cmd/fn/main.go
- cpp/apps/primitives_gallery/playground/tables/CMakeLists.txt
- cpp/apps/primitives_gallery/playground/tables/data_table.cpp
- cpp/apps/primitives_gallery/playground/tables/data_table_logic.cpp
- cpp/apps/primitives_gallery/playground/tables/data_table_logic.h
- cpp/apps/primitives_gallery/playground/tables/self_test.cpp
- cpp/apps/primitives_gallery/playground/tables/tql.cpp
- cpp/apps/primitives_gallery/playground/tables/viz.cpp
- cpp/apps/primitives_gallery/playground/tables/viz.h
- ...

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 00:50:34 +02:00

93 lines
3.0 KiB
Python

"""vault_profile_dispatch — CLI dispatcher that routes a single vault file to the right profiler.
Usage:
python3 vault_profile_dispatch.py --vault <path> --rel-path <p> --kind csv|pdf|md [--db-path <p>]
Exit codes:
0 success (result printed as JSON)
1 missing required argument
2 unknown kind
3 profiler raised an error
"""
from __future__ import annotations
import argparse
import json
import sys
import os
from pathlib import Path
def _python_path_setup() -> None:
"""Ensure the registry python/functions directory is on sys.path."""
# Try FN_REGISTRY_ROOT env first, then walk up from this file's location.
registry_root = os.environ.get("FN_REGISTRY_ROOT", "")
if not registry_root:
# This file lives at python/functions/infra/vault_profile_dispatch.py
# So the registry root is four levels up from __file__.
candidate = Path(__file__).resolve().parent.parent.parent.parent
if (candidate / "go.mod").exists():
registry_root = str(candidate)
if registry_root:
fn_path = str(Path(registry_root) / "python" / "functions")
if fn_path not in sys.path:
sys.path.insert(0, fn_path)
def dispatch(vault_path: str, rel_path: str, kind: str, db_path: str | None) -> dict:
"""Call the appropriate profiler based on kind."""
if kind == "csv":
from datascience.vault_csv_profile import vault_csv_profile
return vault_csv_profile(vault_path, rel_path, db_path)
elif kind == "pdf":
from datascience.vault_pdf_extract import vault_pdf_extract
return vault_pdf_extract(vault_path, rel_path, db_path)
elif kind == "md":
from infra.vault_knowledge_parse import vault_knowledge_parse
return vault_knowledge_parse(vault_path, rel_path, db_path)
else:
raise ValueError(f"unknown kind: {kind!r} (expected csv, pdf, or md)")
def main(argv: list[str] | None = None) -> int:
_python_path_setup()
parser = argparse.ArgumentParser(
prog="vault_profile_dispatch",
description="Route a single vault file to the right profiler (csv/pdf/md).",
)
parser.add_argument("--vault", required=True, help="Absolute path to vault root")
parser.add_argument("--rel-path", required=True, dest="rel_path", help="Relative path of file inside vault")
parser.add_argument(
"--kind",
required=True,
choices=["csv", "pdf", "md"],
help="Profiler kind: csv | pdf | md",
)
parser.add_argument(
"--db-path",
dest="db_path",
default=None,
help="Override path to vault_index.db (default: <vault>/vault_index.db)",
)
args = parser.parse_args(argv)
try:
result = dispatch(args.vault, args.rel_path, args.kind, args.db_path)
except ValueError as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
except Exception as exc:
print(f"error: {exc}", file=sys.stderr)
return 3
print(json.dumps(result, indent=2, default=str))
return 0
if __name__ == "__main__":
sys.exit(main())