1a5cc72067
Recognises products by trade name or substance (Slovak inflection aware), queries LightRAG and shows the answer with the retrieved subgraph and ADC source texts.
88 lines
3.1 KiB
Python
88 lines
3.1 KiB
Python
"""Build a compact index of all ADC products for the web demo.
|
|
|
|
The demo cannot load the 800 MB structured dataset, so this script extracts
|
|
the fields needed to recognise a product in a question (name, active
|
|
substances, form, prescription status, ...) and links every product to the
|
|
LightRAG document that represents its active substance set (`rep`). Products
|
|
without a representative have no clinical data in the knowledge graph.
|
|
|
|
python scripts/demo/build_product_index.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "scripts" / "lightrag_ingest"))
|
|
from ingest_adc_structured import DEFAULT_INPUT, dedupe_key, load_json, write_json # noqa: E402
|
|
|
|
DEFAULT_SECTION_DOCS = Path("outputs/lightrag_sections/adc_section_docs.json")
|
|
DEFAULT_OUTPUT = Path("outputs/demo/product_index.json")
|
|
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
|
|
def as_list(value) -> list[str]:
|
|
if not value:
|
|
return []
|
|
if isinstance(value, str):
|
|
return [value]
|
|
return [str(item) for item in value if str(item).strip()]
|
|
|
|
|
|
def compact(record: dict, representative: str | None) -> dict:
|
|
product = record.get("product") or {}
|
|
hints = record.get("graph_hints") or {}
|
|
urls = record.get("urls") or {}
|
|
fields = product.get("detail_fields") or {}
|
|
classes = [c.get("name") for c in hints.get("classification_codes") or [] if isinstance(c, dict)]
|
|
return {
|
|
"name": product.get("name") or hints.get("drug") or "",
|
|
"url": urls.get("detail") or "",
|
|
"pil": urls.get("pil") or "",
|
|
"subs": as_list(hints.get("active_substances")),
|
|
"form": hints.get("dosage_form") or "",
|
|
"rx": fields.get("vydaj") or "",
|
|
"type": fields.get("typ_produktu") or "",
|
|
"group": classes[-1] if classes else "",
|
|
"sukl": hints.get("sukl_code") or "",
|
|
"ind": str(hints.get("indications_text") or "")[:500],
|
|
"rep": representative,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
parser.add_argument("--input", type=Path, default=DEFAULT_INPUT)
|
|
parser.add_argument("--section-docs", type=Path, default=DEFAULT_SECTION_DOCS)
|
|
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
|
|
args = parser.parse_args()
|
|
|
|
# Same grouping as build_section_docs.py: group key -> URL of the uploaded document.
|
|
representatives = {
|
|
dedupe_key(doc, "substance"): doc["urls"]["detail"] for doc in load_json(args.section_docs)
|
|
}
|
|
|
|
print(f"Loading {args.input} ...")
|
|
records = load_json(args.input)
|
|
index = []
|
|
stats: Counter = Counter()
|
|
for record in records:
|
|
rep = representatives.get(dedupe_key(record, "substance"))
|
|
stats["in_graph" if rep else "not_in_graph"] += 1
|
|
index.append(compact(record, rep))
|
|
|
|
write_json(args.output, index)
|
|
print(f"Products: {len(index)} | {dict(stats)}")
|
|
print(f"Saved: {args.output}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|