diff --git a/services/api/requirements.txt b/services/api/requirements.txt index f019664..7fdef00 100644 --- a/services/api/requirements.txt +++ b/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 diff --git a/services/api/sources/ixbrl.py b/services/api/sources/ixbrl.py new file mode 100644 index 0000000..46f44ee --- /dev/null +++ b/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))