"""DSC quarterly + RMP/DSUR/PSUR docx export + manuscript supplementary. Data Safety Committee 분기 보고서·RMP·DSUR·PSUR 표준 섹션의 자동 채움. 참고용·연구용. 임상·규제 제출 전에 메디컬 라이터·QPPV 검토 필수. """ from __future__ import annotations import csv import json from dataclasses import asdict from pathlib import Path from typing import Dict, List, Optional from .hys_law import HysCase, summarize_hys from .rucam import RUCAMResult from .class_panel import PanelSignal def _safe_docx_import(): try: from docx import Document # type: ignore return Document except Exception: return None def _placebo_subtract( cases: List[HysCase], placebo_ref: Dict[str, float] ) -> Dict[str, float]: """placebo arm reference 대비 시험약 시그널 비교 (단순 차이).""" drug_cases = [c for c in cases if c.arm.lower() != "placebo"] drug_n = max(len(drug_cases), 1) drug_classical = sum(1 for c in drug_cases if c.classical_hys) / drug_n drug_temple = sum(1 for c in drug_cases if c.quadrant == "Temple_corollary") / drug_n return { "drug_classical_hys_rate": round(drug_classical, 4), "drug_temple_corollary_rate": round(drug_temple, 4), "placebo_ref_classical_hys_rate": placebo_ref.get("classical_hys_rate", 0.0), "placebo_ref_temple_corollary_rate": placebo_ref.get("temple_rate", 0.0), "attributable_classical_hys": round( drug_classical - placebo_ref.get("classical_hys_rate", 0.0), 4), "attributable_temple": round( drug_temple - placebo_ref.get("temple_rate", 0.0), 4), } def load_placebo_reference(path: str | Path) -> Dict[str, float]: """placebo arm reference CSV → summary dict.""" path = Path(path) if not path.exists(): return {"classical_hys_rate": 0.003, "temple_rate": 0.025} with path.open(newline="", encoding="utf-8") as f: reader = csv.DictReader(f) rows = list(reader) if not rows: return {"classical_hys_rate": 0.003, "temple_rate": 0.025} hys = [float(r.get("classical_hys_rate", 0) or 0) for r in rows] temple = [float(r.get("temple_rate", 0) or 0) for r in rows] return { "classical_hys_rate": sum(hys) / len(hys) if hys else 0.003, "temple_rate": sum(temple) / len(temple) if temple else 0.025, "n_trials": len(rows), } def build_dsc_report( cases: List[HysCase], rucam_results: List[RUCAMResult], class_signals: Dict[str, List[PanelSignal]], placebo_ref: Dict[str, float], quarter: str = "Q1", drug_label: str = "Study Drug", ) -> Dict: hys_summary = summarize_hys(cases) cat_counts: Dict[str, int] = {} for r in rucam_results: cat_counts[r.category] = cat_counts.get(r.category, 0) + 1 attribution = _placebo_subtract(cases, placebo_ref) flagged_class = { k: sum(1 for s in v if s.flag) for k, v in class_signals.items() } return { "quarter": quarter, "drug": drug_label, "n_patients_evaluable": len(cases), "hys_summary": hys_summary, "rucam_category_counts": cat_counts, "placebo_attribution": attribution, "class_signal_flag_counts": flagged_class, "disclaimer": ( "본 보고서는 자동화 도구로 생성된 1차 안전성 시그널 요약이며, " "확정적 인과성 판정이나 임상적 의사결정의 근거가 아니다." ), } def export_dsc_json(report: Dict, out_path: str | Path) -> Path: out_path = Path(out_path) out_path.parent.mkdir(parents=True, exist_ok=True) with out_path.open("w", encoding="utf-8") as f: json.dump(report, f, ensure_ascii=False, indent=2) return out_path def export_cases_csv(cases: List[HysCase], rucam_results: List[RUCAMResult], out_path: str | Path) -> Path: out_path = Path(out_path) out_path.parent.mkdir(parents=True, exist_ok=True) rucam_map = {r.pid: r for r in rucam_results} fields = [ "pid", "arm", "drug_class", "peak_alt", "peak_tbl", "peak_alp", "alt_x_uln", "alt_x_baseline", "tbl_x_uln", "alp_x_uln", "classical_hys", "baseline_adj_hys", "quadrant", "r_ratio", "pattern", "rucam_total", "rucam_category", "naranjo", "naranjo_category", "maria_victorino", ] with out_path.open("w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=fields) writer.writeheader() for c in cases: r = rucam_map.get(c.pid) writer.writerow({ "pid": c.pid, "arm": c.arm, "drug_class": c.drug_class, "peak_alt": round(c.peak_alt, 2), "peak_tbl": round(c.peak_tbl, 2), "peak_alp": round(c.peak_alp, 2), "alt_x_uln": round(c.alt_ratio_uln, 2), "alt_x_baseline": round(c.alt_ratio_baseline, 2), "tbl_x_uln": round(c.tbl_ratio_uln, 2), "alp_x_uln": round(c.alp_ratio_uln, 2), "classical_hys": int(c.classical_hys), "baseline_adj_hys": int(c.baseline_adj_hys), "quadrant": c.quadrant, "r_ratio": round(c.r_ratio, 2) if c.r_ratio is not None else "", "pattern": c.pattern, "rucam_total": r.total if r else "", "rucam_category": r.category if r else "", "naranjo": r.naranjo if r else "", "naranjo_category": r.naranjo_category if r else "", "maria_victorino": r.maria_victorino if r else "", }) return out_path def export_docx(report: Dict, out_path: str | Path, doc_type: str = "DSC") -> Optional[Path]: """RMP / DSUR / PSUR / DSC 표준 섹션 docx export. python-docx 미설치시 markdown fallback. """ out_path = Path(out_path) out_path.parent.mkdir(parents=True, exist_ok=True) Document = _safe_docx_import() if Document is None: md_path = out_path.with_suffix(".md") _export_md_fallback(report, md_path, doc_type) return md_path doc = Document() title_map = { "DSC": "Data Safety Committee Quarterly Report", "RMP": "Risk Management Plan - Hepatic Safety Module", "DSUR": "Development Safety Update Report", "PSUR": "Periodic Safety Update Report (PBRER)", } doc.add_heading(title_map.get(doc_type, "Safety Report"), 0) doc.add_paragraph(f"Drug: {report.get('drug')} Quarter: {report.get('quarter')}") doc.add_paragraph(report.get("disclaimer", "")) doc.add_heading("1. Evaluable population", 1) doc.add_paragraph(f"N = {report.get('n_patients_evaluable')}") doc.add_heading("2. Hy's law analysis", 1) for k, v in report.get("hys_summary", {}).items(): doc.add_paragraph(f" - {k}: {v}", style=None) doc.add_heading("3. RUCAM causality distribution", 1) for k, v in report.get("rucam_category_counts", {}).items(): doc.add_paragraph(f" - {k}: {v}") doc.add_heading("4. Placebo-attributable signal", 1) for k, v in report.get("placebo_attribution", {}).items(): doc.add_paragraph(f" - {k}: {v}") doc.add_heading("5. Class-effect safety panel flags", 1) for k, v in report.get("class_signal_flag_counts", {}).items(): doc.add_paragraph(f" - {k}: {v} flagged signal(s)") doc.add_heading("6. Recommendations", 1) doc.add_paragraph( "각 probable/highly probable RUCAM case는 사이트 PI와 메디컬 모니터의 " "case-by-case adjudication을 거쳐 SUSAR/SAE 보고 여부를 결정한다." ) doc.save(str(out_path)) return out_path def _export_md_fallback(report: Dict, out_path: Path, doc_type: str) -> None: lines = [f"# {doc_type} Report", f"- Drug: {report.get('drug')}", f"- Quarter: {report.get('quarter')}", f"- Disclaimer: {report.get('disclaimer')}", "", "## Hy's law summary"] for k, v in report.get("hys_summary", {}).items(): lines.append(f"- {k}: {v}") lines.append("\n## RUCAM categories") for k, v in report.get("rucam_category_counts", {}).items(): lines.append(f"- {k}: {v}") lines.append("\n## Placebo attribution") for k, v in report.get("placebo_attribution", {}).items(): lines.append(f"- {k}: {v}") lines.append("\n## Class-effect flags") for k, v in report.get("class_signal_flag_counts", {}).items(): lines.append(f"- {k}: {v}") out_path.write_text("\n".join(lines), encoding="utf-8") def build_manuscript_supplement( cases: List[HysCase], rucam_results: List[RUCAMResult], class_signals: Dict[str, List[PanelSignal]], out_path: str | Path, ) -> Path: """manuscript supplementary (markdown) — eDISH 표·RUCAM 일치율·class별 패널 요약.""" out_path = Path(out_path) out_path.parent.mkdir(parents=True, exist_ok=True) hys = summarize_hys(cases) naranjo_agree = sum( 1 for r in rucam_results if r.category in {"probable", "highly_probable"} and (r.naranjo_category or "") in {"probable", "definite"} ) total_probable = sum( 1 for r in rucam_results if r.category in {"probable", "highly_probable"}) agree_pct = (naranjo_agree / total_probable * 100) if total_probable else 0.0 lines = [ "# Supplementary: MASH-DILISurveil-Kor analysis", "", "## S1. eDISH summary", ] for k, v in hys.items(): lines.append(f"- {k}: {v}") lines.extend([ "", "## S2. RUCAM vs Naranjo agreement", f"- probable+ RUCAM cases: {total_probable}", f"- Naranjo probable+ in same: {naranjo_agree}", f"- agreement: {agree_pct:.1f}%", "", "## S3. Class-effect signals per panel", ]) for cls, sigs in class_signals.items(): flagged = [s for s in sigs if s.flag] lines.append(f"### {cls}") if not flagged: lines.append("- (no flagged signal)") for s in flagged[:20]: lines.append( f"- {s.pid} {s.marker} ({s.role}): " f"{s.pct_change:.1f}% — rule: {s.rule}" ) out_path.write_text("\n".join(lines), encoding="utf-8") return out_path