med-det-samme/tools/fest_flatten.py

#!/usr/bin/env python3
"""Flatten DMP's FEST M30 extract into the app's slim.json (docs/fest-dataset.md).

Downloads fest251.zip (the human-use Rekvirent extract) from DMP, streams the
~115 MB XML with iterparse, and writes the slim autocomplete dataset:

    {"version": 1, "updated": "YYYY-MM-DD", "entries": [
        {"name", "strength", "unit", "form", "atc", "packageSize"}, ...]}

Stdlib only. Designed to be run by Gradle before release builds:
- exits 0 immediately if the existing output is younger than --max-age-days;
- on download failure with an existing output, warns and keeps it (an offline
  release build must never break on a stale-but-working dataset).
"""

import argparse
import datetime as dt
import io
import json
import re
import sys
import tempfile
import urllib.request
import xml.etree.ElementTree as ET
import zipfile
from collections import Counter
from pathlib import Path

FEST_URL = (
    "https://www.dmp.no/globalassets/documents/om-oss/"
    "distribusjon-av-legemiddeldata/fest/festfiler/fest251.zip"
)

# Strength starts at the first token containing a digit, e.g.
# "Thylip kaps 200 mg" -> "200 mg"; "Kodimagnyl ... 9,6/500 mg" works too.
STRENGTH_RE = re.compile(r"\s(\d[\d,./]*\s?\S*.*)$")


def local(tag: str) -> str:
    """Strip XML namespace: '{ns}Varenavn' -> 'Varenavn'."""
    return tag.rsplit("}", 1)[-1]


def parse_fest(xml_file) -> dict:
    """One streaming pass: brand entries + package sizes keyed by brand id."""
    updated = ""
    brands: dict[str, dict] = {}  # merkevare Id -> entry
    pack_sizes: dict[str, Counter] = {}  # merkevare Id -> Counter of Mengde

    context = ET.iterparse(xml_file, events=("end",))
    for _, elem in context:
        tag = local(elem.tag)

        if tag == "HentetDato":
            updated = (elem.text or "")[:10]

        elif tag == "OppfLegemiddelMerkevare":
            status = elem.find("./{*}Status")
            mv = elem.find("./{*}LegemiddelMerkevare")
            if mv is not None and status is not None and status.get("V") == "A":
                brand = flatten_brand(mv)
                if brand:
                    brands[brand.pop("_id")] = brand
            elem.clear()

        elif tag == "OppfLegemiddelpakning":
            status = elem.find("./{*}Status")
            if status is not None and status.get("V") == "A":
                for info in elem.iter():
                    if local(info.tag) == "Pakningsinfo":
                        ref = info.find("./{*}RefLegemiddelMerkevare")
                        mengde = info.find("./{*}Mengde")
                        if (
                            ref is not None
                            and ref.text
                            and mengde is not None
                            and mengde.text
                        ):
                            try:
                                size = float(mengde.text.replace(",", "."))
                            except ValueError:
                                continue
                            pack_sizes.setdefault(ref.text, Counter())[size] += 1
            elem.clear()

    for brand_id, sizes in pack_sizes.items():
        if brand_id in brands and sizes:
            # Most common package size; ties resolved towards the smaller pack.
            best = max(sorted(sizes), key=lambda s: sizes[s])
            brands[brand_id]["packageSize"] = best

    entries = sorted(brands.values(), key=lambda e: (e["name"].lower(), e["strength"]))
    return {"version": 1, "updated": updated, "entries": entries}


def flatten_brand(mv) -> dict | None:
    fields = {}
    for child in mv:
        fields[local(child.tag)] = child
    name_el = fields.get("Varenavn")
    if name_el is None or not (name_el.text or "").strip():
        return None
    name = name_el.text.strip()

    nfs = (
        fields.get("NavnFormStyrke") is not None and fields["NavnFormStyrke"].text
    ) or ""
    # Names may themselves contain digits ("2-Hydroxyethyl …") — strip the name
    # prefix before hunting for the strength part.
    if nfs.lower().startswith(name.lower()):
        nfs = nfs[len(name) :]
    m = STRENGTH_RE.search(nfs)
    strength = m.group(1).strip() if m else ""

    form_el = fields.get("LegemiddelformKort")
    form = form_el.get("DN", "") if form_el is not None else ""

    unit = ""
    adm = fields.get("AdministreringLegemiddel")
    if adm is not None:
        enhet = adm.find("./{*}EnhetDosering")
        if enhet is not None:
            unit = enhet.get("DN", "")

    atc_el = fields.get("Atc")
    atc = atc_el.get("V") if atc_el is not None else None

    inner_id = None
    for child in mv:
        if local(child.tag) == "Id":
            inner_id = child.text
            break
    if not inner_id:
        return None

    entry = {
        "_id": inner_id,
        "name": name,
        "strength": strength,
        "unit": unit or "dose",
        "form": form,
    }
    if atc:
        entry["atc"] = atc
    return entry


def is_fresh(out_path: Path, max_age_days: int) -> bool:
    if max_age_days <= 0 or not out_path.exists():
        return False
    try:
        updated = json.loads(out_path.read_text())["updated"]
        age = dt.date.today() - dt.date.fromisoformat(updated)
        return age.days <= max_age_days
    except (KeyError, ValueError, json.JSONDecodeError):
        return False  # unreadable/odd file -> regenerate


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--out", required=True, type=Path)
    ap.add_argument("--url", default=FEST_URL)
    ap.add_argument("--max-age-days", type=int, default=30)
    ap.add_argument("--force", action="store_true")
    args = ap.parse_args()

    if not args.force and is_fresh(args.out, args.max_age_days):
        print(
            f"fest_flatten: {args.out} is fresh (<= {args.max_age_days} days), skipping"
        )
        return 0

    print(f"fest_flatten: downloading {args.url} ...", flush=True)
    try:
        with tempfile.TemporaryFile() as tmp:
            with urllib.request.urlopen(args.url, timeout=300) as resp:
                while chunk := resp.read(1 << 20):
                    tmp.write(chunk)
            tmp.seek(0)
            with zipfile.ZipFile(tmp) as zf:
                inner = zf.namelist()[0]
                print(f"fest_flatten: parsing {inner} ...", flush=True)
                with zf.open(inner) as xml_file:
                    dataset = parse_fest(io.BufferedReader(xml_file))
    except Exception as e:  # noqa: BLE001 - any failure has the same handling
        if args.out.exists():
            print(
                f"fest_flatten: WARNING: refresh failed ({e}); keeping existing {args.out}"
            )
            return 0
        print(f"fest_flatten: ERROR: {e} and no existing dataset", file=sys.stderr)
        return 1

    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(json.dumps(dataset, ensure_ascii=False, separators=(",", ":")))
    print(
        f"fest_flatten: wrote {len(dataset['entries'])} entries (updated {dataset['updated']}) to {args.out}"
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())