diff --git a/ask_aclasta.py b/ask_aclasta.py new file mode 100644 index 0000000..f3a3a41 --- /dev/null +++ b/ask_aclasta.py @@ -0,0 +1,84 @@ + + +"""Ask the local LightRAG server a question that exists in the current graph. + +Usage: + python ask_aclasta.py +""" + +from __future__ import annotations + +import argparse +import json +import sys +import urllib.error +import urllib.request +from pathlib import Path + +from pharmacy_agent import QUERY_SETTINGS, prepare_query + + +DEFAULT_URL = "http://localhost:9621" +DEFAULT_MODE = "mix" +DEFAULT_QUESTION = "Ake su kontraindikacie lieku ACARIZAX?" + + +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8") +if hasattr(sys.stderr, "reconfigure"): + sys.stderr.reconfigure(encoding="utf-8") + + +def post_query(base_url: str, question: str, mode: str, timeout: int) -> dict: + index = Path(__file__).resolve().parent / "outputs/demo/product_index.json" + products = json.loads(index.read_text(encoding="utf-8")) if index.exists() else [] + prepared = prepare_query(question, products) + payload = json.dumps( + {**QUERY_SETTINGS, "query": prepared["query"], "user_prompt": prepared["user_prompt"], + "mode": mode, "include_references": True, "include_chunk_content": True}, + ensure_ascii=False, + ).encode("utf-8") + request = urllib.request.Request( + f"{base_url.rstrip('/')}/query", + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=timeout) as response: + return json.loads(response.read().decode("utf-8")) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Query LightRAG about ACARIZAX contraindications.") + parser.add_argument("--url", default=DEFAULT_URL, help="LightRAG server URL.") + parser.add_argument("--mode", default=DEFAULT_MODE, choices=("mix", "hybrid", "local", "global", "naive")) + parser.add_argument("--timeout", type=int, default=120) + parser.add_argument("--question", default=DEFAULT_QUESTION) + args = parser.parse_args() + + try: + result = post_query(args.url, args.question, args.mode, args.timeout) + except urllib.error.URLError as exc: + raise SystemExit( + "LightRAG server is not reachable. Start it first with:\n" + " python start_servers.py\n\n" + f"Details: {exc}" + ) from exc + except json.JSONDecodeError as exc: + raise SystemExit(f"LightRAG returned invalid JSON: {exc}") from exc + + response = result.get("response") + if response is None: + print(json.dumps(result, ensure_ascii=False, indent=2)) + return + + print(response) + references = result.get("references") or [] + if references: + print("\nZdroje:") + for ref in references: + print(f"[{ref.get('reference_id', '?')}] {ref.get('file_path', '')}") + + +if __name__ == "__main__": + main() diff --git a/demo/app.py b/demo/app.py new file mode 100644 index 0000000..346b90b --- /dev/null +++ b/demo/app.py @@ -0,0 +1,367 @@ +"""Web demo of the pharmacy decision-support agent (Streamlit). + +The agent answers questions about indications, contraindications and +interactions using the LightRAG knowledge graph built from ADC leaflets. +Every answer is shown together with its justification: the part of the +knowledge graph that was retrieved and the ADC leaflet texts it came from. + +Run (LightRAG server must be running, see start_servers.py): + + streamlit run demo/app.py +""" + +from __future__ import annotations + +import json +import html +import sys +import re +import unicodedata +from collections import Counter, defaultdict +from pathlib import Path + +import requests +import streamlit as st +from pyvis.network import Network + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) +from pharmacy_agent import QUERY_SETTINGS, normalize, build_aliases, detect_drugs, describe_match, prepare_query +PRODUCT_INDEX = ROOT / "outputs" / "demo" / "product_index.json" +DEFAULT_URL = "http://localhost:9621" + +MODES = { + "mix": "mix – graf + textové úryvky (odporúčané)", + "hybrid": "hybrid – lokálny + globálny graf", + "local": "local – entity a ich susedia", + "global": "global – vzťahy", + "naive": "naive – len textové úryvky (bez grafu)", +} + +EXAMPLES = [ + "Môžem užívať Nurofen spolu s warfarínom?", + "Kedy sa nesmie užívať metformín?", + "Na čo sa používa Aclexa?", + "Aké lieky sa nemajú kombinovať s klaritromycínom?", +] + +ENTITY_COLORS = { + "drug": "#4C78A8", + "activesubstance": "#72B7B2", + "drugclass": "#B279A2", + "indication": "#54A24B", + "condition": "#F58518", + "contraindication": "#E45756", + "interaction": "#EECA3B", + "patientgroup": "#9D755D", +} +DEFAULT_ENTITY_COLOR = "#BAB0AC" + +# Relation keyword -> (label, colour) used to highlight edges in the graph. +RELATION_KINDS = ( + ("kontraindik", "kontraindikácia", "#E45756"), + ("interak", "interakcia", "#F58518"), + ("indik", "indikácia", "#54A24B"), + ("upozorn", "upozornenie", "#EECA3B"), +) + +# Words that start product names but do not identify a product. + + +# ---------------------------------------------------------------- helpers + + + +def api(base_url: str, method: str, path: str, timeout: int = 300, **kwargs): + response = requests.request(method, f"{base_url}{path}", timeout=timeout, **kwargs) + response.raise_for_status() + return response.json() + + +@st.cache_data(show_spinner="Načítavam index liekov ADC ...") +def load_product_index() -> list[dict] | None: + if not PRODUCT_INDEX.exists(): + return None + return json.loads(PRODUCT_INDEX.read_text(encoding="utf-8")) + + + + + + + + +def clean(text: str | None, limit: int = 600) -> str: + """LightRAG joins merged descriptions with ; show them as paragraphs.""" + text = str(text or "").replace("", "\n\n") + return text if len(text) <= limit else text[:limit].rsplit(" ", 1)[0] + " …" + + +def relation_kind(keywords: str, description: str) -> tuple[str, str]: + text = normalize(f"{keywords} {description}") + for needle, label, color in RELATION_KINDS: + if needle in text: + return label, color + return "iný vzťah", "#9E9E9E" + + +def render_graph(nodes: list[dict], edges: list[dict], highlight: set[str], height: int = 560) -> None: + net = Network(height=f"{height}px", width="100%", bgcolor="#ffffff", font_color="#222222", cdn_resources="in_line") + net.barnes_hut(gravity=-800, central_gravity=0.15, spring_length=100) + for node in nodes: + entity_type = str(node.get("type") or "").lower() + is_focus = normalize(node["id"]) in highlight + net.add_node( + node["id"], + label=node["id"][:40], + title=f"{node['id']} ({entity_type or '?'})\n\n{clean(node.get('description'))}", + color=ENTITY_COLORS.get(entity_type, DEFAULT_ENTITY_COLOR), + size=28 if is_focus else 14, + borderWidth=4 if is_focus else 1, + font={"size":18,"strokeWidth":3,"strokeColor":"#ffffff"}, + ) + names = {node["id"] for node in nodes} + for edge in edges: + if edge["src"] not in names or edge["tgt"] not in names: + continue + label, color = relation_kind(edge.get("keywords", ""), edge.get("description", "")) + net.add_edge( + edge["src"], edge["tgt"], color=color, width=3 if label != "iný vzťah" else 1, + title=f"{label}: {edge.get('keywords', '')}\n\n{clean(edge.get('description'))}", + ) + graph_html = net.generate_html(notebook=False) + graph_html = graph_html.replace("", """""") + st.iframe(graph_html, height=height + 20) + + +def graph_legend() -> None: + entity_html = " ".join( + f"{t}" + for t, c in ENTITY_COLORS.items() + ) + relation_html = " ".join( + f"{label}" for _, label, c in RELATION_KINDS + ) + st.markdown(f"Uzly: {entity_html}
Hrany: {relation_html}", unsafe_allow_html=True) + + +def product_name(url: str, by_url: dict) -> str: + product = by_url.get(url) + return product["name"] if product else url + + +# ---------------------------------------------------------------- page + +st.set_page_config(page_title="Lekárnický asistent", page_icon="💊", layout="wide") + +with st.sidebar: + st.header("Nastavenia") + base_url = st.text_input("LightRAG server", DEFAULT_URL) + mode = st.selectbox("Režim vyhľadávania", list(MODES), format_func=MODES.get) + top_k = st.slider("Počet entít / vzťahov (top_k)", 10, 80, 40, step=5) + max_graph_nodes = st.slider("Max. uzlov v zobrazenom grafe", 10, 120, 50, step=10) + try: + health = api(base_url, "GET", "/health", timeout=5) + config = health.get("configuration") or {} + st.success(f"Server beží · LLM: {config.get('llm_model', '?')}") + except Exception as exc: # noqa: BLE001 + st.error(f"Server nedostupný: {exc}") + st.caption( + "Odpovede sú generované jazykovým modelom na základe príbalových letákov ADC " + "a nenahrádzajú konzultáciu s lekárom alebo lekárnikom." + ) + +products = load_product_index() +if products is None: + st.warning( + f"Chýba index liekov `{PRODUCT_INDEX.relative_to(ROOT)}` – rozpoznávanie obchodných názvov je vypnuté. " + "Vytvorte ho príkazom `python scripts/demo/build_product_index.py`." + ) + aliases, by_prefix, by_url = {}, {}, {} +else: + aliases, by_prefix, by_url = build_aliases(products) + +st.title("💊 Lekárnický asistent nad znalostným grafom ADC") +st.caption("Indikácie · kontraindikácie · interakcie — odpoveď s odôvodnením zo znalostného grafu a príbalových letákov") + +tab_ask, tab_explore = st.tabs(["Otázka", "Prieskum grafu"]) + +# ---------------------------------------------------------------- tab: question + +with tab_ask: + if "question" not in st.session_state: + st.session_state.question = EXAMPLES[0] + cols = st.columns(len(EXAMPLES)) + for col, example in zip(cols, EXAMPLES): + if col.button(example, width="stretch"): + st.session_state.question = example + question = st.text_area("Otázka", key="question", height=80) + ask = st.button("Opýtať sa", type="primary") + + if ask and question.strip(): + prepared = prepare_query(question, products or [], aliases, by_prefix) + matches = prepared["matches"] + query = prepared["query"] + payload = {**QUERY_SETTINGS,"query":query,"mode":mode,"top_k":top_k, + "user_prompt":prepared["user_prompt"]} + try: + with st.spinner("Vyhľadávam v znalostnom grafe ..."): + data = api(base_url, "POST", "/query/data", json=payload) + with st.spinner("Generujem odpoveď ..."): + answer = api(base_url, "POST", "/query", json={**payload, "include_references": True, "include_chunk_content": True}) + except Exception as exc: # noqa: BLE001 + st.error(f"Chyba pri dopyte na LightRAG: {exc}") + st.stop() + st.session_state.result = { + "question": question, "query": query, "user_prompt": payload.get("user_prompt"), + "matches": matches, "data": data, "answer": answer, + } + + result = st.session_state.get("result") + if result: + matches, data, answer = result["matches"], result["data"].get("data") or {}, result["answer"] + + if matches: + st.subheader("Rozpoznané lieky") + for col, m in zip(st.columns(min(len(matches), 4)), matches): + with col.container(border=True): + st.markdown(f"**{m['label']}** \n{'účinná látka' if m['kind'] == 'substance' else 'obchodný názov'}") + if m["substances"]: + st.write("Látka: " + ", ".join(m["substances"])) + if m.get("ambiguous"): + st.warning("Značka má viac zložení. Uveďte presný prípravok.") + ex = m["example"] + st.write(f"{ex['rx'] or ''} \n{ex['form'] or ''}") + if m["in_graph"]: + st.success("Index obsahuje klinického reprezentanta") + else: + st.warning("Index neobsahuje klinického reprezentanta; dôvod treba overiť v ADC.") + st.caption(f"{m['count']} balení v ADC · [ADC]({ex['url']})") + + answer_tab, graph_tab, source_tab, debug_tab = st.tabs( + ["Odpoveď", "Znalostný graf", "Zdroje z ADC", "Detaily dopytu"] + ) + + with answer_tab: + st.markdown(answer.get("response", "")) + refs = answer.get("references") or [] + if refs: + st.markdown("**Referencie**") + for ref in refs: + path = ref.get("file_path", "") + st.markdown(f"[{ref.get('reference_id')}] [{product_name(path, by_url)}]({path})") + + with graph_tab: + entities = data.get("entities") or [] + relations = data.get("relationships") or [] + focus = {normalize(s) for m in matches for s in m["substances"] + [m["label"]]} + # Keep the most connected entities when the retrieved subgraph is large. + degree = Counter() + for r in relations: + degree[r["src_id"]] += 1 + degree[r["tgt_id"]] += 1 + # Context limits can omit endpoint entities while retaining their edges. + # Recover endpoint labels from the actual relations, without inventing descriptions. + present = {e["entity_name"] for e in entities} + entities = list(entities) + [ + {"entity_name":name,"entity_type":"unknown","description":""} + for name in degree if name not in present + ] + ranked = sorted(entities, key=lambda e: (normalize(e["entity_name"]) not in focus, -degree[e["entity_name"]])) + shown = ranked[:max_graph_nodes] + st.caption( + f"Samostatne vyhľadaný podgraf: {len(entities)} entít a {len(relations)} vzťahov " + f"(zobrazených {len(shown)} uzlov). Prejdite myšou na uzol alebo hranu pre popis." + ) + graph_legend() + render_graph( + [{"id": e["entity_name"], "type": e.get("entity_type"), "description": e.get("description")} for e in shown], + [{"src": r["src_id"], "tgt": r["tgt_id"], "keywords": r.get("keywords", ""), + "description": r.get("description", "")} for r in relations], + focus, + ) + with st.expander(f"Tabuľka vzťahov ({len(relations)})"): + st.dataframe( + [{ + "typ": relation_kind(r.get("keywords", ""), r.get("description", ""))[0], + "od": r["src_id"], "k": r["tgt_id"], + "kľúčové slová": r.get("keywords", ""), + "popis": r.get("description", ""), + "zdroj": product_name(str(r.get("file_path", "")).split("")[0], by_url), + } for r in relations], + width="stretch", hide_index=True, + ) + + with source_tab: + chunks = [ + {"file_path":ref.get("file_path", ""),"content":content} + for ref in answer.get("references") or [] for content in ref.get("content") or [] + ] + st.caption(f"{len(chunks)} zdrojových úryvkov vrátených s touto odpoveďou. " + "Podgraf bol vyhľadaný samostatným volaním a nemusí presne zodpovedať kontextu generovania.") + if not chunks: + st.warning("Odpoveď neobsahuje zdrojové úryvky. Jej tvrdenia nemožno overiť z tohto výstupu.") + grouped = defaultdict(list) + for chunk in chunks: + grouped[chunk.get("file_path", "")].append(chunk) + for path, items in grouped.items(): + product = by_url.get(path) + title = product["name"] if product else path + with st.expander(f"📄 {title} ({len(items)} úryvkov)"): + if product: + st.markdown(f"[Detail v ADC]({path}) · [Príbalový leták]({product['pil']})") + for chunk in items: + st.markdown(f"
{html.escape(chunk.get('content', ''))}
", + unsafe_allow_html=True) + + with debug_tab: + meta = result["data"].get("metadata") or {} + st.write("**Dopyt odoslaný do LightRAG:**") + st.code(result["query"], language=None) + if result.get("user_prompt"): + st.write("**Doplňujúce údaje z ADC pre model:**") + st.code(result["user_prompt"], language=None) + st.write("**Kľúčové slová extrahované LLM:**", meta.get("keywords")) + st.json(meta.get("processing_info") or {}) + +# ---------------------------------------------------------------- tab: explore + +with tab_explore: + st.write("Vyhľadajte entitu (liečivo, ochorenie, skupinu ...) a zobrazte jej okolie v znalostnom grafe.") + term = st.text_input("Hľadať entitu", "warfarín") + labels = [] + if term.strip(): + try: + labels = api(base_url, "GET", "/graph/label/search", params={"q": term, "limit": 30}, timeout=30) + except Exception as exc: # noqa: BLE001 + st.error(f"Chyba: {exc}") + if labels: + col1, col2 = st.columns([3, 1]) + label = col1.selectbox("Entita", labels) + depth = col2.radio("Hĺbka", [1, 2], horizontal=True) + graph = api(base_url, "GET", "/graphs", params={"label": label, "max_depth": depth, "max_nodes": max_graph_nodes}, timeout=60) + nodes = [{ + "id": n["id"], + "type": (n.get("properties") or {}).get("entity_type"), + "description": (n.get("properties") or {}).get("description"), + } for n in graph.get("nodes", [])] + edges = [{ + "src": e["source"], "tgt": e["target"], + "keywords": (e.get("properties") or {}).get("keywords", ""), + "description": (e.get("properties") or {}).get("description", ""), + } for e in graph.get("edges", [])] + st.caption(f"{len(nodes)} uzlov, {len(edges)} hrán") + graph_legend() + render_graph(nodes, edges, {normalize(label)}) + center = next((n for n in nodes if n["id"] == label), None) + if center: + st.markdown(f"**{label}** ({center['type']})") + st.markdown(clean(center["description"], limit=3000)) + elif term.strip(): + st.info("Nenašla sa žiadna entita.") diff --git a/demo/requirements.txt b/demo/requirements.txt new file mode 100644 index 0000000..56deb51 --- /dev/null +++ b/demo/requirements.txt @@ -0,0 +1,3 @@ +streamlit>=1.40 +pyvis>=0.3.2 +requests>=2.31 diff --git a/pharmacy_agent.py b/pharmacy_agent.py new file mode 100644 index 0000000..33f4c68 --- /dev/null +++ b/pharmacy_agent.py @@ -0,0 +1,138 @@ +"""Shared, offline-testable product recognition for the demo and evaluation.""" +from __future__ import annotations +import re +import unicodedata +from collections import Counter, defaultdict + +FORM_WORDS = { + "tbl", "cps", "sol", "crm", "gel", "ung", "inj", "inf", "plv", "sus", "sir", "gtt", "spr", + "flm", "obd", "eff", "ret", "pro", "por", "ijf", "dis", "lyo", "aer", "emp", "pst", "liq", +} + + +def normalize(text: str) -> str: + """Lowercase, strip diacritics and punctuation (Slovak inflection is handled by prefix matching).""" + text = unicodedata.normalize("NFKD", text) + text = "".join(ch for ch in text if not unicodedata.combining(ch)) + return re.sub(r"[^a-z0-9]+", " ", text.lower()).strip() + + +def build_aliases(_products: list[dict]) -> tuple[dict, dict, dict]: + """Map normalized names to products. + + Returns (aliases, by_prefix, by_url): aliases maps an alias to + {"kind": "substance"|"product", "label": str, "products": [index, ...]}. + """ + aliases: dict[str, dict] = {} + substance_products: dict[str, list[int]] = defaultdict(list) + brand_products: dict[str, list[int]] = defaultdict(list) + + for i, product in enumerate(_products): + for substance in product["subs"]: + key = normalize(substance) + if len(key) >= 4: + substance_products[key].append(i) + tokens = normalize(product["name"]).split() + brand = [] + for token in tokens: + if any(ch.isdigit() for ch in token) or token in FORM_WORDS: + if brand: + break + continue # leading "3M", "0 9 %" ... are not the brand + brand.append(token) + if brand and len(brand[0]) >= 3: + brand_products[brand[0]].append(i) + if len(brand) >= 2: + brand_products[" ".join(brand[:2])].append(i) + + for key, ids in substance_products.items(): + label = Counter(s for i in ids for s in _products[i]["subs"] if normalize(s) == key).most_common(1)[0][0] + aliases[key] = {"kind": "substance", "label": label, "products": ids} + for key, ids in brand_products.items(): + if key in aliases: + continue + # A word shared by many unrelated products (e.g. "vitamin") is not a brand. + if len({tuple(_products[i]["subs"]) for i in ids}) > 6: + continue + aliases[key] = {"kind": "product", "label": key.title(), "products": ids} + + by_prefix: dict[str, list[str]] = defaultdict(list) + for key in aliases: + by_prefix[key[:4]].append(key) + by_url = {p["url"]: p for p in _products if p["url"]} + return aliases, by_prefix, by_url + + +def detect_drugs(question: str, aliases: dict, by_prefix: dict) -> list[dict]: + """Find products and active substances mentioned in the question. + + Slovak nouns are inflected (warfarínom, ibuprofénu), so a question word + matches an alias it starts with, allowing a short ending. + """ + tokens = normalize(question).split() + found, used = {}, set() + for n in (2, 1): + for i in range(len(tokens) - n + 1): + if any(j in used for j in range(i, i + n)): + continue + phrase = " ".join(tokens[i:i + n]) + for alias in by_prefix.get(phrase[:4], []): + if len(alias.split()) != n: + continue + exact = phrase == alias + inflected = len(alias) >= 5 and phrase.startswith(alias) and len(phrase) - len(alias) <= 4 + if exact or inflected: + found[alias] = aliases[alias] + used.update(range(i, i + n)) + break + return list(found.values()) + + +def describe_match(match: dict, products: list[dict]) -> dict: + items = [products[i] for i in match["products"]] + substances = Counter(tuple(p["subs"]) for p in items if p["subs"]).most_common() + covered = [p for p in items if p["rep"]] + example = (covered or items)[0] + return { + "label": match["label"], + "kind": match["kind"], + "substances": list(substances[0][0]) if len(substances) == 1 else [], + "ambiguous": len(substances) > 1, + "substance_variants": [list(s) for s, _ in substances], + "count": len(items), + "in_graph": bool(covered), + "example": example, + } + + +BASE_PROMPT = ( + "Odpovedaj po slovensky a iba v rozsahu dostupných dôkazov. " + "Rozlišuj indikácie, kontraindikácie, upozornenia a interakcie. " + "Neprítomnosť informácie neznamená neprítomnosť interakcie ani bezpečnosť kombinácie. " + "Uvádzaj zdroje k tvrdeniam, ak ich kontext poskytuje. Ak dôkaz chýba, priznaj to. " + "Nehádaj silu, liekovú formu ani zloženie nejednoznačnej značky. " + "Údaje z reprezentanta rovnakej účinnej látky nemusia pokrývať konkrétny prípravok." +) + +# Reserve space for original ADC chunks after graph context and prompt overhead. +QUERY_SETTINGS = {"top_k":40,"chunk_top_k":10,"max_total_tokens":8000, + "max_entity_tokens":1000,"max_relation_tokens":1500,"enable_rerank":False} + + +def prepare_query(question, products, aliases=None, by_prefix=None): + if aliases is None or by_prefix is None: + aliases, by_prefix, _ = build_aliases(products) + matches = [describe_match(m, products) for m in detect_drugs(question, aliases, by_prefix)] + hints, notes = [], [] + for match in matches: + if match["ambiguous"]: + notes.append(f"Názov {match['label']} je nejednoznačný; vyžiadaj presný názov a zloženie.") + elif match["substances"]: + hints.append(f"{match['label']} = účinná látka {', '.join(match['substances'])}") + if not match["in_graph"]: + ex = match["example"] + notes.append(f"Pre {ex['name']} nie je v indexe reprezentant s klinickými údajmi; " + "príčinou môže byť chýbajúci leták, parsovanie alebo výber dokumentov. " + f"Základný údaj ADC: {ex.get('ind','')}. Chýbajúce údaje neznamenajú bezpečnosť.") + query = question + ("\n\n(Lieky v otázke: " + "; ".join(hints) + ")" if hints else "") + return {"query":query,"matches":matches,"user_prompt":BASE_PROMPT + "\n" + "\n".join(notes)} diff --git a/scripts/demo/build_product_index.py b/scripts/demo/build_product_index.py new file mode 100644 index 0000000..5615d70 --- /dev/null +++ b/scripts/demo/build_product_index.py @@ -0,0 +1,87 @@ +"""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()