gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
71 lines
3.0 KiB
Python
71 lines
3.0 KiB
Python
"""Import operator fiscal context (CSV or JSON) into the Hub — the answers to docs/LOCAL_FISCAL_CONTEXT_QUESTIONS_V1.csv.
|
|
|
|
python scripts/import_fiscal_context.py <file.csv|file.json> --token <AGENT_TOKEN> [--hub http://localhost:8000] [--dry-run]
|
|
|
|
CSV columns (the questionnaire CSV works as-is): company_key, scope, scope_key, key, value, reason (rows with an empty `value` are skipped).
|
|
JSON: {"company_key": "...", "entries": [{"scope","scope_key","key","value","reason"}]} or a list of rows that carry company_key.
|
|
Entries are validated locally (same rules as the Hub) and sent per company; the Hub applies each batch all-or-nothing.
|
|
"""
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
from central.app.context_keys import validate_entry # noqa: E402
|
|
|
|
|
|
def load_rows(path):
|
|
p = Path(path)
|
|
if p.suffix.lower() == ".json":
|
|
data = json.loads(p.read_text(encoding="utf-8"))
|
|
if isinstance(data, dict):
|
|
return [{"company_key": data.get("company_key"), **e} for e in data.get("entries", [])]
|
|
return data
|
|
with p.open(newline="", encoding="utf-8-sig") as f:
|
|
return list(csv.DictReader(f))
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("file"); ap.add_argument("--token", required=True); ap.add_argument("--hub", default="http://localhost:8000"); ap.add_argument("--dry-run", action="store_true")
|
|
a = ap.parse_args(argv)
|
|
by_company, errors, skipped = defaultdict(list), [], 0
|
|
for i, row in enumerate(load_rows(a.file), start=1):
|
|
if not str(row.get("value") or "").strip():
|
|
skipped += 1
|
|
continue
|
|
entry, err = validate_entry(row)
|
|
if err:
|
|
errors.append((i, err))
|
|
else:
|
|
by_company[str(row.get("company_key") or "").strip()].append(entry)
|
|
if errors:
|
|
for i, e in errors[:30]:
|
|
print(f"row {i}: {e}")
|
|
raise SystemExit(f"{len(errors)} invalid row(s); nothing was sent")
|
|
total = 0
|
|
for company, entries in by_company.items():
|
|
if not company:
|
|
raise SystemExit("company_key is required on every row")
|
|
total += len(entries)
|
|
if a.dry_run:
|
|
continue
|
|
req = urllib.request.Request(f"{a.hub}/v1/context/import", data=json.dumps({"company_key": company, "entries": entries}).encode(), method="POST",
|
|
headers={"Authorization": f"Bearer {a.token}", "Content-Type": "application/json"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=60) as r:
|
|
print(company, json.load(r))
|
|
except urllib.error.HTTPError as e:
|
|
raise SystemExit(f"Hub refused {company}: {e.read().decode()[:300]}")
|
|
print(f"{'validated' if a.dry_run else 'imported'} {total} entries ({skipped} unanswered rows skipped)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|