Strategy-as-Code/tools/verify_nap2027.py

#!/usr/bin/env python3
"""NAP2027 datnes pārbaude pret avota PDF ar citu PDF nolasītāju (poppler pdftotext), nevis to, ar kuru datne veidota (pdfplumber).

  python3 tools/verify_nap2027.py data/nap2027/lv-nap2027.xml sources/nap2027/NAP2027.pdf verification/lv-nap2027.verify.json

Pārbauda:
  1. numurētie punkti [1]–[N] — katrs PDF drukātais numurs ir datnē tieši vienreiz, un otrādi;
  2. teksta punkti — punkta burtu secība ir PDF tekstā (bez atstarpēm, pieturzīmēm un cipariem; lapu galvenes izņemtas);
  3. tabulu rindas (indikatori, uzdevumi) — katrs datnes vārds atrodams PDF teksta fragmentā starp [n] un [n+1]
     (vārds vai tā daļas, ja PDF to pārnesis citā rindā);
  4. pamatojuma punkti — burtu secība ir PDF tekstā;
  5. uzdevumi — katram ir atbildīgā institūcija ar VPK ID; indikatori — nosaukums un vismaz viena vērtība.
"""
import hashlib
import json
import re
import subprocess
import sys

from lxml import etree

NS = {"s": "urn:pppa:vpk:strategija:0.1"}
HEADER = "latvijasnacionālaisattīstībasplānsgadam"
BOILER = set()
SUBS = str.maketrans("₀₁₂₃₄₅₆₇₈₉", "0123456789")


def letters(s):
    return re.sub(r"[^a-zāčēģīķļņšūž]", "", (s or "").lower())


def words(s):
    return [w for w in re.findall(r"[0-9a-zāčēģīķļņšūž]+", (s or "").lower().translate(SUBS))]


def runs(t, flat, min_run=8):
    """Number of contiguous pieces of t found in flat (PDF reading order may interleave footnotes or table cells);
    None if some piece shorter than min_run is not found."""
    n, i = 0, 0
    while i < len(t):
        lo, hi = 0, len(t) - i
        while lo < hi:
            m = (lo + hi + 1) // 2
            if t[i:i + m] in flat:
                lo = m
            else:
                hi = m - 1
        if lo < min(min_run, len(t) - i):
            return None
        n, i = n + 1, i + lo
    return n


def main(xml_path, pdf_path, out):
    raw = subprocess.run(["pdftotext", pdf_path, "-"], capture_output=True, text=True, check=True).stdout
    lay = subprocess.run(["pdftotext", "-layout", pdf_path, "-"], capture_output=True, text=True, check=True).stdout
    flat = letters(raw).replace(HEADER, "")
    doc = etree.parse(xml_path)
    # words that may stand between table rows in the PDF and belong elsewhere in the file: footnotes, sub-section titles
    BOILER.update(words(" ".join(x.text for x in doc.findall(".//s:Footnote", NS))))
    BOILER.update(words(" ".join(x.text for x in doc.findall(".//s:Section/s:Title", NS))))
    items = doc.findall(".//s:Item", NS)
    res = {"file": xml_path, "pdf_sha256": hashlib.sha256(open(pdf_path, "rb").read()).hexdigest(), "checks": {}}

    # 1. numbering
    printed = sorted({int(m) for m in re.findall(r"\[(\d{1,3})\]", lay)})
    have = [int(i.get("n")) for i in items]
    dup = sorted({n for n in have if have.count(n) > 1})
    res["checks"]["numbering"] = {"printed": len(printed), "in_file": len(have), "missing": sorted(set(printed) - set(have)),
                                  "extra": sorted(set(have) - set(printed)), "duplicates": dup}

    # layout blocks between consecutive markers
    pos = {int(m.group(1)): m.end() for m in re.finditer(r"\[(\d{1,3})\]", lay)}
    start = {int(m.group(1)): m.start() for m in re.finditer(r"\[(\d{1,3})\]", lay)}
    order = sorted(pos)

    def block(n):
        k = order.index(n)
        end = start[order[k + 1]] if k + 1 < len(order) else len(lay)
        return lay[pos[n]:end]

    text_bad, text_split, row_bad, row_extra, row_words, text_n, row_n = [], [], [], [], 0, 0, 0

    def block_rows(n):
        """The row's lines only: stop at a blank line followed by body text, drop page header/footer lines."""
        out = []
        for ln in block(n).split("\n"):
            st = ln.strip()
            if not st or re.fullmatch(r"\d{1,3}", st) or st.startswith("Latvijas Nacionālais attīstības plāns"):
                continue
            if re.match(r"^(Rīcības virziena|RĪCĪBAS|PRIORITĀTES|Prioritāte|NAP2027|Nr\.|\*|\d{1,2} [A-ZĀČĒĢĪĶĻŅŠŪŽa-z])", st) or len(ln) - len(ln.lstrip()) == 0 and len(st) > 60:
                break
            out.append(ln)
        return "\n".join(out)
    for it in items:
        n, kind = int(it.get("n")), it.get("kind")
        if kind in ("indicator", "task"):
            row_n += 1
            toks = set(words(block(n)))
            # footnote numbers printed after a word or number ("piesaisti10", "4977"): also try without them
            toks |= {re.sub(r"\d{1,2}$", "", w) for w in toks if re.search(r"[a-zāčēģīķļņšūž]\d{1,2}$", w)}
            toks |= {w[:-k] for w in toks if w.isdigit() and len(w) > 2 for k in (1, 2)}
            vals = [x.text for x in it.iter() if x.text and x.text.strip() and etree.QName(x).localname not in ("Item",)]
            vals += [a.get("printed") for a in it.iter() if a.get("printed")]
            missing = []
            for w in words(" ".join(vals)):
                row_words += 1
                if w in toks:
                    continue
                if any(w[:k] in toks and w[k:] in toks for k in range(1, len(w))):
                    continue
                if any(w[:k] in toks and w[k:j] in toks and w[j:] in toks for k in range(1, len(w)) for j in range(k + 1, len(w))):
                    continue
                missing.append(w)
            if missing:
                row_bad.append({"n": n, "missing_words": missing[:10]})
            # reverse: every word printed in the row's part of the PDF is in the file (nothing dropped)
            xw = set(words(" ".join(vals)))
            xjoin = " ".join(sorted(xw))
            extra = []
            for w in words(block_rows(n)):
                w2 = re.sub(r"(?<=[a-zāčēģīķļņšūž])\d{1,2}$", "", w)  # footnote number printed after a word
                if w in xw or w2 in xw or w in BOILER or w2 in BOILER or (w.isdigit() or len(w) >= 2) and w in xjoin:
                    continue
                if w.isdigit() and len(w) > 2 and (w[:-1] in xw or w[:-2] in xw):  # number followed by a footnote number
                    continue
                extra.append(w)
            if extra:
                row_extra.append({"n": n, "words_not_in_file": extra[:12]})
        else:
            text_n += 1
            ks = [runs(letters(x.text), flat) for x in it.findall("s:Text", NS) + it.findall("s:Area", NS)]
            k = None if None in ks else max(ks)
            if k is None or k > 4:
                text_bad.append(n)
            elif k > 1:
                text_split.append(n)
    res["checks"]["text_items"] = {"checked": text_n, "not_found_verbatim": text_bad,
                                   "found_in_2_to_4_pieces": text_split}
    res["checks"]["table_rows"] = {"checked": row_n, "words": row_words, "rows_with_missing_words": row_bad,
                                   "rows_with_words_not_in_file": row_extra}

    ev = doc.findall(".//s:Evidence", NS)
    ev_bad = [e.get("id") for e in ev if (runs(letters((e.findtext("s:Problem", namespaces=NS) or "") + e.findtext("s:Text", namespaces=NS)), flat) or 99) > 4]
    res["checks"]["evidence"] = {"checked": len(ev), "numbers_continuous": [int(e.get("n")) for e in ev if e.get("n")] == list(range(1, len([e for e in ev if e.get("n")]) + 1)),
                                 "not_found_verbatim": ev_bad}

    tasks = doc.findall(".//s:Task", NS)
    no_resp = [t.getparent().get("n") for t in tasks if not t.findall("s:Responsible/s:Actor[@org]", NS)]
    inds = doc.findall(".//s:Indicator", NS)
    no_val = [i.getparent().get("n") for i in inds if not (i.findtext("s:BaseValue", namespaces=NS) or i.findall("s:Target", NS))]
    res["checks"]["completeness"] = {"tasks": len(tasks), "tasks_without_responsible_vpk_id": no_resp,
                                     "indicators": len(inds), "indicators_without_values": no_val}
    c = res["checks"]
    res["passed"] = not (c["numbering"]["missing"] or c["numbering"]["extra"] or c["numbering"]["duplicates"] or text_bad
                         or row_bad or row_extra or ev_bad or no_resp)
    with open(out, "w", encoding="utf-8") as f:
        json.dump(res, f, ensure_ascii=False, indent=1)
    print(json.dumps({k: {kk: (vv if not isinstance(vv, list) else (len(vv) if len(vv) > 12 else vv)) for kk, vv in v.items()}
                      for k, v in c.items()}, ensure_ascii=False, indent=1))
    print("passed", res["passed"])


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