Rihards Gailums
8c2a006427
Datu struktūra schemas/mk-noteikumi-0.1.xsd; rīki convert.py (vestnesis.lv un likumi.lv), build_all.py (lejuplāde, pārveide, pārbaude pret avota lapu), validate.py. Noteikumi Nr. 322 (2006), Nr. 982 (2006), Nr. 635 (2023), Nr. 821 (2023) no oficiālās publikācijas — likumi.lv slēgts, vēlākie grozījumi nav iestrādāti. Datu līgums contracts/B19-mk-noteikumi.odcs.yaml. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016679RwmHsuTFfxt26wP6rk
110 lines
4.3 KiB
Python
110 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Ielādē visus noteikumus no kataloga mk-noteikumi.yaml, pārveido un pārbauda pret avota lapu.
|
|
|
|
python3 tools/build_all.py [--cache <mape>] [--date GGGG-MM-DD] [id ...]
|
|
|
|
Katram ierakstam:
|
|
1. lejuplādē lapu (likumi.lv konsolidēto redakciju, ja source: likumi, citādi vestnesis.lv/ta/id/<id>), ar pauzi starp
|
|
pieprasījumiem; HTML glabā tikai kešatmiņā, ne repozitorijā;
|
|
2. tools/convert.py → sources/<id>.txt un data/<id>.xml;
|
|
3. verification/<id>.verify.json: lapas sha256, un vai katrs punkta, nodaļas un pielikuma teksts ir lapas tekstā
|
|
burtiski un tādā pašā secībā.
|
|
"""
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
|
|
import yaml
|
|
from lxml import etree
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import convert # noqa: E402
|
|
|
|
ROOT = convert.ROOT
|
|
NS = {"n": "urn:pppa:vpk:mk-noteikumi:0.1"}
|
|
UA = {"User-Agent": "Mozilla/5.0 (PPPA Valsts PirmKods; MK-noteikumi-as-Code)"}
|
|
|
|
|
|
def fetch(url, dest):
|
|
if os.path.exists(dest):
|
|
return open(dest, encoding="utf-8", errors="replace").read()
|
|
for attempt in range(5):
|
|
try:
|
|
with urllib.request.urlopen(urllib.request.Request(url, headers=UA), timeout=30) as r:
|
|
page = r.read().decode("utf-8", "replace")
|
|
break
|
|
except urllib.error.HTTPError as e:
|
|
if e.code != 429 or attempt == 4:
|
|
raise
|
|
time.sleep(30)
|
|
os.makedirs(os.path.dirname(dest), exist_ok=True)
|
|
with open(dest, "w", encoding="utf-8") as f:
|
|
f.write(page)
|
|
time.sleep(2)
|
|
return page
|
|
|
|
|
|
def page_text(page):
|
|
"""All visible text of the page in one normalised string (superscripts as ¹², checkboxes as ☐)."""
|
|
import lxml.html
|
|
root = lxml.html.fromstring(page)
|
|
return " ".join(convert.paras_of(root))
|
|
|
|
|
|
def verify(doc_id, page):
|
|
x = etree.parse(f"{ROOT}/data/{doc_id}.xml").getroot()
|
|
hay = re.sub(r"\s+", " ", page_text(page))
|
|
pos, n, missing = 0, 0, []
|
|
for e in x.iter("{%s}Text" % NS["n"], "{%s}Title" % NS["n"], "{%s}Line" % NS["n"], "{%s}Heading" % NS["n"]):
|
|
if etree.QName(e.getparent()).localname == "Act":
|
|
continue
|
|
# one check per line; a table row (cells joined with " | ") is checked cell by cell
|
|
parts = [c.strip() for line in (e.text or "").split("\n") for c in line.split(" | ") if c.strip()]
|
|
for part in parts:
|
|
n += 1
|
|
i = hay.find(part, pos)
|
|
if i < 0:
|
|
owner = e.getparent().get("id") or etree.QName(e.getparent()).localname
|
|
missing.append(f"{owner}: {part[:80]}")
|
|
else:
|
|
pos = i
|
|
return {"id": doc_id, "page_sha256": hashlib.sha256(page.encode("utf-8")).hexdigest(), "texts": n,
|
|
"missing": missing, "passed": not missing}
|
|
|
|
|
|
def main():
|
|
args = sys.argv[1:]
|
|
cache = args[args.index("--cache") + 1] if "--cache" in args else os.path.join(ROOT, ".cache")
|
|
date = args[args.index("--date") + 1] if "--date" in args else datetime.date.today().isoformat()
|
|
only = [a for i, a in enumerate(args) if not a.startswith("--") and (i == 0 or args[i - 1] not in ("--cache", "--date"))]
|
|
cat = yaml.safe_load(open(f"{ROOT}/mk-noteikumi.yaml", encoding="utf-8"))
|
|
os.makedirs(f"{ROOT}/verification", exist_ok=True)
|
|
bad = 0
|
|
for r in cat["noteikumi"]:
|
|
if only and r["id"] not in only:
|
|
continue
|
|
src = r.get("source", "vestnesis")
|
|
url = f"https://likumi.lv/ta/id/{r['likumi_id']}" if src == "likumi" else f"https://www.vestnesis.lv/ta/id/{r['likumi_id']}"
|
|
page = fetch(url, os.path.join(cache, f"{src}-{r['likumi_id']}.html"))
|
|
doc = convert.build(page, r["likumi_id"], date, r.get("org"))
|
|
if doc != r["id"]:
|
|
raise SystemExit(f"{r['id']}: lapā ir {doc}")
|
|
rep = verify(doc, page)
|
|
rep.update({"url": url, "checked": date})
|
|
with open(f"{ROOT}/verification/{doc}.verify.json", "w", encoding="utf-8") as f:
|
|
json.dump(rep, f, ensure_ascii=False, indent=1)
|
|
print(f"{'OK ' if rep['passed'] else 'FAIL'} {doc} teksti {rep['texts']} trūkst {len(rep['missing'])}")
|
|
for m in rep["missing"][:5]:
|
|
print(" ", m)
|
|
bad += not rep["passed"]
|
|
sys.exit(1 if bad else 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|