Nolikumi-as-Code/tools/cites.py

"""cites.py — atsauces uz tiesību aktiem nolikumā (Cite) no likumi.lv lapas saitēm.

Katrs nolikuma punkts (un „Izdoti saskaņā ar …”) saņem <Cite> katram tiesību aktam, uz kuru likumi.lv lapā ir saite
šī punkta tekstā. Neņem: saites uz grozījumu noteikumiem (redakcijas piezīmes, „svītrots ar …”), saites uz pašu nolikumu.

  python3 tools/cites.py <kešs> [<veidi.csv>]

<kešs>/<likumi_id>.html — nolikuma lapa; trūkstošās lapas (nolikumi un citētie akti) tiek lejupielādētas.
<veidi.csv> (nav obligāts) — jau zināmi citēto aktu nosaukumi un veidi (kolonnas likumi_id, nosaukums, veids, statuss),
lai nav jālejupielādē katra citētā akta lapa.
Ja atsauces failā nemainās, fails netiek pārrakstīts; ja mainās — pievieno DataVersion ierakstu.
"""
import copy, csv, datetime, html as H, os, re, sys, time, urllib.request
import yaml
import lxml.html
from lxml import etree

NS = "urn:pppa:vpk:nolikums:0.1"
Q = "{%s}" % NS
UA = {"User-Agent": "Mozilla/5.0 (nolikumi-as-code/0.2 cites)"}
SUP = str.maketrans("0123456789", "⁰¹²³⁴⁵⁶⁷⁸⁹")
DATE = re.compile(r"^\d{2}\.\d{2}\.\d{4}\.?$")
NUM = re.compile(r"^(\d+(?:\.\d+)*)\.?([⁰¹²³⁴⁵⁶⁷⁸⁹]*)\.?\s")  # 4.22.² / 4.22² → 4.22²
LINK = re.compile(r"^(?:https?://likumi\.lv)?/ta/id/(\d+)(?:-([^#?]*))?(?:#(\w+))?")


def get(lid, cache):
    p = os.path.join(cache, f"{lid}.html")
    if not os.path.exists(p) or os.path.getsize(p) == 0:
        time.sleep(1)
        with urllib.request.urlopen(urllib.request.Request(f"https://likumi.lv/ta/id/{lid}", headers=UA), timeout=60) as r:
            open(p, "wb").write(r.read())
    return open(p, encoding="utf-8").read()


def pase(page):
    i = page.find("pase-container")
    t = " ".join(H.unescape(re.sub(r"<[^>]+>", " ", re.sub(r"(?s)<(script|style)\b.*?</\1>", " ", page[i:i + 20000]))).split())
    f = {}
    for k in ["Nosaukums", "Statuss", "Veids"]:
        m = re.search(re.escape(k) + r":\s*(.*?)\s+(?=Statuss:|Izdevējs:|Veids:|Pieņemts:|Stājas spēkā:|Tēma:|Zaudē spēku:|Publicēts:|Numurs:)", t)
        if m:
            f[k] = m.group(1).strip()
    if "Statuss" in f:
        w = f["Statuss"].split(); h = len(w) // 2
        if len(w) % 2 == 0 and [x.lower() for x in w[:h]] == [x.lower() for x in w[h:]]:
            f["Statuss"] = " ".join(w[:h])
    return f


def anchor_unit(a):
    """'p16' → pants 16; 'p16_1' → 16¹; 'piel2' → pielikums 2; 'n3' → nodaļa 3."""
    if not a:
        return None, None
    m = re.fullmatch(r"p(\d+)(?:_(\d+))?", a)
    if m:
        return "article", m.group(1) + (m.group(2) or "").translate(SUP)
    m = re.fullmatch(r"piel(\d+)", a)
    if m:
        return "annex", m.group(1)
    m = re.fullmatch(r"n(\d+)", a)
    if m:
        return "chapter", m.group(1)
    return None, None


def links_in(p, self_id):
    """Saites rindkopā p: [(likumi_id, enkurs, saites teksts)], bez grozījumu un paššaitēm."""
    out = []
    for a in p.iter("a"):
        m = LINK.match(a.get("href") or "")
        if not m:
            continue
        lid, slug, anc = m.group(1), m.group(2) or "", m.group(3)
        txt = " ".join(a.text_content().split())
        if lid == str(self_id) or slug.startswith("grozij") or DATE.match(txt) or not txt:
            continue
        if out and out[-1][0] == lid and (not out[-1][1] or not anc):
            # viena atsauce, vairākas saites: "Valsts pārvaldes iekārtas likuma" + "16.panta"; "Latvijas" + "vides … likumā"
            out[-1] = (lid, out[-1][1] or anc, out[-1][2] + " " + txt)
            continue
        out.append((lid, anc, txt))
    return out


def html_cites(page, self_id):
    """→ (legal_basis_cites, {punkta n: [cites]})"""
    r = lxml.html.fromstring(page)
    body = r.xpath("//div[contains(concat(' ', normalize-space(@class), ' '), ' TV207 ')]")[0].getparent()
    basis, by_n = [], {}
    for el in body:
        if not isinstance(el.tag, str):
            continue
        c = ((el.get("class") or "").split() or [""])[0]
        if c == "TV900":
            basis += links_in(el, self_id)
            continue
        if c != "TV213":
            continue
        top = next((x.get("name")[1:] for x in el.iter("a") if re.fullmatch(r"p\d+", x.get("name") or "")), None)
        cur = top
        for p in el.iter("p"):
            if "labojumu_pamats" in (p.get("class") or ""):
                continue
            q = copy.deepcopy(p)
            for sp in q.iter("sup"):
                sp.text = (sp.text or "").translate(SUP)
            t = " ".join(q.text_content().split())
            m = NUM.match(t)
            if m:
                cur = m.group(1) + m.group(2)
            ls = links_in(p, self_id)
            if ls and cur:
                by_n.setdefault(cur, []).extend(ls)
    return basis, by_n


def cite_el(lid, anc, txt, acts):
    a = acts.get(lid, {})
    e = etree.Element(Q + "Cite", likumiId=lid)
    if a.get("veids"): e.set("type", a["veids"])
    if a.get("nosaukums"): e.set("title", a["nosaukums"])
    if a.get("statuss"): e.set("status", a["statuss"])
    unit, n = anchor_unit(anc)
    if unit: e.set(unit, n)
    e.text = txt
    return e


def sig(root):
    return [(x.getparent().get("id"), x.get("likumiId"), x.get("article"), x.text) for x in root.iter(Q + "Cite")]


def place(parent, cites, acts):
    """Cite pēc Text/Note/LegalBasis, pirms apakšpunktiem / nodaļām."""
    for x in parent.findall(Q + "Cite"):
        parent.remove(x)
    kids = list(parent)
    pos = next((k for k, x in enumerate(kids) if x.tag in (Q + "Point", Q + "Chapter")), len(kids))
    for k, (lid, anc, txt) in enumerate(cites):
        parent.insert(pos + k, cite_el(lid, anc, txt, acts))


def main(cache, kinds=None):
    os.makedirs(cache, exist_ok=True)
    cat = yaml.safe_load(open("nolikumi.yaml", encoding="utf-8"))["nolikumi"]
    acts = {}
    if kinds:
        for r in csv.DictReader(open(kinds, encoding="utf-8")):
            acts[r["likumi_id"]] = {"nosaukums": r.get("nosaukums"), "veids": r.get("veids"), "statuss": r.get("statuss")}
    today = datetime.date.today().isoformat()
    tot = {"files": 0, "changed": 0, "cites": 0, "unmatched": 0}
    for e in cat:
        page = get(e["likumi_id"], cache)
        basis, by_n = html_cites(page, e["likumi_id"])
        for lid in {x[0] for x in basis} | {x[0] for v in by_n.values() for x in v}:
            if lid not in acts or not acts[lid].get("veids"):
                p = pase(get(lid, cache))
                acts[lid] = {"nosaukums": p.get("Nosaukums"), "veids": p.get("Veids"), "statuss": p.get("Statuss")}
        path = e["file"]
        tree = etree.parse(path)
        root = tree.getroot()
        before = sig(root)
        if root.find(Q + "LegalBasis") is not None:
            place(root, basis, acts)
        pts = {p.get("n"): p for p in root.iter(Q + "Point")}
        for p in pts.values():
            place(p, [], acts)
        for n, cs in by_n.items():
            if n in pts:
                place(pts[n], cs, acts)
            else:
                tot["unmatched"] += 1
                print(f"  {e['org']}: punkts {n} nav XML ({len(cs)} saites)")
        after = sig(root)
        tot["files"] += 1; tot["cites"] += len(after)
        if after != before:
            nums = [int(d.get("number")) for d in root.findall(Q + "DataVersion")]
            dv = etree.SubElement(root, Q + "DataVersion", number=str(max(nums) + 1), date=today)
            etree.SubElement(dv, Q + "Change").text = ("Atsauces uz tiesību aktiem (Cite) pēc likumi.lv lapas saitēm: "
                                                       f"{len(after)}.")
            etree.indent(root, space="  ")
            data = etree.tostring(root, xml_declaration=True, encoding="UTF-8")
            open(path, "wb").write(data if data.endswith(b"\n") else data + b"\n")
            tot["changed"] += 1
    print(tot)


if __name__ == "__main__":
    main(*sys.argv[1:])