MK-Lemumi-as-Code/tools/tap.py

"""TAP portāla (tapportals.mk.gov.lv) lapu nolasīšana: Ministru kabineta sēdes darba kārtība un dokumentu teksts.

TAP nepublicē sēžu darba kārtību un lēmumu tekstu mašīnlasāmā formā. Darba kārtība ir HTML lapā /meetings/cabinet_ministers/<id>,
dokumenta teksts — HTML lapā /structuralizer/data/nodes/<id>/preview, kur katrs numurētais punkts ir atsevišķs elements.
Šis modulis no šīm lapām nolasa tikai to, kas tajās ir: neko nepapildina un neizdomā.
"""
import re
import urllib.request

from lxml import html as H

BASE = "https://tapportals.mk.gov.lv"
UA = {"User-Agent": "Mozilla/5.0 (compatible; valsts-pirmkods/0.1; +https://processgit.org/Valsts-Pirmkods/MK-Lemumi-as-Code)"}
# darba kārtības dokumenti, kas ir lēmums (ne normatīvā akta teksts vai ziņojums)
DECISION_DOCS = ("MK sēdes protokollēmuma projekts", "MK sēdes protokollēmums", "Rīkojuma projekts", "Rīkojuma (grozījumu) projekts",
                 "Rīkojums")
FLAGS = {"iekļauts papildus": "addedLate", "jautājums tiks skatīts A sadaļā": "movedToA",
         "jautājuma izskatīšana atlikta": "postponed"}


def get(url, accept=None):
    h = dict(UA)
    if accept:
        h["Accept"] = accept
    return urllib.request.urlopen(urllib.request.Request(url, headers=h), timeout=60).read()


def norm(s):
    return re.sub(r"\s+", " ", (s or "").replace("\xa0", " ")).strip()


def text(e):
    return norm(e.text_content()) if e is not None else ""


# ------------------------------------------------------------------ sēdes darba kārtība
def parse_meeting(raw):
    """-> {'title', 'sections': [{'n', 'title', 'items': [...]}]}; item: n, tap, restricted, flags, title, presenter,
    invited, status, legal_act (TAP uuid), documents [(label, href)], approvals [label], participation."""
    doc = H.fromstring(raw)
    title = text((doc.xpath("//h1") or [None])[0])
    sections, cur = [], None
    rows = doc.xpath('//div[contains(@class,"meeting__section-row") or '
                     'contains(concat(" ", normalize-space(@class), " "), " meeting-question ")]')
    for r in rows:
        if "meeting__section-row" in r.get("class"):
            m = re.match(r"(\d+)\.\s*(.*)", text(r))
            cur = {"n": m.group(1) if m else "", "title": m.group(2) if m else text(r), "items": []}
            sections.append(cur)
            continue
        c = {x.get("data-column-header-name"): x for x in r.xpath("./div[@data-column-header-name]")}
        if "Nr.p.k." not in c:
            continue
        pid = c.get("Projekta ID")
        a = pid.xpath(".//a") if pid is not None else []
        q = c["Jautājums"]
        for s in q.xpath('.//div[contains(@class,"document-summary")]'):
            s.getparent().remove(s)
        qtext, flags = text(q), []
        while True:   # "(iekļauts papildus) (jautājums tiks skatīts A sadaļā) Informatīvais ziņojums …"
            m = re.match(r"\(([^)]*)\)\s*", qtext)
            if not m or m.group(1) not in FLAGS:
                break
            flags.append(FLAGS[m.group(1)])
            qtext = qtext[m.end():]
        tgt = r.xpath(".//span[@data-collapse-target]")
        docs, approvals, participation = [], [], ""
        if tgt:
            rec = doc.xpath(f'//div[@data-collapse-recipient="{tgt[0].get("data-collapse-target")}"]')
            if rec:
                for sec in rec[0].xpath('.//div[contains(@class,"meeting-question__additional-information-section")]'):
                    head = text((sec.xpath('.//div[contains(@class,"additional-information-title")]') or [None])[0])
                    body = sec.xpath("./div[@data-collapse-recipient]")
                    if head == "Dokumenti":
                        docs = [(text(x), x.get("href")) for x in sec.xpath(".//a[@href]")]
                    elif head == "Saskaņojumi":
                        approvals = [text(x) for x in sec.xpath(".//a")]
                    elif head == "Sabiedrības līdzdalība" and body:
                        participation = text(body[0])
        pid_text = text(pid)
        cur["items"].append({
            "n": text(c["Nr.p.k."]), "tap": pid_text.replace("(IP)", "").strip(), "restricted": "(IP)" in pid_text,
            "legal_act": (re.search(r"/legal_acts/([0-9a-f-]{36})", a[0].get("href")) or [None, None])[1] if a else None,
            "flags": flags, "title": qtext, "presenter": text(c.get("Ziņo")), "invited": text(c.get("Uzaicināti")),
            "status": text(c.get("Statuss")), "documents": docs, "approvals": approvals, "participation": participation})
    return {"title": title, "sections": sections}


# ------------------------------------------------------------------ dokumenta teksts (structuralizer)
def parse_node(raw):
    """-> {'heading': [rindas virs virsraksta], 'header': TAP ID, 'title', 'points': [...], 'annexes': [...], 'signatures': [...]}.
    Punkts: {'n', 'text', 'quotes': [jaunā redakcija pēdiņās, ja punktā ir vairāki teksta lauki], 'children': [...]}.
    Nenumurēts teksta lauks ir punkts ar n=""."""
    doc = H.fromstring(raw)
    tree = doc.xpath('//div[contains(@class,"structuralizer-tree--preview")]')
    if not tree:
        raise ValueError("nav dokumenta teksta (structuralizer-tree)")
    tree = tree[0]
    for x in tree.xpath('.//div[contains(@class,"node-preview__actions")]'):
        x.getparent().remove(x)
    hdr = tree.xpath('.//div[contains(@class,"structuralizer-tree__header")]')
    header_id = norm(hdr[0].text) if hdr else ""
    title = text((tree.xpath('.//div[contains(@class,"structuralizer-tree__header")]//div[contains(@class,"structuralizer-field--preview")]') or [None])[0])
    heading = []
    if hdr:
        for w in hdr[0].itersiblings(preceding=True):
            heading[:0] = [text(g) for g in w.xpath('./div/div | .//div[contains(@class,"structuralizer-tree__content-wrapper")]/div/div')
                           if text(g)]
    signatures = [text(x) for x in tree.xpath('.//div[contains(@class,"signature-list")]/div') if text(x)]

    def is_wrapper(e):
        return e.tag == "div" and e.get("data-numbered-field-wrapper") == "true"

    def fields(w):
        return [text(f) for f in w.xpath('./div[contains(@class,"numbered-field")]/div[@data-numbered-field-preview-field="true"]')]

    def own_number(w):
        n = w.xpath('./div[contains(@class,"numbered-field")]/span[contains(@class,"numbered-field__number")]')
        return text(n[0]) if n else ""

    def skip(ch):
        c = ch.get("class") or ""
        return ("structuralizer__annex-wrapper" in c or "signature-list" in c or "speaker-field-group" in c
                or ch.get("data-numbered-field-preview-field") == "true" or "numbered-field__number" in c)

    def children(e):
        out = []
        for ch in e.iterchildren():
            if not isinstance(ch.tag, str) or skip(ch):
                continue
            if is_wrapper(ch):
                f = fields(ch)
                out.append({"n": own_number(ch), "text": f[0] if f else "", "quotes": f[1:], "children": children(ch)})
            elif "structuralizer-field--preview" in (ch.get("class") or ""):
                t = text(ch)
                if t:
                    out.append({"n": "", "text": t, "quotes": [], "children": []})
            else:
                out.extend(children(ch))
        return out

    points = []
    if hdr:
        for w in hdr[0].itersiblings():
            points.extend(children(w))
    annexes = [{"n": text((a.xpath('.//*[contains(@class,"annex-number")]') or [None])[0]), "text": text(a)}
               for a in tree.xpath('.//div[contains(@class,"structuralizer__annex-wrapper")]')]
    return {"heading": heading, "header": header_id, "title": title, "points": [p for p in points if p["text"] or p["children"]],
            "annexes": annexes, "signatures": signatures}


def flatten(points, depth=0):
    for p in points:
        yield depth, p
        yield from flatten(p["children"], depth + 1)