Bernt
7d40cf95bf
- FastAPI Identify API (port 8081) - Neo4j propertygraf (64 noder, 80 kanter) - Ingest-pipeline (Python) - Byggscript + integrationstester - Docker Compose setup
63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Exportera master-data till CSV för Neo4j-import.
|
|
"""
|
|
import json
|
|
import csv
|
|
from pathlib import Path
|
|
|
|
DATA = Path(__file__).parent.parent / "10-data"
|
|
DB = Path(__file__).parent.parent / "db"
|
|
|
|
def main():
|
|
poster = {}
|
|
for fn in ("lvx-klassposter-master-v3.json", "lvx-produktmodeller-vag2.json",
|
|
"lvx-produktmodeller-vag3.json", "lvx-produktmodeller-auto.json"):
|
|
fp = DATA / fn
|
|
if not fp.exists(): continue
|
|
with open(fp, encoding="utf-8") as f:
|
|
for p in json.load(f).get("poster", []):
|
|
poster[p["lvx_id"]] = p
|
|
|
|
# Noder
|
|
with open(DB / "noder.csv", "w", newline="", encoding="utf-8") as f:
|
|
w = csv.DictWriter(f, fieldnames=["lvx_id","posttyp","namn_sv","namn_en","doman","tier","verifieringsniva","konfidens"])
|
|
w.writeheader()
|
|
for p in poster.values():
|
|
prov = p.get("proveniens", {})
|
|
w.writerow({
|
|
"lvx_id": p["lvx_id"],
|
|
"posttyp": p.get("posttyp", ""),
|
|
"namn_sv": p.get("namn", {}).get("sv", ""),
|
|
"namn_en": p.get("namn", {}).get("en", ""),
|
|
"doman": p.get("doman", ""),
|
|
"tier": p.get("tier", ""),
|
|
"verifieringsniva": prov.get("verifieringsniva", "obekraftad"),
|
|
"konfidens": prov.get("konfidens", 0),
|
|
})
|
|
|
|
# Kanter
|
|
with open(DATA / "lvx-graf-kanter-v3.json", encoding="utf-8") as f:
|
|
kanter = json.load(f)["kanter"]
|
|
|
|
with open(DB / "kanter.csv", "w", newline="", encoding="utf-8") as f:
|
|
w = csv.DictWriter(f, fieldnames=["subjekt","predikat","objekt","niva","beskrivning"])
|
|
w.writeheader()
|
|
for k in kanter:
|
|
if k["objekt_typ"] != "lvx":
|
|
continue # hoppa över textnoder i grafen
|
|
w.writerow({
|
|
"subjekt": k["subjekt"],
|
|
"predikat": k["predikat"],
|
|
"objekt": k["objekt"],
|
|
"niva": k.get("niva", ""),
|
|
"beskrivning": k.get("beskrivning", ""),
|
|
})
|
|
|
|
print(f"Exporterat: {len(poster)} noder, {len(kanter)} kanter")
|
|
print(f" → {DB}/noder.csv")
|
|
print(f" → {DB}/kanter.csv")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|