Strategy-as-Code/tools/validate.py

#!/usr/bin/env python3
"""Strategy-as-Code pārbaudītājs.

  python3 tools/validate.py [--registry organizacijas.xml]

Katram dokumentam katalogā planosanas-dokumenti.yaml:
  - datne atbilst XSD shēmai schemas/strategija-0.1.xsd (arī atslēgas un atsauces shēmā);
  - datnes nosaukums = dokumenta identifikators;
  - avota PDF sha256 sakrīt ar Metadata/Source/SHA256;
  - visi VPK ID (Actors, Responsible, CoResponsible, Abbreviation, Developer) ir Valsts institūciju reģistrā;
  - visi pašvaldību ATVK kodi (atribūts atvk) ir Pašvaldību reģistrā (Pasvaldibu-registrs);
  - pārbaudes atskaite verification/<id>.verify.json ir par šo pašu avota PDF un ir izturēta.
Reģistrs pēc noklusējuma — ProcessGit Organizations_register organizacijas.xml;
pašvaldību reģistrs (--pasvaldibas) — ProcessGit Pasvaldibu-registrs data/pasvaldibas-atvk.xml.
"""
import hashlib
import json
import os
import sys
import urllib.request

import yaml
from lxml import etree

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
NS = {"s": "urn:pppa:vpk:strategija:0.1"}
REG_URL = "https://processgit.org/Valsts-Pirmkods/Organizations_register/raw/branch/main/organizacijas.xml"
PAS_URL = "https://processgit.org/Valsts-Pirmkods/Pasvaldibu-registrs/raw/branch/main/data/pasvaldibas-atvk.xml"


def main():
    reg_src = sys.argv[sys.argv.index("--registry") + 1] if "--registry" in sys.argv else REG_URL
    reg = etree.parse(reg_src if os.path.exists(reg_src) else
                      urllib.request.urlopen(urllib.request.Request(reg_src, headers={"User-Agent": "strategy-as-code-validate/0.1"})))
    ids = {o.get("id") for o in reg.getroot() if etree.QName(o).localname == "Organization"}
    pas_src = sys.argv[sys.argv.index("--pasvaldibas") + 1] if "--pasvaldibas" in sys.argv else PAS_URL
    pas = etree.parse(pas_src if os.path.exists(pas_src) else urllib.request.urlopen(urllib.request.Request(pas_src, headers={"User-Agent": "valsts-pirmkods-validate/0.1"})))
    atvk_ids = {m.get("atvk") for m in pas.getroot() if etree.QName(m).localname == "Municipality"}
    xsd = etree.XMLSchema(etree.parse(os.path.join(ROOT, "schemas/strategija-0.1.xsd")))
    cat = yaml.safe_load(open(os.path.join(ROOT, "planosanas-dokumenti.yaml"), encoding="utf-8"))
    errors, n = [], 0
    for d in cat["documents"]:
        if not d.get("as_code", {}).get("path"):
            continue
        n += 1
        path = os.path.join(ROOT, d["as_code"]["path"])
        doc = etree.parse(path)
        if not xsd.validate(doc):
            errors += [f"{d['id']}: XSD {e.line}: {e.message}" for e in list(xsd.error_log)[:10]]
        root = doc.getroot()
        if root.get("id") != d["id"] or os.path.basename(path) != d["id"] + ".xml":
            errors.append(f"{d['id']}: datnes nosaukums vai id neatbilst katalogam")
        src = os.path.join(ROOT, root.findtext("s:Metadata/s:Source/s:File", namespaces=NS))
        sha = hashlib.sha256(open(src, "rb").read()).hexdigest()
        if sha != root.findtext("s:Metadata/s:Source/s:SHA256", namespaces=NS):
            errors.append(f"{d['id']}: avota sha256 nesakrīt")
        used = {e.get(k) for e in root.iter() for k in ("org", "currentOrg") if e.get(k)}
        missing = sorted(used - ids)
        if missing:
            errors.append(f"{d['id']}: VPK ID nav reģistrā: {', '.join(missing)}")
        bad_atvk = sorted({e.get("atvk") for e in root.iter() if e.get("atvk")} - atvk_ids)
        if bad_atvk:
            errors.append(f"{d['id']}: ATVK kods nav pašvaldību reģistrā: {', '.join(bad_atvk)}")
        vpath = os.path.join(ROOT, "verification", d["id"] + ".verify.json")
        if os.path.exists(vpath):
            v = json.load(open(vpath, encoding="utf-8"))
            if v.get("pdf_sha256") != sha or not v.get("passed"):
                errors.append(f"{d['id']}: pārbaudes atskaite nav izturēta vai ir par citu avota datni")
        else:
            errors.append(f"{d['id']}: nav pārbaudes atskaites {vpath}")
        print(f"{d['id']}: {len(doc.findall('.//s:Item', NS))} punkti, {len(used)} VPK ID")
    for e in errors:
        print("KĻŪDA", e)
    print("pārbaudīti dokumenti:", n, "kļūdas:", len(errors))
    sys.exit(1 if errors else 0)


if __name__ == "__main__":
    main()