sources: iXBRL-extraktor for Bolagsverkets HVD-arsredovisningar

- arelle-release + lxml for taxonomi-generisk extraction
- alla tаggade fakta (concept+context+unit+varde+dimensioner) per bolag
- union = hela svenska bastaxonomin (tusentals variabler)
- iter_corpus() for direkt batch over ytterzip->innerzip->XHTML
- batch-runner i workers/ixbrl_batch.py

services/api/requirements.txt

@@ -8,3 +8,6 @@ python-dotenv==1.0.1
 pyyaml==6.0.2
 rapidfuzz==3.11.0
 tenacity==9.0.0
+# iXBRL-extraction (Bolagsverket HVD-årsredovisningar) — se sources/ixbrl.py
+arelle-release==2.44.4
+lxml==6.1.2

services/api/sources/ixbrl.py

@@ -0,0 +1,221 @@
+"""
+0pt data engine — Bolagsverket iXBRL-extractor (official bulk, HVD).
+
+Bolagsverket publicerar årsredovisningar som inline-XBRL (XHTML med ix:-fakta
+taggade mot svensk bastaxonomi: se-gen-base, se-cd-base, se-bol-base,
+se-k2-type, se-k3-type, se-mem-base m.fl.). Den här modulen extraherar ALLA
+taggade fakta taxonomi-generiskt — ingen hårdkodad concept-lista. Unionen av
+alla concept över bolagen = hela bastaxonomin (tusentals variabler per bolag:
+nettoomsättning, resultat, balansposter, eget kapital, soliditet, antal
+anställda, styrelse, revisor, fastställelseintyg, dimensioner …).
+
+Källa (verifierad 2026-08-25): vardefulla-datamangder.bolagsverket.se
+(sourcing chain nivå 1 — official bulk). Landning: /data/0pt/landing/bolagsverket/
+
+Kräver: arelle-release + lxml (se services/api/requirements.txt).
+
+Varje faktum levereras med provenance:
+  source_id=bolagsverket · level 3 (XBRL-maskinläst = garanterad) · confidence 0.99
+"""
+from __future__ import annotations
+
+import io
+import json
+import os
+import zipfile
+from typing import Any, Iterator
+
+from arelle import Cntlr, ModelManager
+
+SOURCE_ID = "bolagsverket"
+SOURCE_URL = "https://vardefulla-datamangder.bolagsverket.se"
+
+
+def _iso(dt):
+    if dt is None:
+        return None
+    try:
+        return dt.isoformat()
+    except Exception:
+        return str(dt)
+
+
+def _qname_split(qstr: str, nsmap: dict):
+    if not qstr:
+        return None, None
+    if ":" in qstr:
+        prefix, local = qstr.split(":", 1)
+        return nsmap.get(prefix, None), local
+    return None, qstr
+
+
+def _dim_dict(ctx, nsmap):
+    out: dict[str, str] = {}
+    try:
+        qd = ctx.qnameDims
+    except Exception:
+        return out
+    if not qd:
+        return out
+    for dq, mq in qd.items():
+        dloc = dq.localName if hasattr(dq, "localName") else str(dq)
+        mloc = mq.localName if hasattr(mq, "localName") else str(mq)
+        out[dloc] = mloc
+    return out
+
+
+def _new_manager() -> ModelManager:
+    cntlr = Cntlr.Cntlr(logFileName="logToPrint")
+    return ModelManager.initialize(cntlr)
+
+
+def extract_company_xhtml(path: str, manager: ModelManager | None = None) -> dict[str, Any]:
+    """Extrahera ALLA iXBRL-fakta ur en årsredovisnings-XHTML (ett bolag)."""
+    owns = manager is None
+    manager = manager or _new_manager()
+    try:
+        model_xbrl = manager.load(path)
+        if model_xbrl is None:
+            raise RuntimeError(f"kunde inte ladda {path}")
+
+        nsmap = dict(model_xbrl.modelDocument.xmlRootElement.nsmap or {})
+        contexts = model_xbrl.contexts
+
+        def build(f, tuple_of=None):
+            ctx = contexts.get(f.contextID) if f.contextID else None
+            ent = getattr(ctx, "entityIdentifier", None) if ctx is not None else None
+            qstr = f.get("name")
+            ns, local = _qname_split(qstr, nsmap)
+            return {
+                "concept": local,
+                "concept_qname": qstr,
+                "namespace": ns,
+                "tuple_of": tuple_of,
+                "value": f.value,
+                "numeric": f.xValue if getattr(f, "isNumeric", False) else None,
+                "decimals": f.decimals,
+                "scale": f.scale,
+                "sign": getattr(f, "sign", None),
+                "precision": f.precision,
+                "is_nil": getattr(f, "isNil", False),
+                "period_start": _iso(ctx.startDatetime) if ctx is not None else None,
+                "period_end": _iso(ctx.endDatetime) if ctx is not None else None,
+                "instant": _iso(ctx.instantDatetime) if ctx is not None else None,
+                "entity_identifier": ent[1] if ent else None,
+                "entity_scheme": ent[0] if ent else None,
+                "unit": f.unitID,
+                "unit_measures": [str(m) for m in f.unit.measures] if f.unit is not None else None,
+                "dimensions": _dim_dict(ctx, nsmap) if ctx is not None else None,
+            }
+
+        facts = []
+        for fact in model_xbrl.facts:
+            facts.append(build(fact))
+            if getattr(fact, "isTuple", False):
+                tq = fact.get("name")
+                for sub in getattr(fact, "modelTupleFacts", []) or []:
+                    facts.append(build(sub, tuple_of=tq))
+
+        orgnr = next((f["entity_identifier"] for f in facts if f["entity_identifier"]), None)
+
+        try:
+            parts = [t.strip() for t in model_xbrl.modelDocument.xmlRootElement.itertext() if t and t.strip()]
+            narrative = " ".join(parts)[:200000]
+        except Exception:
+            narrative = None
+
+        return {
+            "source_id": SOURCE_ID,
+            "source_url": SOURCE_URL,
+            "file": os.path.basename(path),
+            "orgnr": orgnr,
+            "facts": facts,
+            "fact_count": len(facts),
+            "narrative": narrative,
+            "provenance": {
+                "source_id": SOURCE_ID,
+                "level": 3,
+                "confidence": 0.99,
+                "method": "iXBRL (maskinläst bastaxonomi)",
+            },
+            "error": None,
+        }
+    finally:
+        if owns:
+            manager.close()
+
+
+def _iter_xhtml_from_outer_zip(outer_path: str) -> Iterator[tuple[str, str]]:
+    """Yield (rel_name, xhtml_text) för varje bolag i en ytterzip."""
+    with zipfile.ZipFile(outer_path) as outer:
+        for inner_name in outer.namelist():
+            if not inner_name.lower().endswith(".zip"):
+                continue
+            inner_bytes = outer.read(inner_name)
+            with zipfile.ZipFile(io.BytesIO(inner_bytes)) as inner:
+                for xname in inner.namelist():
+                    if xname.lower().endswith((".xhtml", ".xhtm", ".html", ".htm", ".xml")):
+                        yield f"{os.path.basename(outer_path)}/{inner_name}/{xname}", inner.read(xname)
+
+
+def iter_corpus(base_dir: str) -> Iterator[dict[str, Any]]:
+    """Iterera hela korpusen: ytterzip -> innerzip -> XHTML -> extraherade fakta."""
+    manager = _new_manager()
+    try:
+        outer_zips = []
+        for root, _dirs, names in os.walk(base_dir):
+            for n in names:
+                if n.lower().endswith(".zip"):
+                    outer_zips.append(os.path.join(root, n))
+        outer_zips.sort()
+
+        for outer in outer_zips:
+            for rel, xhtml_text in _iter_xhtml_from_outer_zip(outer):
+                tmp = f"/tmp/ixbrl_tmp_{os.getpid()}.xhtml"
+                with open(tmp, "wb") as fh:
+                    fh.write(xhtml_text)
+                try:
+                    rec = extract_company_xhtml(tmp, manager=manager)
+                except Exception as e:
+                    rec = {"source_id": SOURCE_ID, "file": rel, "facts": [],
+                           "fact_count": 0, "narrative": None,
+                           "error": f"{type(e).__name__}: {e}"}
+                yield rec
+    finally:
+        manager.close()
+
+
+def extract_company_bytes(xhtml_bytes: bytes, manager: ModelManager | None = None) -> dict[str, Any]:
+    tmp = f"/tmp/ixbrl_tmp_{os.getpid()}.xhtml"
+    with open(tmp, "wb") as fh:
+        fh.write(xhtml_bytes)
+    try:
+        return extract_company_xhtml(tmp, manager=manager)
+    finally:
+        try:
+            os.unlink(tmp)
+        except OSError:
+            pass
+
+
+def to_jsonl(record: dict[str, Any]) -> str:
+    return json.dumps(record, ensure_ascii=False, default=str)
+
+
+if __name__ == "__main__":
+    import sys
+
+    target = sys.argv[1]
+    if os.path.isfile(target) and target.lower().endswith(".zip"):
+        manager = _new_manager()
+        try:
+            for rel, xhtml_text in _iter_xhtml_from_outer_zip(target):
+                rec = extract_company_bytes(xhtml_text.encode("utf-8", "replace"), manager=manager)
+                print(to_jsonl(rec))
+        finally:
+            manager.close()
+    elif os.path.isfile(target):
+        print(to_jsonl(extract_company_xhtml(target)))
+    else:
+        for rec in iter_corpus(target):
+            print(to_jsonl(rec))