""" create_report.py Verze: 1.6 Datum: 2026-06-02 Generuje Excel report (.xlsm) pro studii 77242113UCO3001 z MongoDB databáze Clario. Výstup: U:/Dropbox/!!!Days/Downloads Z230/YYYY-MM-DD 77242113UCO3001 Clario Reports.xlsm Zdroj dat: MongoDB 192.168.1.76, databáze Clario Kolekce Clario.MayoScore — skóre Mayo per pacient × visit Kolekce Clario.MayoDiary — denní záznamy deníku pacienta Kolekce Clario.eCOA_DCRs — data correction requests eCOA Kolekce Clario.ECG_DCRs — data correction requests ECG Listy: MayoScore — jeden řádek = pacient × visit sloupec „KLIKNI SEM" naviguje na filtrovaný EligibleDays řádky I-0 s Modified Mayo < 5 červeně tučně MayoDiary — jeden řádek = denní záznam deníku pacienta Compliance — jeden řádek = pacient × visit; kolik dní v okně mezi návštěvami mělo být vyplněno v MayoDiary a kolik jich pacient skutečně vyplnil + procento. Okno I-0 = od první diary po I-0; ostatní = od (předchozí visit +1) po aktuální visit. Unscheduled se ignorují. Řádky s compliance ≥ 100 % zeleně. EligibleDays — jeden řádek = jeden eligible day z MayoScore obohacený o data z MayoDiary; included/excluded flag, excluded dny šedě na žlutém pozadí eCOA_DCRs — všechna pole z kolekce Clario.eCOA_DCRs ECG_DCRs — všechna pole z kolekce Clario.ECG_DCRs VBA makro (Worksheet_SelectionChange na listu MayoScore): Klik na sloupec „KLIKNI SEM" → přepne na EligibleDays a vyfiltruje záznamy pro daného pacienta a visit. Vyžaduje povolení maker při otevření souboru. """ VERSION = "1.7" from datetime import datetime, timedelta from pathlib import Path import time from pymongo import MongoClient from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter import xlwings as xw # --------------------------------------------------------------------------- # Konfigurace # --------------------------------------------------------------------------- MONGO_URI = "mongodb://192.168.1.76:27017" DB_NAME = "Clario" OUTPUT_DIR = Path(r"U:\Dropbox\!!!Days\Downloads Z230") VISIT_ORDER = ["I-0", "I-2", "I-4", "I-8", "I-12", "M-4"] COLUMNS_SCORE = [ ("KLIKNI SEM", lambda d: "▶ klikni sem"), ("Site", lambda d: d.get("site", {}).get("name", "")), ("Subject ID", lambda d: d.get("subject", {}).get("id", "")), ("Visit", lambda d: d["fields"].get("Visit", "")), ("Visit Date", lambda d: d["fields"].get("Visit Date", "")), ("Baseline Stool Frequency", lambda d: _num(d["fields"].get("Baseline Stool Frequency", ""))), ("Central Endoscopy Score", lambda d: _num(d["fields"].get("Central Endoscopy Score", ""))), ("PGA Score", lambda d: _num(d["fields"].get("PGA Score", ""))), ("Stool Frequency Sub-score", lambda d: _num(d["fields"].get("Stool Frequency Sub-score", ""))), ("Rectal Bleeding Sub-score", lambda d: _num(d["fields"].get("Rectal Bleeding Sub-score", ""))), ("Partial Mayo Score", lambda d: _num(d["fields"].get("Partial Mayo Score", ""))), ("Modified Mayo Score", lambda d: _num(d["fields"].get("Modified Mayo Score", ""))), ("Full Mayo Score", lambda d: _num(d["fields"].get("Full Mayo Score", ""))), ("Site Action", lambda d: d.get("Site Action") or ""), ("Last Mayo Score Submission", lambda d: d.get("Last Mayo Score Submission") or ""), ("Wk I-12 Responder", lambda d: d.get("Week I-12 Clinical Responder") or ""), ("Wk I-12 Remission", lambda d: d.get("Week I-12 Clinical Remission") or ""), ("Clinical Flare", lambda d: d.get("Clinical Flare") or ""), ("Loss of Response", lambda d: d.get("Loss of Response") or ""), ("Partial Mayo Post LoR", lambda d: d.get("Partial Mayo Response Post Loss of Response") or ""), ("Partial Mayo Non-Resp", lambda d: d.get("Partial Mayo Response for Clinical Non-Responders") or ""), ] COLUMNS_DIARY = [ ("Subject ID", lambda d: d.get("subject", {}).get("id", "")), ("Report Date", lambda d: d["fields"].get("Report Date", "")), ("Baseline Stool Count", lambda d: _num(d["fields"].get("Baseline Stool Count", ""))), ("Stool Frequency", lambda d: _num(d["fields"].get("Stool Frequency", ""))), ("MAYO050", lambda d: d["fields"].get("MAYO050", "")), ("Not Applicable", lambda d: d["fields"].get("Not Applicable", "")), ("Constipation", lambda d: d["fields"].get("Constipation", "")), ("Diarrhea", lambda d: d["fields"].get("Diarrhea", "")), ("Irregularity", lambda d: d["fields"].get("Irregularity", "")), ] COLUMNS_ECOA_DCRS = [ ("Site", lambda d: d.get("site", {}).get("name", "")), ("Subject ID", lambda d: d.get("subject", {}).get("id", "")), ("Data Correction ID", lambda d: d["fields"].get("Data Correction ID", "")), ("PI Name", lambda d: d["fields"].get("PI Name", "")), ("Creation Date UTC", lambda d: d["fields"].get("Creation Date UTC", "")), ("Date of Last Action UTC", lambda d: d["fields"].get("Date of Last Action UTC", "")), ("Status", lambda d: d["fields"].get("Status", "")), ("Type", lambda d: d["fields"].get("Type", "")), ("Next Action Required", lambda d: d["fields"].get("Next Action Required", "")), ("Category", lambda d: d["fields"].get("Category", "")), ("Total Open Period", lambda d: d["fields"].get("Total Open Period", "")), ("Total Open Time (Days)", lambda d: _num(d["fields"].get("Total Open Time (Days)", ""))), ("Current Status Time (Days)", lambda d: _num(d["fields"].get("Current Status Time (Days)", ""))), ("Reason for Change", lambda d: d["fields"].get("Reason for Change", "")), ("Description", lambda d: d["fields"].get("Description", "")), ("Resolution", lambda d: d["fields"].get("Resolution", "")), ("Query History", lambda d: d["fields"].get("Query History", "")), ("Age at Informed Consent", lambda d: d["fields"].get("Age at Informed Consent", "")), ("Baseline Stool Count", lambda d: _num(d["fields"].get("Baseline Stool Count", ""))), ("firstSeen", lambda d: d.get("firstSeen", "")), ("lastSeen", lambda d: d.get("lastSeen", "")), ] COLUMNS_ECG_DCRS = [ ("Site ID", lambda d: d.get("site", {}).get("name", "")), ("Subject Number", lambda d: d.get("subject", {}).get("id", "")), ("Data Correction ID", lambda d: d["fields"].get("Data Correction ID", "")), ("PI Name", lambda d: d["fields"].get("PI_NAME", "")), ("Age", lambda d: d["fields"].get("Age", "")), ("Creation Date UTC", lambda d: d["fields"].get("Creation Date UTC", "")), ("Date of Last Action UTC", lambda d: d["fields"].get("Date of Last Action UTC", "")), ("Status", lambda d: d["fields"].get("Status", "")), ("Type", lambda d: d["fields"].get("Type", "")), ("Next Action Required", lambda d: d["fields"].get("Next Action Required", "")), ("Category", lambda d: d["fields"].get("Category", "")), ("Total Open Period", lambda d: d["fields"].get("Total Open Period", "")), ("Total Open Time (Days)", lambda d: _num(d["fields"].get("Total Open Time (Days)", ""))), ("Current Status Time (Days)", lambda d: _num(d["fields"].get("Current Status Time (Days)", ""))), ("Reason for Change", lambda d: d["fields"].get("Reason for Change", "")), ("Query History", lambda d: d["fields"].get("Query History", "")), ("firstSeen", lambda d: d.get("firstSeen", "")), ("lastSeen", lambda d: d.get("lastSeen", "")), ] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _num(value): """Převede číselný string na int, jinak vrátí původní hodnotu nebo None.""" if value == "" or value is None: return None try: return int(value) except (ValueError, TypeError): try: return float(value) except (ValueError, TypeError): return value def _visit_sort_key(doc): visit = doc["fields"].get("Visit", "") try: idx = VISIT_ORDER.index(visit) except ValueError: idx = len(VISIT_ORDER) return (doc.get("site", {}).get("name", ""), doc.get("subject", {}).get("id", ""), idx, visit) def _iso_to_date(value): """ISO string → Python date pro Excel.""" if not isinstance(value, str): return value try: return datetime.fromisoformat(value).date() except ValueError: return value # --------------------------------------------------------------------------- # Styly # --------------------------------------------------------------------------- HEADER_FILL = PatternFill("solid", fgColor="1F497D") HEADER_FONT = Font(bold=True, color="FFFFFF", size=10) CELL_FONT = Font(size=10) ALIGN_CTR = Alignment(horizontal="center", vertical="center", wrap_text=False) ALIGN_LEFT = Alignment(horizontal="left", vertical="center") THIN = Side(style="thin", color="BFBFBF") BORDER = Border(left=THIN, right=THIN, top=THIN, bottom=THIN) # zebra FILL_ODD = PatternFill("solid", fgColor="FFFFFF") FILL_EVEN = PatternFill("solid", fgColor="EBF1DE") # DCR status barvy FILL_DCR_SITE = PatternFill("solid", fgColor="FFFF00") # žlutá — čeká lékař FILL_DCR_CLARIO = PatternFill("solid", fgColor="BDD7EE") # modrá — čeká Clario FILL_DCR_QC = PatternFill("solid", fgColor="F4B942") # oranžová — ReadyForQC FILL_DCR_DONE = PatternFill("solid", fgColor="FFFFFF") # bílá — Completed SCORE_COLS = {"Partial Mayo Score", "Modified Mayo Score", "Full Mayo Score"} SCORE_FILL = PatternFill("solid", fgColor="FFC7CE") # červená pro skóre ≥ 5 (placeholder — nepoužíváme podmíněné formátování) # --------------------------------------------------------------------------- # Sestavení sheetu # --------------------------------------------------------------------------- def _build_sheet(ws, docs, columns, date_cols, center_cols, col_widths, row_font_fn=None, wrap_cols=None, header_row=1): headers = [c[0] for c in columns] for col_idx, header in enumerate(headers, 1): cell = ws.cell(row=header_row, column=col_idx, value=header) cell.font = HEADER_FONT cell.fill = HEADER_FILL cell.alignment = ALIGN_CTR cell.border = BORDER ws.row_dimensions[header_row].height = 28 data_start = header_row + 1 for row_idx, doc in enumerate(docs, data_start): fill = FILL_EVEN if (row_idx - header_row) % 2 == 0 else FILL_ODD font = row_font_fn(doc) if row_font_fn else CELL_FONT for col_idx, (col_name, getter) in enumerate(columns, 1): value = getter(doc) if col_name in date_cols and isinstance(value, str): value = _iso_to_date(value) cell = ws.cell(row=row_idx, column=col_idx, value=value) cell.font = font cell.fill = fill cell.border = BORDER if wrap_cols and col_name in wrap_cols: cell.alignment = Alignment(horizontal="left", vertical="top", wrap_text=True) else: cell.alignment = ALIGN_CTR if col_name in center_cols else ALIGN_LEFT for col_idx, (col_name, _) in enumerate(columns, 1): ws.column_dimensions[get_column_letter(col_idx)].width = col_widths.get(col_name, 14) for col_name in date_cols: if col_name in headers: letter = get_column_letter(headers.index(col_name) + 1) for row_idx in range(data_start, len(docs) + data_start): ws[f"{letter}{row_idx}"].number_format = "DD-MMM-YYYY" ws.freeze_panes = f"A{data_start}" ws.auto_filter.ref = f"A{header_row}:{get_column_letter(len(headers))}{header_row}" def _score_row_font(doc): visit = doc["fields"].get("Visit", "") try: mod_mayo = int(doc["fields"].get("Modified Mayo Score", "")) except (ValueError, TypeError): mod_mayo = None if visit == "I-0" and mod_mayo is not None and mod_mayo < 5: return Font(size=10, bold=True, color="FF0000") return CELL_FONT def build_mayo_score_sheet(ws, docs): _build_sheet( ws, docs, COLUMNS_SCORE, date_cols={"Visit Date", "Last Mayo Score Submission"}, center_cols={"KLIKNI SEM", "Visit", "Central Endoscopy Score", "PGA Score", "Stool Frequency Sub-score", "Rectal Bleeding Sub-score", "Partial Mayo Score", "Modified Mayo Score", "Full Mayo Score", "Baseline Stool Frequency", "Wk I-12 Responder", "Wk I-12 Remission", "Clinical Flare", "Loss of Response", "Partial Mayo Post LoR", "Partial Mayo Non-Resp", "Last Mayo Score Submission"}, col_widths={ "KLIKNI SEM": 14, "Site": 18, "Subject ID": 16, "Visit": 12, "Visit Date": 14, "Baseline Stool Frequency": 14, "Central Endoscopy Score": 14, "PGA Score": 10, "Stool Frequency Sub-score": 14, "Rectal Bleeding Sub-score": 14, "Partial Mayo Score": 14, "Modified Mayo Score": 14, "Full Mayo Score": 13, "Site Action": 22, "Last Mayo Score Submission": 16, "Wk I-12 Responder": 14, "Wk I-12 Remission": 14, "Clinical Flare": 14, "Loss of Response": 14, "Partial Mayo Post LoR": 20, "Partial Mayo Non-Resp": 20, }, row_font_fn=_score_row_font, ) # Speciální styl pro sloupec KLIKNI SEM — vypadá jako tlačítko/odkaz link_font = Font(size=10, bold=True, color="FFFFFF") link_fill = PatternFill("solid", fgColor="2E75B6") for row in range(2, len(docs) + 2): cell = ws.cell(row=row, column=1) cell.font = link_font cell.fill = link_fill cell.alignment = ALIGN_CTR def build_mayo_diary_sheet(ws, docs): _build_sheet( ws, docs, COLUMNS_DIARY, date_cols={"Report Date"}, center_cols={"Baseline Stool Count", "Stool Frequency", "Not Applicable", "Constipation", "Diarrhea", "Irregularity"}, col_widths={ "Subject ID": 16, "Report Date": 14, "Baseline Stool Count": 14, "Stool Frequency": 14, "MAYO050": 48, "Not Applicable": 14, "Constipation": 14, "Diarrhea": 12, "Irregularity": 14, }, ) def build_eligible_days_sheet(ws, score_docs, diary_docs): # Lookup diary records by (subject_id, date_part YYYY-MM-DD) diary_lookup: dict[tuple, dict] = {} for d in diary_docs: subj = d.get("subject", {}).get("id", "") date_iso = d["fields"].get("Report Date", "") date_part = date_iso[:10] if date_iso else "" if subj and date_part: diary_lookup[(subj, date_part)] = d headers = [ "Included", "Subject ID", "Visit", "Visit Date", "Day", "Report Date", "Baseline Stool Count", "Stool Frequency", "MAYO050", "Not Applicable", "Constipation", "Diarrhea", "Irregularity", ] col_widths = { "Included": 10, "Subject ID": 16, "Visit": 10, "Visit Date": 14, "Day": 8, "Report Date": 14, "Baseline Stool Count": 14, "Stool Frequency": 14, "MAYO050": 48, "Not Applicable": 14, "Constipation": 14, "Diarrhea": 12, "Irregularity": 14, } center_cols = {"Included", "Visit", "Day", "Baseline Stool Count", "Stool Frequency", "Not Applicable", "Constipation", "Diarrhea", "Irregularity"} date_cols = {"Visit Date", "Report Date"} no_fill = PatternFill("solid", fgColor="FFF2CC") # žlutá pro excluded dny for col_idx, header in enumerate(headers, 1): cell = ws.cell(row=1, column=col_idx, value=header) cell.font = HEADER_FONT cell.fill = HEADER_FILL cell.alignment = ALIGN_CTR cell.border = BORDER ws.row_dimensions[1].height = 28 row_idx = 2 for score_doc in score_docs: subj = score_doc.get("subject", {}).get("id", "") visit = score_doc["fields"].get("Visit", "") visit_date = score_doc["fields"].get("Visit Date", "") for n in range(1, 11): day_date_iso = score_doc["fields"].get(f"Eligible Day (-{n})") if not day_date_iso or day_date_iso == "-": continue date_part = day_date_iso[:10] excl_reason = score_doc["fields"].get(f"Day (-{n}) Excluded Reason(s)", "") included = "No" if excl_reason and excl_reason != "-" else "Yes" diary = diary_lookup.get((subj, date_part), {}) df = diary.get("fields", {}) fill = no_fill if included == "No" else (FILL_EVEN if row_idx % 2 == 0 else FILL_ODD) font = Font(size=10, color="808080") if included == "No" else CELL_FONT values = [ included, subj, visit, _iso_to_date(visit_date) if isinstance(visit_date, str) else visit_date, f"-{n}", _iso_to_date(day_date_iso), _num(df.get("Baseline Stool Count", "")), _num(df.get("Stool Frequency", "")), df.get("MAYO050", ""), df.get("Not Applicable", ""), df.get("Constipation", ""), df.get("Diarrhea", ""), df.get("Irregularity", ""), ] for col_idx, (header, value) in enumerate(zip(headers, values), 1): cell = ws.cell(row=row_idx, column=col_idx, value=value) cell.font = font cell.fill = fill cell.border = BORDER if header in date_cols: cell.number_format = "DD-MMM-YYYY" cell.alignment = ALIGN_CTR if header in center_cols else ALIGN_LEFT row_idx += 1 for col_idx, header in enumerate(headers, 1): ws.column_dimensions[get_column_letter(col_idx)].width = col_widths.get(header, 14) ws.freeze_panes = "A2" ws.auto_filter.ref = f"A1:{get_column_letter(len(headers))}1" def _build_dcr_legend(ws): """Vloží legendu do řádků 1–4, prázdný řádek 5. Data začínají od řádku 6.""" legend = [ (FILL_DCR_SITE, "Čeká lékař — Next Action Required = Site (lékař musí odpovědět nebo potvrdit)"), (FILL_DCR_CLARIO, "Čeká Clario — Next Action Required = Clario DM (Clario dostalo podklady, provede změnu)"), (FILL_DCR_QC, "ReadyForQC — Clario provedlo změny, čeká na finální QC kontrolu"), (FILL_DCR_DONE, "Completed / Resolved — DCR je uzavřen"), ] for i, (fill, text) in enumerate(legend, 1): a = ws.cell(row=i, column=1, value="") a.fill = fill a.border = BORDER b = ws.cell(row=i, column=2, value=text) b.font = Font(size=10, bold=True) b.alignment = ALIGN_LEFT # řádek 5 prázdný — nic nedělat def _dcr_row_fill(doc): """Vrátí fill barvu dle stavu DCR.""" status = doc["fields"].get("Status", "") next_action = doc["fields"].get("Next Action Required", "") if status in ("Completed", "Resolved"): return FILL_DCR_DONE if status == "ReadyForQC": return FILL_DCR_QC if "Site" in next_action: return FILL_DCR_SITE if "Clario" in next_action or next_action == "": return FILL_DCR_CLARIO return FILL_ODD def build_ecoa_dcrs_sheet(ws, docs): _build_dcr_legend(ws) docs_sorted = sorted(docs, key=lambda d: ( d.get("site", {}).get("name", ""), d.get("subject", {}).get("id", ""), d["fields"].get("Creation Date UTC", ""), )) _build_sheet( ws, docs_sorted, COLUMNS_ECOA_DCRS, date_cols={"Creation Date UTC", "Date of Last Action UTC"}, center_cols={"Status", "Type", "Next Action Required", "Category", "Total Open Time (Days)", "Current Status Time (Days)", "Baseline Stool Count", "firstSeen", "lastSeen"}, col_widths={ "Site": 16, "Subject ID": 16, "Data Correction ID": 18, "PI Name": 18, "Creation Date UTC": 14, "Date of Last Action UTC": 14, "Status": 14, "Type": 16, "Next Action Required": 16, "Category": 20, "Total Open Period": 14, "Total Open Time (Days)": 14, "Current Status Time (Days)": 16, "Reason for Change": 20, "Description": 50, "Resolution": 50, "Query History": 60, "Age at Informed Consent": 14, "Baseline Stool Count": 14, "firstSeen": 12, "lastSeen": 12, }, wrap_cols={"Reason for Change", "Description", "Resolution", "Query History"}, header_row=6, row_font_fn=lambda doc: CELL_FONT, ) # Přebarvení řádků dle DCR stavu (přepíše zebra fill) data_start = 7 for row_idx, doc in enumerate(docs_sorted, data_start): fill = _dcr_row_fill(doc) for col_idx in range(1, len(COLUMNS_ECOA_DCRS) + 1): ws.cell(row=row_idx, column=col_idx).fill = fill def build_ecg_dcrs_sheet(ws, docs): _build_dcr_legend(ws) docs_sorted = sorted(docs, key=lambda d: ( d.get("site", {}).get("name", ""), d.get("subject", {}).get("id", ""), d["fields"].get("Creation Date UTC", ""), )) _build_sheet( ws, docs_sorted, COLUMNS_ECG_DCRS, date_cols={"Creation Date UTC", "Date of Last Action UTC"}, center_cols={"Status", "Type", "Next Action Required", "Category", "Total Open Time (Days)", "Current Status Time (Days)", "firstSeen", "lastSeen"}, col_widths={ "Site ID": 14, "Subject Number": 16, "Data Correction ID": 16, "PI Name": 18, "Age": 10, "Creation Date UTC": 14, "Date of Last Action UTC": 14, "Status": 14, "Type": 12, "Next Action Required": 16, "Category": 14, "Total Open Period": 14, "Total Open Time (Days)": 14, "Current Status Time (Days)": 16, "Reason for Change": 20, "Query History": 60, "firstSeen": 12, "lastSeen": 12, }, wrap_cols={"Query History"}, header_row=6, row_font_fn=lambda doc: CELL_FONT, ) # Přebarvení řádků dle DCR stavu data_start = 7 for row_idx, doc in enumerate(docs_sorted, data_start): fill = _dcr_row_fill(doc) for col_idx in range(1, len(COLUMNS_ECG_DCRS) + 1): ws.cell(row=row_idx, column=col_idx).fill = fill # --------------------------------------------------------------------------- # List Compliance # --------------------------------------------------------------------------- # Pořadí návštěv pro výpočet oken (Unscheduled apod. se ignorují) COMPLIANCE_VISIT_ORDER = ["I-0", "I-2", "I-4", "I-8", "I-12", "M-4"] FILL_COMPLIANCE_OK = PatternFill("solid", fgColor="C6EFCE") # zelená — compliance ≥ 100 % FONT_COMPLIANCE_OK = Font(size=10, color="006100") def build_compliance_sheet(ws, score_docs, diary_docs): """Compliance vyplňování MayoDiary mezi návštěvami. Okno pro I-0 = od první MayoDiary daného pacienta po datum I-0. Okno pro ostatní = od (datum předchozí návštěvy + 1 den) po datum návštěvy. Vyplněno = počet MayoDiary záznamů pacienta s Report Date uvnitř okna. Dní v okně = počet kalendářních dní okna (včetně obou krajů). """ # -- MayoDiary datumy per pacient (jeden průchod) ------------------------ diary_by_subj: dict[str, list] = {} for d in diary_docs: subj = d.get("subject", {}).get("id", "") rd = d["fields"].get("Report Date", "") dt = _iso_to_date(rd) if isinstance(rd, str) else rd if subj and hasattr(dt, "year"): diary_by_subj.setdefault(subj, []).append(dt) first_diary = {s: min(dts) for s, dts in diary_by_subj.items() if dts} def _vidx(v): try: return COMPLIANCE_VISIT_ORDER.index(v) except ValueError: return len(COMPLIANCE_VISIT_ORDER) # -- Návštěvy per pacient (jen známé visity) ----------------------------- by_subj: dict[str, list] = {} for sd in score_docs: if sd["fields"].get("Visit", "") not in COMPLIANCE_VISIT_ORDER: continue subj = sd.get("subject", {}).get("id", "") by_subj.setdefault(subj, []).append(sd) rows = [] for subj in sorted(by_subj): visits = sorted(by_subj[subj], key=lambda d: _vidx(d["fields"].get("Visit", ""))) prev_end = None for sd in visits: visit = sd["fields"].get("Visit", "") vdate = _iso_to_date(sd["fields"].get("Visit Date", "")) if not hasattr(vdate, "year"): continue if visit == "I-0": start = first_diary.get(subj) else: start = (prev_end + timedelta(days=1)) if prev_end else first_diary.get(subj) prev_end = vdate if not start or not hasattr(start, "year"): continue days = (vdate - start).days + 1 if days <= 0: continue filled = sum(1 for dt in diary_by_subj.get(subj, []) if start <= dt <= vdate) pct = round(filled / days * 100) rows.append({ "site": sd.get("site", {}).get("name", ""), "subj": subj, "visit": visit, "start": start, "end": vdate, "days": days, "filled": filled, "pct": pct, }) # -- Zápis listu --------------------------------------------------------- headers = ["Site", "Subject ID", "Visit", "Okno od", "Okno do", "Dní v okně", "Vyplněno", "Compliance %"] col_widths = {"Site": 18, "Subject ID": 16, "Visit": 10, "Okno od": 14, "Okno do": 14, "Dní v okně": 12, "Vyplněno": 12, "Compliance %": 14} center_cols = {"Visit", "Dní v okně", "Vyplněno", "Compliance %"} date_cols = {"Okno od", "Okno do"} for col_idx, header in enumerate(headers, 1): cell = ws.cell(row=1, column=col_idx, value=header) cell.font = HEADER_FONT cell.fill = HEADER_FILL cell.alignment = ALIGN_CTR cell.border = BORDER ws.row_dimensions[1].height = 28 for row_idx, r in enumerate(rows, 2): is_ok = r["pct"] >= 100 if is_ok: fill = FILL_COMPLIANCE_OK font = FONT_COMPLIANCE_OK else: fill = FILL_EVEN if row_idx % 2 == 0 else FILL_ODD font = CELL_FONT values = [r["site"], r["subj"], r["visit"], r["start"], r["end"], r["days"], r["filled"], r["pct"]] for col_idx, (header, value) in enumerate(zip(headers, values), 1): cell = ws.cell(row=row_idx, column=col_idx, value=value) cell.font = font cell.fill = fill cell.border = BORDER if header in date_cols: cell.number_format = "DD-MMM-YYYY" if header == "Compliance %": cell.number_format = '0"%"' cell.alignment = ALIGN_CTR if header in center_cols else ALIGN_LEFT for col_idx, header in enumerate(headers, 1): ws.column_dimensions[get_column_letter(col_idx)].width = col_widths.get(header, 14) ws.freeze_panes = "A2" ws.auto_filter.ref = f"A1:{get_column_letter(len(headers))}1" # --------------------------------------------------------------------------- # Helpers: výstupní cesta # --------------------------------------------------------------------------- def _unique_path(directory: Path, stem: str, suffix: str) -> Path: candidate = directory / f"{stem}{suffix}" if not candidate.exists(): return candidate n = 2 while True: candidate = directory / f"{stem} ({n}){suffix}" if not candidate.exists(): return candidate n += 1 # --------------------------------------------------------------------------- # Timing helper # --------------------------------------------------------------------------- def _tick(label: str, t0: float) -> float: """Vypíše dobu od t0 a vrátí aktuální čas jako nový t0.""" elapsed = time.perf_counter() - t0 print(f" {label:<30} {elapsed:6.2f} s") return time.perf_counter() # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): t_total = time.perf_counter() print("Spouštím generování reportu...") print() # -- 1. MongoDB: připojení + načtení + seřazení -------------------------- t = time.perf_counter() client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=5000) client.admin.command("ping") db = client[DB_NAME] score_docs = list(db["Clario.MayoScore"].find({})) diary_docs = list(db["Clario.MayoDiary"].find({})) ecoa_dcr_docs = list(db["Clario.eCOA_DCRs"].find({})) ecg_dcr_docs = list(db["Clario.ECG_DCRs"].find({})) client.close() score_docs.sort(key=_visit_sort_key) diary_docs.sort(key=lambda d: ( d.get("subject", {}).get("id", ""), d["fields"].get("Report Date", ""), )) t = _tick(f"MongoDB (ping, fetch, sort → {len(score_docs)} + {len(diary_docs)} + {len(ecoa_dcr_docs)} + {len(ecg_dcr_docs)} záznamů)", t) # -- 2–4. Tvorba listů --------------------------------------------------- wb = Workbook() ws_score = wb.active ws_score.title = "MayoScore" build_mayo_score_sheet(ws_score, score_docs) t = _tick("List MayoScore (KLIKNI SEM, zebra, červené I-0, autofilter)", t) ws_diary = wb.create_sheet("MayoDiary") build_mayo_diary_sheet(ws_diary, diary_docs) t = _tick("List MayoDiary (zebra, formátování dat, autofilter)", t) ws_comp = wb.create_sheet("Compliance") build_compliance_sheet(ws_comp, score_docs, diary_docs) t = _tick("List Compliance (okna mezi visitami, % vyplnění, zelená ≥100 %)", t) ws_days = wb.create_sheet("EligibleDays") build_eligible_days_sheet(ws_days, score_docs, diary_docs) t = _tick("List EligibleDays (diary lookup, included/excluded flag, autofilter)", t) ws_ecoa = wb.create_sheet("eCOA_DCRs") build_ecoa_dcrs_sheet(ws_ecoa, ecoa_dcr_docs) t = _tick(f"List eCOA_DCRs ({len(ecoa_dcr_docs)} záznamů)", t) ws_ecg = wb.create_sheet("ECG_DCRs") build_ecg_dcrs_sheet(ws_ecg, ecg_dcr_docs) t = _tick(f"List ECG_DCRs ({len(ecg_dcr_docs)} záznamů)", t) # -- 5. Uložení XLSX ----------------------------------------------------- OUTPUT_DIR.mkdir(parents=True, exist_ok=True) today = datetime.now().strftime("%Y-%m-%d") base_stem = f"{today} 77242113UCO3001 Clario Reports" xlsm_path = _unique_path(OUTPUT_DIR, base_stem, ".xlsm") xlsx_path = xlsm_path.with_suffix(".xlsx") wb.save(str(xlsx_path)) t = _tick("Uložení XLSX (openpyxl, dočasný soubor)", t) # -- 6. Injektování VBA -------------------------------------------------- inject_vba(xlsx_path, xlsm_path) xlsx_path.unlink(missing_ok=True) _tick("Injektování VBA (xlwings: open → AddFromString → SaveAs .xlsm)", t) # -- Souhrn -------------------------------------------------------------- total = time.perf_counter() - t_total print() print(f" {'Celkem':<30} {total:6.2f} s") print() print(f"Uloženo: {xlsm_path}") def inject_vba(xlsx_path: Path, xlsm_path: Path) -> None: vba_code = '''\ Private Sub Worksheet_SelectionChange(ByVal Target As Range) If Target.Row < 2 Then Exit Sub If Target.Rows.Count > 1 Then Exit Sub If Target.Column <> 1 Then Exit Sub Dim subjectId As String Dim visit As String subjectId = CStr(Me.Cells(Target.Row, 3).Value) visit = CStr(Me.Cells(Target.Row, 4).Value) If subjectId = "" Or visit = "" Then Exit Sub Dim ws As Worksheet On Error Resume Next Set ws = ThisWorkbook.Sheets("EligibleDays") On Error GoTo 0 If ws Is Nothing Then Exit Sub Application.ScreenUpdating = False ws.AutoFilterMode = False ws.Range("A1").AutoFilter ws.Range("A1").AutoFilter Field:=2, Criteria1:=subjectId ws.Range("A1").AutoFilter Field:=3, Criteria1:=visit ws.Activate ws.Range("A2").Select Application.ScreenUpdating = True End Sub ''' app = xw.App(visible=False) try: wb = app.books.open(str(xlsx_path)) # Najdi VBComponent odpovídající listu "MayoScore" podle tab názvu vb_comp = None for comp in wb.api.VBProject.VBComponents: if comp.Type == 100: # xlSheet try: if comp.Properties("Name").Value == "MayoScore": vb_comp = comp break except Exception: pass if vb_comp is None: # fallback: první sheet (Sheet1) vb_comp = wb.api.VBProject.VBComponents("Sheet1") vb_comp.CodeModule.AddFromString(vba_code) wb.api.SaveAs(str(xlsm_path), FileFormat=52) # 52 = xlOpenXMLWorkbookMacroEnabled wb.close() finally: app.quit() if __name__ == "__main__": main()