import gradio as gr import pandas as pd import numpy as np from datetime import datetime, timedelta import os import io def process_files(thuoc_file, xetnghiem_file, thuoc_plt_file, icd_plt_file): if not all([thuoc_file, xetnghiem_file, thuoc_plt_file, icd_plt_file]): return "Vui lòng upload đầy đủ các file trước khi chạy.", None, None, None, None, None, None # Đọc file từ temp paths df_thuoc = pd.read_excel(thuoc_file.name, sheet_name='Sheet1') df_xetnghiem = pd.read_excel(xetnghiem_file.name, sheet_name='Sheet 1') df_thuoc_plt = pd.read_excel(thuoc_plt_file.name, sheet_name='Sheet1') df_icd_plt = pd.read_excel(icd_plt_file.name, sheet_name='Sheet1') # Code chuẩn hóa def to_str(df, column): if column in df.columns: return df[column].astype(str).str.replace(r'\.0$', '', regex=True).str.strip() return df df_thuoc['makcb'] = to_str(df_thuoc, 'makcb') df_thuoc['mabhyt'] = to_str(df_thuoc, 'mabhyt') df_xetnghiem['Mã KCB'] = to_str(df_xetnghiem, 'Mã KCB') df_thuoc_plt['mabhyt'] = to_str(df_thuoc_plt, 'mabhyt') df_icd_plt['maicd'] = to_str(df_icd_plt, 'maicd').str.upper() df_thuoc['ngaychidinh'] = pd.to_datetime(df_thuoc['ngaychidinh'], dayfirst=True, errors='coerce', format='mixed') df_xetnghiem['Ngày KQ'] = pd.to_datetime(df_xetnghiem['Ngày KQ'], dayfirst=True, errors='coerce', format='mixed') exclude_icds = set(df_icd_plt['maicd'].unique()) def check_exclusion(icd_string): if pd.isna(icd_string): return False patient_codes = set(str(icd_string).upper().replace(' ', '').split(';')) return not patient_codes.isdisjoint(exclude_icds) excluded_patients = df_thuoc[df_thuoc['maicd'].apply(check_exclusion)]['makcb'].unique() num_excluded = len(excluded_patients) suspect_mabhyt = set(df_thuoc_plt['mabhyt'].unique()) final_filtered_thuoc = df_thuoc[ (df_thuoc['mabhyt'].isin(suspect_mabhyt)) & (~df_thuoc['makcb'].isin(excluded_patients)) ].copy() check_thuoc = df_thuoc[df_thuoc['mabhyt'].isin(suspect_mabhyt)].copy() num_suspect_patients = len(set(check_thuoc['makcb'])) patient_start_dates = final_filtered_thuoc.groupby('makcb')['ngaychidinh'].min().reset_index() patient_start_dates.columns = ['makcb', 'start_date'] df_plt = df_xetnghiem[df_xetnghiem['Tên XN'] == 'PLT'].copy() df_plt['Kết quả XN'] = df_plt['Kết quả XN'].astype(str).str.replace(',', '.') df_plt['Kết quả XN'] = pd.to_numeric(df_plt['Kết quả XN'], errors='coerce') check_plt_low = df_plt[df_plt['Kết quả XN'] < 100].copy() num_plt_low = len(set(check_plt_low['Mã KCB'])) ids_thuoc = set(check_thuoc['makcb']) ids_low = set(check_plt_low['Mã KCB']) inter_ids = ids_thuoc.intersection(ids_low) num_inter = len(inter_ids) merged = pd.merge(df_plt, patient_start_dates, left_on='Mã KCB', right_on='makcb') baseline_candidates = merged[merged['Ngày KQ'] <= merged['start_date']] baseline_plt = baseline_candidates.sort_values(['makcb', 'Ngày KQ'], ascending=[True, False]).drop_duplicates('makcb') baseline_plt = baseline_plt.rename(columns={'Kết quả XN': 'PLT_baseline', 'Ngày KQ': 'Ngày_PLT_baseline'}) num_baseline = len(baseline_plt) after_drug_tests = merged[merged['Ngày KQ'] > merged['start_date']] nadir_plt = after_drug_tests.sort_values(['makcb', 'Kết quả XN'], ascending=[True, True]).drop_duplicates('makcb') nadir_plt = nadir_plt.rename(columns={'Kết quả XN': 'PLT_nadir', 'Ngày KQ': 'Ngày_PLT_nadir'}) final_analysis = pd.merge( baseline_plt[['makcb', 'Họ tên', 'PLT_baseline', 'Ngày_PLT_baseline', 'start_date']], nadir_plt[['makcb', 'PLT_nadir', 'Ngày_PLT_nadir']], on='makcb', how='outer' ) final_analysis = final_analysis.merge(df_plt[['Mã KCB', 'Họ tên']].drop_duplicates(), left_on='makcb', right_on='Mã KCB', how='left') final_analysis['Họ tên'] = final_analysis['Họ tên_y'].fillna(final_analysis['Họ tên_x']) final_analysis = final_analysis.drop(columns=['Họ tên_x', 'Họ tên_y', 'Mã KCB'] if 'Họ tên_x' in final_analysis.columns else ['Họ tên_y', 'Mã KCB']) final_analysis['Percent_Drop'] = np.where( (final_analysis['PLT_baseline'].notna()) & (final_analysis['PLT_nadir'].notna()), ((final_analysis['PLT_baseline'] - final_analysis['PLT_nadir']) / final_analysis['PLT_baseline']) * 100, 0 ) def identify_signal(row): if pd.notna(row['PLT_nadir']): if pd.notna(row['PLT_baseline']): if row['PLT_baseline'] >= 150: if row['PLT_nadir'] < 100 or row['Percent_Drop'] >= 50: return "Nghi ngờ cao (DITP)" elif 100 <= row['PLT_baseline'] < 150: if row['PLT_nadir'] < 75 or row['Percent_Drop'] >= 40: return "Nghi ngờ (DITP)" elif row['PLT_baseline'] < 100: if row['Percent_Drop'] >= 30: return "Trầm trọng thêm do thuốc" else: if row['PLT_nadir'] < 100: if pd.notna(row['Ngày_PLT_nadir']) and pd.notna(row['start_date']) and (row['Ngày_PLT_nadir'] - row['start_date']) <= timedelta(hours=24): return "Nghi ngờ (thiếu baseline nhưng gần)" return "Nghi ngờ (thiếu baseline)" return "Không rõ rệt" final_analysis['Signal'] = final_analysis.apply(identify_signal, axis=1) def get_ctcae_plt(plt_value): if pd.isna(plt_value): return "N/A" if plt_value >= 150: return "Bình thường" elif 100 <= plt_value < 150: return "Độ 1 (Nhẹ)" elif 75 <= plt_value < 100: return "Độ 2 (Trung bình)" elif 50 <= plt_value < 75: return "Độ 3 (Nặng)" elif 25 <= plt_value < 50: return "Độ 4 (Rất nặng)" else: return "Độ 5 (Tử vong)" final_analysis['CTCAE_Baseline'] = final_analysis['PLT_baseline'].apply(get_ctcae_plt) final_analysis['CTCAE_Nadir'] = final_analysis['PLT_nadir'].apply(get_ctcae_plt) def get_nguy_hiem(plt_value): if pd.isna(plt_value): return "N/A" if plt_value >= 150: return "An toàn" elif 100 <= plt_value < 150: return "Theo dõi" elif 50 <= plt_value < 100: return "Có nguy cơ, theo dõi chặt" else: return "Nguy hiểm cao, can thiệp ngay" final_analysis['Nguy_Hiem_Baseline'] = final_analysis['PLT_baseline'].apply(get_nguy_hiem) final_analysis['Nguy_Hiem_Nadir'] = final_analysis['PLT_nadir'].apply(get_nguy_hiem) num_drop_30 = len(final_analysis[(final_analysis['Percent_Drop'] >= 30) & (final_analysis['Signal'] != "Không rõ rệt")]['makcb'].unique()) # Xuất các file check file1 = "1_Check_Thuoc_Nghi_Ngo.xlsx" check_thuoc.to_excel(file1, sheet_name='Sheet1', index=False) file2 = "2_Check_Loai_ICD.xlsx" pd.DataFrame({'Excluded': list(excluded_patients)}).to_excel(file2, sheet_name='Sheet1', index=False) file3 = "3_Check_PLT_Duoi_100.xlsx" check_plt_low.to_excel(file3, sheet_name='Sheet1', index=False) inter_df = df_plt[df_plt['Mã KCB'].isin(inter_ids)].copy() inter_df['CTCAE_Grade'] = inter_df['Kết quả XN'].apply(get_ctcae_plt) inter_df['Nguy_Hiem'] = inter_df['Kết quả XN'].apply(get_nguy_hiem) file4 = "4_Check_Giao_Thoa_Thuoc_PLT.xlsx" inter_df.to_excel(file4, sheet_name='Sheet1', index=False) file5 = "5_Check_Baseline_Nadir_Full.xlsx" final_analysis.to_excel(file5, sheet_name='Sheet1', index=False) def export_realtime_reports(final_report, df_xetnghiem): today = datetime.now() filerealtime = f"Danh_Sach_Can_Thiep_DLS_{today.year}.xlsx" final_report = final_report.merge(final_filtered_thuoc[['makcb', 'tenhh', 'hoatchat', 'maicd', 'chandoan']].drop_duplicates(), on='makcb', how='left') final_report = final_report.merge(df_xetnghiem[['Mã KCB', 'Ngày vào', 'Ngày ra', 'Khoa']].drop_duplicates(), left_on='makcb', right_on='Mã KCB', how='left') final_report['Ngày vào'] = pd.to_datetime(final_report['Ngày vào'], errors='coerce') final_report['Ngày ra'] = pd.to_datetime(final_report['Ngày ra'], errors='coerce') current_patients = final_report[final_report['Ngày ra'].isna() & (final_report['Signal'] != "Không rõ rệt")].copy() hoi_cuu = final_report[final_report['Ngày ra'].notna() & (final_report['Signal'] != "Không rõ rệt")].copy() cols = ['makcb', 'Họ tên', 'Ngày vào', 'Khoa', 'chandoan', 'start_date', 'PLT_baseline', 'Ngày_PLT_baseline', 'PLT_nadir', 'Ngày_PLT_nadir', 'Percent_Drop', 'Signal', 'CTCAE_Nadir', 'Nguy_Hiem_Nadir', 'tenhh', 'hoatchat'] include_priority = True include_recommendations = True num_need_plt = 0 num_can_thiep = 0 if not current_patients.empty: current_patients['So_Ngay_Nam_Vien'] = ((today - current_patients['Ngày vào']).dt.days).fillna(0).astype(int) if include_priority: def assess_urgency(row): days = row['So_Ngay_Nam_Vien'] if days >= 7: return "🔴 Rất cao (>= 7 ngày chưa XN lại)" if days >= 3: return "🟠 Cao (3-6 ngày chưa XN lại)" return "🟡 Trung bình" current_patients['Muc_Do_Cap_Bach'] = current_patients.apply(assess_urgency, axis=1) current_patients = current_patients.sort_values('So_Ngay_Nam_Vien', ascending=False) if include_recommendations: def compose_message(row): thuoc = row['tenhh'] if pd.notna(row['tenhh']) else row['hoatchat'] plt_nen = row['PLT_baseline'] if pd.notna(row['PLT_baseline']) else 'N/A (thiếu baseline)' ngay = row['So_Ngay_Nam_Vien'] nadir = row.get('PLT_nadir', 'N/A') ctcae_nadir = row['CTCAE_Nadir'] nguy_hiem_nadir = row['Nguy_Hiem_Nadir'] if "Nghi ngờ (thiếu baseline" in row['Signal']: return (f"CẢNH BÁO ADR: Tiểu cầu bệnh nhân thấp ({nadir} G/L - {ctcae_nadir} - {nguy_hiem_nadir}) khi dùng {thuoc} (thuốc đã được báo cáo làm giảm tiểu cầu). " f"Đề xuất theo dõi và làm thêm XN Công thức máu (PLT) trong những ngày tới để cân nhắc điều trị.") elif "Trầm trọng" in row['Signal'] or "Nghi ngờ" in row['Signal']: return (f"CẢNH BÁO ADR: Tiểu cầu bệnh nhân sụt giảm rõ rệt từ {plt_nen} G/L xuống còn {nadir} G/L " f"({ctcae_nadir} - {nguy_hiem_nadir}) sau khi dùng {thuoc}. " f"Đề xuất bác sĩ xem xét tạm ngưng hoặc đổi thuốc, truyền tiểu cầu nếu cần, hoặc theo dõi chặt chẽ tùy mức độ.") else: return (f"Bệnh nhân đã điều trị {ngay} ngày và đang sử dụng {thuoc} (có nguy cơ giảm PLT). " f"Kết quả PLT nền thấp ({plt_nen} G/L) và chưa có xét nghiệm kiểm tra lại. " f"Dược lâm sàng đề xuất làm thêm XN Công thức máu (PLT) trong ngày hôm nay.") current_patients['Mau_Cau_Khuyen_Cao'] = current_patients.apply(compose_message, axis=1) num_need_plt = len(set(current_patients[current_patients['Mau_Cau_Khuyen_Cao'].str.contains("làm thêm XN")]['makcb'])) num_can_thiep = len(set(current_patients[~current_patients['Mau_Cau_Khuyen_Cao'].str.contains("làm thêm XN")]['makcb'])) with pd.ExcelWriter(filerealtime) as writer: sheet_written = False if not current_patients.empty: current_patients[cols + ['So_Ngay_Nam_Vien', 'Muc_Do_Cap_Bach', 'Mau_Cau_Khuyen_Cao']].to_excel(writer, sheet_name='CAN_THIEP_TRUC_TIEP', index=False) sheet_written = True if not hoi_cuu.empty: hoi_cuu[cols].to_excel(writer, sheet_name='HOI_CUU', index=False) sheet_written = True if not sheet_written: pd.DataFrame([{"Message": "Không có dữ liệu báo cáo để xuất."}]).to_excel(writer, sheet_name='Khong_co_du_lieu', index=False) num_hoicuu = len(set(hoi_cuu['makcb'])) return filerealtime, num_can_thiep, num_hoicuu, num_need_plt filerealtime, num_can_thiep, num_hoicuu, num_need_plt = export_realtime_reports(final_analysis, df_xetnghiem) output_text = f"✅ Số bệnh nhân có PLT dưới 100: {num_plt_low}\n" output_text += f"✅ Số bệnh nhân sử dụng thuốc nghi ngờ và có PLT dưới 100: {num_inter}\n" output_text += f"✅ Số bệnh nhân sử dụng thuốc nghi ngờ và có PLT giảm >=30%: {num_drop_30}\n" output_text += f"✅ Số bệnh nhân cần làm thêm xét nghiệm PLT: {num_need_plt}\n" output_text += f"✅ Số bệnh nhân cần can thiệp: {num_can_thiep}\n" output_text += f"✅ Số bệnh nhân cần hồi cứu báo cáo ADR: {num_hoicuu}\n" output_text += f"✅ Đã xuất file '{os.path.basename(filerealtime)}'.\n" return output_text, file1, file2, file3, file4, file5, filerealtime with gr.Blocks(title="triggertool PLT") as demo: # ===== INJECT CSS ===== gr.HTML("""""") # ===== HEADER HTML ===== credit_text = "Được tạo bởi DSCKI. Nguyễn Thành Trung - Trung tâm y tế khu vực Bảo Thắng" animated_credit_html = ''.join([f'{char}' if char != ' ' else ' ' for i, char in enumerate(credit_text)]) gr.HTML(f"""
🩸 triggertool PLT
{animated_credit_html}
""") # ===== MAIN LAYOUT ===== with gr.Row(): # === CỘT TRÁI: Upload === with gr.Column(scale=1): gr.HTML("
📂 Tải lên file dữ liệu
") with gr.Group(elem_classes=["upload-card"]): thuoc_upload = gr.File(label="💊 thuoc.xlsx") icd_plt_upload = gr.File(label="🏷️ ICD_PLT.xlsx") xetnghiem_upload= gr.File(label="🧪 xetnghiem.xlsx") thuoc_plt_upload= gr.File(label="📋 Thuoc_PLT.xlsx") run_button = gr.Button("🔍 Kiểm tra", variant="primary", size="lg", elem_classes=["run-btn"]) # === CỘT PHẢI: Kết quả === with gr.Column(scale=1): gr.HTML("
📊 Kết quả xử lý
") output_text = gr.Textbox( label="Tổng hợp kết quả", lines=10, max_lines=12, autoscroll=False, elem_id="result-summary" ) # ===== DOWNLOAD SECTION ===== with gr.Accordion("📥 Tải xuống file kết quả", open=False, elem_classes=["download-accordion"]): with gr.Row(): file1_download = gr.File(label="1 · Thuốc nghi ngờ") file2_download = gr.File(label="2 · Loại ICD đã loại trừ") file3_download = gr.File(label="3 · PLT dưới 100") with gr.Row(): file4_download = gr.File(label="4 · Giao thoa Thuốc × PLT") file5_download = gr.File(label="5 · Baseline & Nadir đầy đủ") realtime_download = gr.File(label="🏥 Danh sách can thiệp DLS") run_button.click( process_files, inputs=[thuoc_upload, xetnghiem_upload, thuoc_plt_upload, icd_plt_upload], outputs=[output_text, file1_download, file2_download, file3_download, file4_download, file5_download, realtime_download] ) demo.launch()