Rihards
69932849a5
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016679RwmHsuTFfxt26wP6rk
79 lines
3.3 KiB
Python
79 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Pašvaldību reģistra lasītājs.
|
|
|
|
python3 tools/reader.py # visas pašvaldības, tabula
|
|
python3 tools/reader.py 0038001 # viena pašvaldība pēc ATVK koda (arī iepriekšējā koda)
|
|
python3 tools/reader.py --json | --csv # cits izvades formāts
|
|
python3 tools/reader.py --remote # lasīt no ProcessGit (main zars), nevis no lokālās kopijas
|
|
python3 tools/reader.py --no-validate # nepārbaudīt pret shēmu
|
|
|
|
Pēc noklusējuma datne tiek pārbaudīta pret shēmu schemas/pasvaldibu-registrs-0.1.xsd. Ar --remote datne un shēma
|
|
tiek lejupielādētas no https://processgit.org/Valsts-Pirmkods/Pasvaldibu-registrs. Vajadzīga bibliotēka lxml.
|
|
"""
|
|
import csv
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.request
|
|
|
|
from lxml import etree
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
RAW = "https://processgit.org/Valsts-Pirmkods/Pasvaldibu-registrs/raw/branch/main/"
|
|
DATA = "data/pasvaldibas-atvk.xml"
|
|
SCHEMA = "schemas/pasvaldibu-registrs-0.1.xsd"
|
|
NS = {"p": "urn:pppa:vpk:pasvaldibas:0.1"}
|
|
FIELDS = ["atvk", "nosaukums", "tips", "iepriekšējie_kodi"]
|
|
|
|
|
|
def load(rel, remote):
|
|
if remote:
|
|
req = urllib.request.Request(RAW + rel, headers={"User-Agent": "pasvaldibu-registrs-reader/0.1"})
|
|
return etree.parse(io.BytesIO(urllib.request.urlopen(req, timeout=60).read()))
|
|
return etree.parse(os.path.join(ROOT, rel))
|
|
|
|
|
|
def read(doc):
|
|
rows = [{"atvk": m.get("atvk"), "nosaukums": m.findtext("p:Name", namespaces=NS), "tips": m.get("kind"),
|
|
"iepriekšējie_kodi": " ".join(x.text for x in m.findall("p:FormerCode", NS))}
|
|
for m in doc.getroot().findall("p:Municipality", NS)]
|
|
src = doc.getroot().find("p:Metadata/p:Source", NS)
|
|
meta = {"avots": src.findtext("p:Act", namespaces=NS), "speka_no": src.get("inForce"),
|
|
"datu_versija": doc.getroot().find("p:Metadata/p:DataVersion", NS).get("number")}
|
|
return rows, meta
|
|
|
|
|
|
def main(argv):
|
|
flags = {a for a in argv if a.startswith("--")}
|
|
keys = [a for a in argv if not a.startswith("--")]
|
|
remote = "--remote" in flags
|
|
doc = load(DATA, remote)
|
|
if "--no-validate" not in flags:
|
|
xsd = etree.XMLSchema(load(SCHEMA, remote))
|
|
if not xsd.validate(doc):
|
|
for e in list(xsd.error_log)[:10]:
|
|
print(f"KĻŪDA {e.line}: {e.message}", file=sys.stderr)
|
|
sys.exit(1)
|
|
rows, meta = read(doc)
|
|
if keys:
|
|
rows = [r for r in rows if r["atvk"] in keys or set(r["iepriekšējie_kodi"].split()) & set(keys)]
|
|
if not rows:
|
|
print("nav atrasts:", ", ".join(keys), file=sys.stderr)
|
|
sys.exit(2)
|
|
if "--json" in flags:
|
|
print(json.dumps({**meta, "pasvaldibas": rows}, ensure_ascii=False, indent=2))
|
|
elif "--csv" in flags:
|
|
w = csv.DictWriter(sys.stdout, fieldnames=FIELDS)
|
|
w.writeheader()
|
|
w.writerows(rows)
|
|
else:
|
|
for r in rows:
|
|
prev = f" (iepriekš {r['iepriekšējie_kodi']})" if r["iepriekšējie_kodi"] else ""
|
|
print(f"{r['atvk']} {r['tips']:13} {r['nosaukums']}{prev}")
|
|
print(f"spēkā no {meta['speka_no']}, datu versija {meta['datu_versija']}: {len(rows)} pašvaldības", file=sys.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main(sys.argv[1:])
|