#!/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())