Spaces:
Sleeping
Sleeping
| # app.py - 添加双重验证系统的完整版本 | |
| import gradio as gr | |
| import json | |
| import os | |
| import html | |
| from huggingface_hub import HfApi, hf_hub_download | |
| # --- Configuration --- | |
| HF_TOKEN_ENV_VAR = "HF_TOKEN" | |
| GRADIO_USERNAME_ENV_VAR = "GRADIO_USERNAME" | |
| GRADIO_PASSWORD_ENV_VAR = "GRADIO_PASSWORD" | |
| DATA_REPO_ID = "liveclinreview/translated_liveClin" | |
| RESULTS_REPO_ID = "liveclinreview/review_check_data" | |
| RESULTS_FILENAME_IN_REPO = "review_results.json" | |
| MAX_MCQS_PER_FILE = 30 | |
| # --- 辅助函数 --- | |
| def get_translated_text(data_dict, key, fallback_key=None, default_value="N/A"): | |
| """优先获取翻译文本，无翻译时回退到原文""" | |
| if not isinstance(data_dict, dict): | |
| return default_value | |
| translated_key = f"{key}_translated" | |
| if translated_key in data_dict and data_dict[translated_key] is not None and data_dict[translated_key].strip(): | |
| return data_dict[translated_key] | |
| if key in data_dict and data_dict[key] is not None: | |
| return data_dict[key] | |
| if fallback_key: | |
| translated_fallback_key = f"{fallback_key}_translated" | |
| if translated_fallback_key in data_dict and data_dict[translated_fallback_key] is not None and data_dict[translated_fallback_key].strip(): | |
| return data_dict[translated_fallback_key] | |
| if fallback_key in data_dict and data_dict[fallback_key] is not None: | |
| return data_dict[fallback_key] | |
| return default_value | |
| def get_translated_caption(img_data_dict, default_value="图片说明不可用"): | |
| """获取图片翻译说明""" | |
| if not isinstance(img_data_dict, dict): | |
| return default_value | |
| translated_keys_to_try = [ | |
| "image_detail_translated", | |
| "image_caption_translated", | |
| "caption_translated", | |
| "image_caption_with_id_translated" | |
| ] | |
| for key in translated_keys_to_try: | |
| if key in img_data_dict and img_data_dict[key] is not None and img_data_dict[key].strip(): | |
| return img_data_dict[key] | |
| original_keys_to_try = [ | |
| "image_detail", "image_caption", "caption", "image_caption_with_id" | |
| ] | |
| for key in original_keys_to_try: | |
| if key in img_data_dict and img_data_dict[key] is not None: | |
| return img_data_dict[key] | |
| return default_value | |
| def resolve_image_url(img_data_dict): | |
| """从图片数据字典中解析图片URL""" | |
| if not isinstance(img_data_dict, dict): | |
| return None | |
| url = img_data_dict.get('url') | |
| if url and url.strip(): | |
| return url.strip() | |
| for alt_key in ['image_url', 'src', 'href']: | |
| alt_url = img_data_dict.get(alt_key) | |
| if alt_url and alt_url.strip(): | |
| return alt_url.strip() | |
| return None | |
| def create_error_data(error_msg): | |
| """创建错误数据结构""" | |
| return { | |
| "title": f"Error: {error_msg}", | |
| "title_translated": f"错误: {error_msg}", | |
| "case_text": "无法加载案例数据", | |
| "case_text_translated": "无法加载案例数据", | |
| "discussion_text": "无法加载讨论数据", | |
| "discussion_text_translated": "无法加载讨论数据", | |
| "image_details": [], | |
| "table_details": [], | |
| "exam_creation": {"final_policy": {"mcqs": []}} | |
| } | |
| # --- Hugging Face 数据交互函数 --- | |
| def find_json_files_from_hf(): | |
| """从预生成的索引文件快速获取文件列表""" | |
| hf_token = os.environ.get(HF_TOKEN_ENV_VAR) | |
| if not hf_token: | |
| print("Warning: 未找到HF_TOKEN，无法列出文件。") | |
| return [] | |
| try: | |
| print("正在下载文件索引...") | |
| index_path = hf_hub_download( | |
| repo_id=DATA_REPO_ID, | |
| filename="file_index.json", | |
| repo_type="dataset", | |
| token=hf_token | |
| ) | |
| with open(index_path, 'r', encoding='utf-8') as f: | |
| json_files = json.load(f) | |
| print(f"从索引找到 {len(json_files)} 个JSON文件") | |
| return json_files | |
| except Exception as e: | |
| print(f"下载文件索引失败: {e}") | |
| return [] | |
| def load_case_data_from_hf(path_in_repo): | |
| """从Hugging Face数据集加载案例数据""" | |
| if not path_in_repo: | |
| return create_error_data("无效的文件路径"), "INVALID_PATH" | |
| hf_token = os.environ.get(HF_TOKEN_ENV_VAR) | |
| if not hf_token: | |
| return create_error_data("未找到HF token"), path_in_repo | |
| try: | |
| print(f"正在下载文件: {path_in_repo}") | |
| downloaded_file_path = hf_hub_download( | |
| repo_id=DATA_REPO_ID, | |
| filename=path_in_repo, | |
| repo_type="dataset", | |
| token=hf_token | |
| ) | |
| with open(downloaded_file_path, 'r', encoding='utf-8') as f: | |
| data = json.load(f) | |
| print(f"文件加载成功: {path_in_repo}") | |
| return data, path_in_repo | |
| except Exception as e: | |
| print(f"加载文件失败 {path_in_repo}: {e}") | |
| return create_error_data(f"加载失败: {e}"), path_in_repo | |
| def load_results_from_hf(): | |
| """从Hugging Face数据集加载已保存的审核结果""" | |
| hf_token = os.environ.get(HF_TOKEN_ENV_VAR) | |
| if not hf_token: | |
| print("Warning: HF_TOKEN 未设置，无法加载审核结果。") | |
| return {} | |
| try: | |
| print(f"正在下载审核结果...") | |
| downloaded_path = hf_hub_download( | |
| repo_id=RESULTS_REPO_ID, | |
| filename=RESULTS_FILENAME_IN_REPO, | |
| repo_type="dataset", | |
| token=hf_token, | |
| ) | |
| with open(downloaded_path, 'r', encoding='utf-8') as f: | |
| results = json.load(f) | |
| print("审核结果加载成功") | |
| return results | |
| except Exception as e: | |
| print(f"无法加载审核结果文件: {e}") | |
| return {} | |
| def save_results_to_hf(data_to_save): | |
| """保存审核结果到Hugging Face数据集""" | |
| hf_token = os.environ.get(HF_TOKEN_ENV_VAR) | |
| if not hf_token: | |
| msg = "Warning: HF_TOKEN 未设置，无法保存审核结果。" | |
| print(msg) | |
| return msg, data_to_save | |
| temp_file_path = "temp_review_results.json" | |
| try: | |
| with open(temp_file_path, 'w', encoding='utf-8') as f: | |
| json.dump(data_to_save, f, indent=2, ensure_ascii=False) | |
| api = HfApi(token=hf_token) | |
| api.upload_file( | |
| path_or_fileobj=temp_file_path, | |
| path_in_repo=RESULTS_FILENAME_IN_REPO, | |
| repo_id=RESULTS_REPO_ID, | |
| repo_type="dataset", | |
| commit_message="Update review results" | |
| ) | |
| msg = f"结果已成功保存到 HF Dataset: {RESULTS_REPO_ID}" | |
| print(msg) | |
| return msg, data_to_save | |
| except Exception as e: | |
| msg = f"保存结果到 HF Dataset 时出错: {e}" | |
| print(msg) | |
| return msg, data_to_save | |
| finally: | |
| if os.path.exists(temp_file_path): | |
| os.remove(temp_file_path) | |
| # --- Critic Text --- | |
| def get_critic_evaluation_text(case_data_dict, mcq_index): | |
| if not case_data_dict or not isinstance(case_data_dict.get("exam_creation"), dict): | |
| return "Evaluation data missing or invalid." | |
| exam_creation = case_data_dict["exam_creation"] | |
| final_policy = exam_creation.get("final_policy", {}) | |
| critic_data = None | |
| finalized_round_num = final_policy.get("finalized_from_round") | |
| if finalized_round_num: | |
| potential_key = f"round{finalized_round_num}_critic" | |
| if potential_key in exam_creation and isinstance(exam_creation[potential_key], dict): | |
| critic_data = exam_creation[potential_key] | |
| if critic_data is None: | |
| for i in range(3, 0, -1): | |
| potential_key = f"round{i}_critic" | |
| if potential_key in exam_creation and isinstance(exam_creation[potential_key], dict): | |
| potential_critiques = exam_creation[potential_key].get("critiques") | |
| if isinstance(potential_critiques, list) and potential_critiques: | |
| critic_data = exam_creation[potential_key] | |
| break | |
| if critic_data is None and isinstance(final_policy.get("critiques"), list) and final_policy["critiques"]: | |
| critiques_list_in_final_policy = final_policy["critiques"] | |
| for critique_obj in critiques_list_in_final_policy: | |
| current_critique_mcq_idx = critique_obj.get("original_mcq_index") | |
| if current_critique_mcq_idx is None and isinstance(critique_obj.get("critique"), dict): | |
| current_critique_mcq_idx = critique_obj["critique"].get("mcq_index") | |
| if current_critique_mcq_idx == mcq_index: | |
| actual_critique_content = critique_obj.get("critique", {}) | |
| if isinstance(actual_critique_content, dict): | |
| correctness_eval = actual_critique_content.get("correctness_evaluation", {}) | |
| suggestions = get_translated_text(correctness_eval, "critique_and_suggestions", default_value=f"Evaluation for MCQ {mcq_index + 1} in final_policy found, but content is empty or invalid.") | |
| return f"{suggestions}" | |
| else: | |
| return str(actual_critique_content) | |
| return f"Could not find specific evaluation for question {mcq_index + 1} in final_policy's critique list." | |
| if critic_data: | |
| critiques = critic_data.get("critiques", []) | |
| if not isinstance(critiques, list): | |
| return f"Invalid 'critiques' format in the selected critic data." | |
| for critique_obj in critiques: | |
| current_critique_mcq_idx = critique_obj.get("original_mcq_index") | |
| if current_critique_mcq_idx is None and isinstance(critique_obj.get("critique"), dict): | |
| current_critique_mcq_idx = critique_obj["critique"].get("mcq_index") | |
| if current_critique_mcq_idx == mcq_index: | |
| actual_critique_content = critique_obj.get("critique", {}) | |
| if isinstance(actual_critique_content, dict): | |
| correctness_eval = actual_critique_content.get("correctness_evaluation", {}) | |
| suggestions = get_translated_text(correctness_eval, "critique_and_suggestions", default_value="评价内容为空。") | |
| return f"{suggestions}" | |
| else: | |
| return str(actual_critique_content) | |
| return f"找到了评价数据，但没有找到问题 {mcq_index + 1} 的具体条目。" | |
| return "未找到对应的模型评价数据（例如 roundX_critic）。" | |
| def get_reassessment_text(case_data_dict, mcq_index): | |
| """获取模型再评估文本""" | |
| if not case_data_dict or not isinstance(case_data_dict.get("quality_assessment"), dict): | |
| return "再评估数据不可用" | |
| qa = case_data_dict["quality_assessment"] | |
| assessment_structured = qa.get("assessment_structured", {}) | |
| if not assessment_structured: | |
| return "再评估数据不可用" | |
| # MCQ索引从0开始，但显示时是从1开始，所以要+1 | |
| mcq_key = f"mcq{mcq_index + 1}" | |
| # 优先获取中文版本 | |
| chinese_version = assessment_structured.get("chinese", {}) | |
| if mcq_key in chinese_version: | |
| text = chinese_version[mcq_key] | |
| # 在 bullet point 前添加换行 | |
| return text.replace("•", "\n•") | |
| # 如果没有中文版本，尝试英文版本 | |
| english_version = assessment_structured.get("english", {}) | |
| if mcq_key in english_version: | |
| text = english_version[mcq_key] | |
| # 在 bullet point 前添加换行 | |
| return text.replace("•", "\n•") | |
| return f"未找到问题 {mcq_index + 1} 的再评估数据" | |
| def get_overall_reassessment_text(case_data_dict): | |
| """获取整体再评估总结文本""" | |
| if not case_data_dict or not isinstance(case_data_dict.get("quality_assessment"), dict): | |
| return "" | |
| qa = case_data_dict["quality_assessment"] | |
| assessment_structured = qa.get("assessment_structured", {}) | |
| if not assessment_structured: | |
| return "" | |
| # 优先获取中文版本 | |
| chinese_version = assessment_structured.get("chinese", {}) | |
| if "overall" in chinese_version: | |
| return chinese_version["overall"] | |
| # 如果没有中文版本，尝试英文版本 | |
| english_version = assessment_structured.get("english", {}) | |
| if "overall" in english_version: | |
| return english_version["overall"] | |
| return "" | |
| # --- UI Update Logic --- | |
| def update_static_content(case_data_dict, current_json_path): | |
| is_error = not case_data_dict or "Error:" in case_data_dict.get("title", "") | |
| if is_error: | |
| err_title = get_translated_text(case_data_dict, "title", default_value="加载数据时出错") | |
| err_text = get_translated_text(case_data_dict, "case_text", fallback_key="abstract", default_value="无法加载或解析案例数据。") | |
| return { | |
| "status_message_md": gr.update(value=f"**加载文件 {os.path.basename(current_json_path)} 时出错：** {err_text}", visible=True), | |
| "case_display_area": gr.update(visible=False), | |
| "title_md": gr.update(value=""), | |
| "level_info_md": gr.update(value="", visible=False), | |
| "ids_md": gr.update(value=""), | |
| "case_text_md": gr.update(value=""), | |
| "discussion_text_md": gr.update(value=""), | |
| "general_tables_display_md": gr.update(value="", visible=False), | |
| "single_general_img": gr.update(value=None, visible=False), | |
| "single_general_caption_md": gr.update(value="", visible=False), | |
| "general_images_gallery_multi": gr.update(value=[], visible=False), | |
| "general_images_captions_md": gr.update(value="", visible=False), | |
| "quality_assessment_md": gr.update(value="", visible=False), | |
| "scoring_summary_md": gr.update(value="", visible=False), | |
| "scenario_text_md": gr.update(value=""), | |
| "single_scenario_img": gr.update(value=None, visible=False), | |
| "single_scenario_caption_md": gr.update(value="", visible=False), | |
| "scenario_images_gallery": gr.update(value=[], visible=False), | |
| "scenario_images_captions_md": gr.update(value="", visible=False), | |
| "overall_reassessment_accordion": gr.update(visible=False), # 添加 | |
| "overall_reassessment_text": gr.update(value=""), | |
| } | |
| updates = { | |
| "status_message_md": gr.update(value="案例加载成功。", visible=False), | |
| "case_display_area": gr.update(visible=True), | |
| "title_md": gr.update(value=f"# {get_translated_text(case_data_dict, 'title', default_value='N/A')}"), | |
| "ids_md": gr.update(value=f"**DOI:** {case_data_dict.get('doi', 'N/A')}"), | |
| "case_text_md": gr.update(value=get_translated_text(case_data_dict, 'case_text', default_value='案例文本不可用。')), | |
| "discussion_text_md": gr.update(value=get_translated_text(case_data_dict, 'discussion_text', default_value='讨论部分不可用。')), | |
| } | |
| # Level信息 | |
| level_parts = [] | |
| level1 = get_translated_text(case_data_dict, 'Level1', default_value="") | |
| level2 = get_translated_text(case_data_dict, 'Level2', default_value="") | |
| level2_detail = get_translated_text(case_data_dict, 'Level2_Detail', default_value="") | |
| if level1 and level1 != "N/A": | |
| level_parts.append(f"**分类级别1:** {html.escape(level1)}") | |
| if level2 and level2 != "N/A": | |
| level_parts.append(f"**分类级别2:** {html.escape(level2)}") | |
| if level2_detail and level2_detail != "N/A": | |
| level_parts.append(f"**详细分类:** {html.escape(level2_detail)}") | |
| level_info_text = "\n\n".join(level_parts) if level_parts else "" | |
| updates["level_info_md"] = gr.update(value=level_info_text, visible=bool(level_parts)) | |
| # 处理表格 | |
| general_tables_parts = [] | |
| if case_data_dict.get('table_details'): | |
| for tbl in case_data_dict['table_details']: | |
| if isinstance(tbl, dict): | |
| caption = get_translated_text(tbl, 'caption', default_value='表格') | |
| content = get_translated_text(tbl, 'table_translated', fallback_key='table', default_value='表格内容为空。') | |
| general_tables_parts.append(f"### {html.escape(caption)}") | |
| general_tables_parts.append(content) | |
| updates["general_tables_display_md"] = gr.update(value="\n\n".join(general_tables_parts), visible=bool(general_tables_parts)) | |
| # 处理通用图片 | |
| general_images_list = case_data_dict.get('image_details', []) | |
| if len(general_images_list) == 1: | |
| img_data = general_images_list[0] | |
| img_url = resolve_image_url(img_data) | |
| caption = get_translated_caption(img_data, default_value='案例图片') | |
| updates["single_general_img"] = gr.update(value=img_url, visible=bool(img_url)) | |
| updates["single_general_caption_md"] = gr.update(value=f"<p style='text-align:center; font-style:italic; margin-top:0px;'>{html.escape(caption)}</p>", visible=bool(img_url)) | |
| updates["general_images_gallery_multi"] = gr.update(value=[], visible=False) | |
| updates["general_images_captions_md"] = gr.update(value="", visible=False) | |
| elif len(general_images_list) > 1: | |
| gallery_items = [] | |
| captions_list_md_parts = ["**图片说明详情：**"] | |
| for idx, img_data in enumerate(general_images_list): | |
| if not isinstance(img_data, dict): | |
| continue | |
| img_url = resolve_image_url(img_data) | |
| short_caption_for_gallery = img_data.get('caption', f'图片 {idx+1}') | |
| long_translated_caption = get_translated_caption(img_data, default_value=f'图片 {idx+1}') | |
| if img_url: | |
| gallery_items.append((img_url, html.escape(short_caption_for_gallery))) | |
| captions_list_md_parts.append(f"{idx+1}. {html.escape(long_translated_caption)}") | |
| updates["single_general_img"] = gr.update(value=None, visible=False) | |
| updates["single_general_caption_md"] = gr.update(value="", visible=False) | |
| updates["general_images_gallery_multi"] = gr.update(value=gallery_items, visible=bool(gallery_items)) | |
| updates["general_images_captions_md"] = gr.update(value="\n".join(captions_list_md_parts), visible=bool(gallery_items)) | |
| else: | |
| updates["single_general_img"] = gr.update(value=None, visible=False) | |
| updates["single_general_caption_md"] = gr.update(value="", visible=False) | |
| updates["general_images_gallery_multi"] = gr.update(value=[], visible=False) | |
| updates["general_images_captions_md"] = gr.update(value="", visible=False) | |
| # 质量评估 | |
| quality_parts = [] | |
| if case_data_dict.get('quality_assessment'): | |
| qa = case_data_dict['quality_assessment'] | |
| assessment_text = get_translated_text(qa, 'assessment', default_value="") | |
| if assessment_text and assessment_text.strip(): | |
| quality_parts.append("### 质量评估") | |
| quality_parts.append(assessment_text) | |
| updates["quality_assessment_md"] = gr.update( | |
| value="\n\n".join(quality_parts) if quality_parts else "", | |
| visible=bool(quality_parts) # 只有有内容时才显示 | |
| ) | |
| # 评分信息 | |
| scoring_parts = [] | |
| if case_data_dict.get('scoring'): | |
| scoring = case_data_dict['scoring'] | |
| # 严重缺陷 - 只有当details有内容时才显示 | |
| critical_flaws = scoring.get('critical_flaws', {}) | |
| if isinstance(critical_flaws, dict): | |
| details = critical_flaws.get('details', []) | |
| if isinstance(details, list) and details: # 只有details非空时才处理 | |
| count = critical_flaws.get('count', len(details)) | |
| scoring_parts.append(f"### 严重缺陷 (数量: {count})") | |
| for i, detail in enumerate(details): | |
| detail_text = "" | |
| if isinstance(detail, dict): | |
| # 优先使用 details_translated | |
| detail_text = detail.get('details_translated', '').strip() | |
| if not detail_text: | |
| detail_text = detail.get('details', '').strip() | |
| elif isinstance(detail, str): | |
| detail_text = detail.strip() | |
| if detail_text: # 只添加非空的detail | |
| scoring_parts.append(f"{i+1}. {detail_text}") | |
| # 次要问题 - 只有当details有内容时才显示 | |
| minor_issues = scoring.get('minor_issues', {}) | |
| if isinstance(minor_issues, dict): | |
| details = minor_issues.get('details', []) | |
| if isinstance(details, list) and details: # 只有details非空时才处理 | |
| count = minor_issues.get('count', len(details)) | |
| if scoring_parts: # 如果前面有内容，添加空行 | |
| scoring_parts.append("") | |
| scoring_parts.append(f"### 次要问题 (数量: {count})") | |
| for i, detail in enumerate(details): | |
| detail_text = "" | |
| if isinstance(detail, dict): | |
| # 优先使用 details_translated | |
| detail_text = detail.get('details_translated', '').strip() | |
| if not detail_text: | |
| detail_text = detail.get('details', '').strip() | |
| elif isinstance(detail, str): | |
| detail_text = detail.strip() | |
| if detail_text: # 只添加非空的detail | |
| scoring_parts.append(f"{i+1}. {detail_text}") | |
| # 总体评价 | |
| overall_summary = get_translated_text(scoring, 'overall_summary', default_value="") | |
| if overall_summary and overall_summary.strip(): | |
| if scoring_parts: # 如果前面有内容，添加空行 | |
| scoring_parts.append("") | |
| scoring_parts.append("### 总体评价") | |
| scoring_parts.append(overall_summary) | |
| updates["scoring_summary_md"] = gr.update( | |
| value="\n".join(scoring_parts) if scoring_parts else "", | |
| visible=bool(scoring_parts) # 只有有内容时才显示 | |
| ) | |
| # 处理场景信息 | |
| exam_creation_data = case_data_dict.get('exam_creation', {}) | |
| final_policy_data = exam_creation_data.get('final_policy', {}) | |
| scenario_text = get_translated_text(final_policy_data, 'new_scenario', fallback_key='scenario', default_value='临床场景描述不可用。') | |
| updates["scenario_text_md"] = gr.update(value=scenario_text) | |
| # 处理场景图片 | |
| scenario_images_list = final_policy_data.get('scenario_image_details', []) | |
| if not isinstance(scenario_images_list, list): | |
| scenario_images_list = [] | |
| if len(scenario_images_list) == 1: | |
| img_data = scenario_images_list[0] | |
| img_url = resolve_image_url(img_data) | |
| caption = get_translated_caption(img_data, default_value='场景图片') | |
| updates["single_scenario_img"] = gr.update(value=img_url, visible=bool(img_url)) | |
| updates["single_scenario_caption_md"] = gr.update(value=f"<p style='text-align:center; font-style:italic; margin-top:0px;'>{html.escape(caption)}</p>", visible=bool(img_url)) | |
| updates["scenario_images_gallery"] = gr.update(value=[], visible=False) | |
| updates["scenario_images_captions_md"] = gr.update(value="", visible=False) | |
| elif len(scenario_images_list) > 1: | |
| gallery_items = [] | |
| captions_list_md_parts = ["**图片说明详情：**"] | |
| for idx, img_data in enumerate(scenario_images_list): | |
| if not isinstance(img_data, dict): | |
| continue | |
| img_url = resolve_image_url(img_data) | |
| full_caption = get_translated_caption(img_data, default_value=f'场景图片 {idx+1}') | |
| if img_url: | |
| gallery_items.append((img_url, html.escape(full_caption))) | |
| captions_list_md_parts.append(f"{idx+1}. {html.escape(full_caption)}") | |
| updates["single_scenario_img"] = gr.update(value=None, visible=False) | |
| updates["single_scenario_caption_md"] = gr.update(value="", visible=False) | |
| updates["scenario_images_gallery"] = gr.update(value=gallery_items, visible=bool(gallery_items)) | |
| updates["scenario_images_captions_md"] = gr.update(value="\n".join(captions_list_md_parts), visible=bool(gallery_items)) | |
| else: | |
| updates["single_scenario_img"] = gr.update(value=None, visible=False) | |
| updates["single_scenario_caption_md"] = gr.update(value="", visible=False) | |
| updates["scenario_images_gallery"] = gr.update(value=[], visible=False) | |
| updates["scenario_images_captions_md"] = gr.update(value="", visible=False) | |
| overall_text = get_overall_reassessment_text(case_data_dict) | |
| updates["overall_reassessment_accordion"] = gr.update(visible=bool(overall_text.strip())) | |
| updates["overall_reassessment_text"] = gr.update(value=overall_text) | |
| return updates | |
| def update_mcq_components(case_data_dict, current_json_path, current_results): | |
| mcq_updates = {} | |
| mcqs_data = [] | |
| if case_data_dict and isinstance(case_data_dict.get("exam_creation"), dict): | |
| final_policy = case_data_dict['exam_creation'].get('final_policy', {}) | |
| mcqs_data = final_policy.get('mcqs', []) | |
| if not isinstance(mcqs_data, list): | |
| mcqs_data = [] | |
| num_mcqs = len(mcqs_data) | |
| file_results = current_results.get(current_json_path, {}) | |
| for i in range(MAX_MCQS_PER_FILE): | |
| mcq_slot = mcq_output_components[i] | |
| if i < num_mcqs: | |
| current_mcq_item_data = mcqs_data[i] | |
| mcq_idx = i | |
| # 使用翻译文本 | |
| stage_text = get_translated_text(current_mcq_item_data, 'stage', default_value='N/A') | |
| header_text = f"### 问题 {mcq_idx + 1} / {num_mcqs} <small>({stage_text})</small>" | |
| mcq_updates[mcq_slot["header"]] = gr.update(value=header_text) | |
| question_text = get_translated_text(current_mcq_item_data, 'question', default_value='问题文本为空。') | |
| mcq_updates[mcq_slot["question"]] = gr.update(value=question_text) | |
| # 处理选项 | |
| options = current_mcq_item_data.get('options_translated', current_mcq_item_data.get('options', {})) | |
| options_text_parts = ["**选项：**"] | |
| if options and isinstance(options, dict): | |
| original_keys = sorted([k for k in options.keys() if len(k) == 1 and k.isalpha()]) | |
| for key in original_keys: | |
| option_text = options.get(key, '') | |
| options_text_parts.append(f"- {key}. {html.escape(str(option_text))}") | |
| else: | |
| options_text_parts.append("未提供选项。") | |
| mcq_updates[mcq_slot["options"]] = gr.update(value="\n".join(options_text_parts)) | |
| # 正确答案 | |
| correct_ans_key = current_mcq_item_data.get('correct_answer', 'N/A') | |
| correct_ans_text = options.get(correct_ans_key, "未提供") if isinstance(options, dict) else "未提供" | |
| answer_text = f"**正确答案：** {correct_ans_key}. {html.escape(str(correct_ans_text))}" | |
| mcq_updates[mcq_slot["answer"]] = gr.update(value=answer_text) | |
| # 处理MCQ表格 | |
| mcq_tables_parts = [] | |
| mcq_tables_list = current_mcq_item_data.get('table_details', []) | |
| if isinstance(mcq_tables_list, list): | |
| for tbl_data in mcq_tables_list: | |
| if isinstance(tbl_data, dict): | |
| caption = get_translated_text(tbl_data, 'caption', default_value='相关表格') | |
| content = get_translated_text(tbl_data, 'table_translated', fallback_key='table', default_value='表格内容为空。') | |
| mcq_tables_parts.append(f"#### {html.escape(caption)}") | |
| mcq_tables_parts.append(content) | |
| mcq_updates[mcq_slot["tables"]] = gr.update(value="\n\n".join(mcq_tables_parts), visible=bool(mcq_tables_parts)) | |
| # 处理MCQ图片 | |
| mcq_img_list = current_mcq_item_data.get('image_details', []) | |
| if not isinstance(mcq_img_list, list): | |
| mcq_img_list = [] | |
| if len(mcq_img_list) == 1: | |
| img_data = mcq_img_list[0] | |
| img_url = resolve_image_url(img_data) | |
| caption = get_translated_caption(img_data, default_value=f'问题 {mcq_idx + 1} 图片') | |
| mcq_updates[mcq_slot["single_img"]] = gr.update(value=img_url, visible=bool(img_url)) | |
| mcq_updates[mcq_slot["single_img_caption"]] = gr.update(value=f"<p style='text-align:center; font-style:italic; margin-top:0px;'>{html.escape(caption)}</p>", visible=bool(img_url)) | |
| mcq_updates[mcq_slot["gallery_img"]] = gr.update(value=[], visible=False) | |
| mcq_updates[mcq_slot["gallery_img_caption"]] = gr.update(value="", visible=False) | |
| elif len(mcq_img_list) > 1: | |
| gallery_items = [] | |
| captions_list_md_parts = ["**图片说明详情：**"] | |
| for i_img, img_data in enumerate(mcq_img_list): | |
| if not isinstance(img_data, dict): | |
| continue | |
| img_url = resolve_image_url(img_data) | |
| full_caption = get_translated_caption(img_data, default_value=f'问题 {mcq_idx + 1} 图片 {i_img+1}') | |
| if img_url: | |
| gallery_items.append((img_url, html.escape(full_caption))) | |
| captions_list_md_parts.append(f"{i_img+1}. {html.escape(full_caption)}") | |
| mcq_updates[mcq_slot["single_img"]] = gr.update(value=None, visible=False) | |
| mcq_updates[mcq_slot["single_img_caption"]] = gr.update(value="", visible=False) | |
| mcq_updates[mcq_slot["gallery_img"]] = gr.update(value=gallery_items, visible=bool(gallery_items)) | |
| mcq_updates[mcq_slot["gallery_img_caption"]] = gr.update(value="\n".join(captions_list_md_parts), visible=bool(gallery_items)) | |
| else: | |
| mcq_updates[mcq_slot["single_img"]] = gr.update(value=None, visible=False) | |
| mcq_updates[mcq_slot["single_img_caption"]] = gr.update(value="", visible=False) | |
| mcq_updates[mcq_slot["gallery_img"]] = gr.update(value=[], visible=False) | |
| mcq_updates[mcq_slot["gallery_img_caption"]] = gr.update(value="", visible=False) | |
| # 评价文本 | |
| critic_text = get_critic_evaluation_text(case_data_dict, mcq_idx) | |
| mcq_updates[mcq_slot["critic_text"]] = gr.update(value=critic_text) | |
| mcq_updates[mcq_slot["critic_accordion"]] = gr.update(open=True) | |
| # 添加再评估文本 | |
| reassessment_text = get_reassessment_text(case_data_dict, mcq_idx) | |
| mcq_updates[mcq_slot["reassessment_text"]] = gr.update(value=reassessment_text) | |
| mcq_updates[mcq_slot["reassessment_accordion"]] = gr.update(open=True) | |
| # 加载保存的双重验证答案 | |
| saved_mcq_data = file_results.get(str(mcq_idx), {}) | |
| if isinstance(saved_mcq_data, dict): | |
| # 事实验证 | |
| factual_check = saved_mcq_data.get("factual_check") | |
| factual_comment = saved_mcq_data.get("factual_comment", "") | |
| mcq_updates[mcq_slot["factual_radio"]] = gr.update(value=factual_check if factual_check in ["正确", "错误"] else None) | |
| mcq_updates[mcq_slot["factual_comment"]] = gr.update(value=factual_comment, visible=factual_check == "错误") | |
| # 逻辑可解性 | |
| logical_check = saved_mcq_data.get("logical_check") | |
| logical_comment = saved_mcq_data.get("logical_comment", "") | |
| mcq_updates[mcq_slot["logical_radio"]] = gr.update(value=logical_check if logical_check in ["是", "否"] else None) | |
| mcq_updates[mcq_slot["logical_comment"]] = gr.update(value=logical_comment, visible=logical_check == "否") | |
| else: | |
| # 如果是旧数据格式或无数据 | |
| mcq_updates[mcq_slot["factual_radio"]] = gr.update(value=None) | |
| mcq_updates[mcq_slot["factual_comment"]] = gr.update(value="", visible=False) | |
| mcq_updates[mcq_slot["logical_radio"]] = gr.update(value=None) | |
| mcq_updates[mcq_slot["logical_comment"]] = gr.update(value="", visible=False) | |
| mcq_updates[mcq_slot["verification_accordion"]] = gr.update(visible=True) | |
| mcq_updates[mcq_slot["group"]] = gr.update(visible=True) | |
| else: | |
| mcq_updates[mcq_slot["group"]] = gr.update(visible=False) | |
| # 添加新组件的隐藏更新 | |
| mcq_updates[mcq_slot["reassessment_accordion"]] = gr.update(visible=False) | |
| mcq_updates[mcq_slot["reassessment_text"]] = gr.update(value="") | |
| mcq_updates[mcq_slot["verification_accordion"]] = gr.update(visible=False) | |
| mcq_updates[mcq_slot["factual_radio"]] = gr.update(value=None) | |
| mcq_updates[mcq_slot["factual_comment"]] = gr.update(value="", visible=False) | |
| mcq_updates[mcq_slot["logical_radio"]] = gr.update(value=None) | |
| mcq_updates[mcq_slot["logical_comment"]] = gr.update(value="", visible=False) | |
| return mcq_updates | |
| def collect_current_page_answers(current_filepath, current_results, *verification_values): | |
| """收集当前页面的双重验证答案""" | |
| if not current_filepath or current_filepath == "INVALID_PATH": | |
| print("Cannot collect answers: Invalid file path state.") | |
| return current_results | |
| updated_results = current_results.copy() | |
| if current_filepath not in updated_results: | |
| updated_results[current_filepath] = {} | |
| # 每个MCQ有4个值：factual_radio, factual_comment, logical_radio, logical_comment | |
| values_per_mcq = 4 | |
| num_mcqs = len(verification_values) // values_per_mcq | |
| for i in range(min(num_mcqs, MAX_MCQS_PER_FILE)): | |
| base_idx = i * values_per_mcq | |
| factual_radio = verification_values[base_idx] | |
| factual_comment = verification_values[base_idx + 1] | |
| logical_radio = verification_values[base_idx + 2] | |
| logical_comment = verification_values[base_idx + 3] | |
| # 只有当至少有一个评分时才保存 | |
| if factual_radio in ["正确", "错误"] or logical_radio in ["是", "否"]: | |
| updated_results[current_filepath][str(i)] = { | |
| "factual_check": factual_radio if factual_radio in ["正确", "错误"] else None, | |
| "factual_comment": factual_comment if factual_comment else "", | |
| "logical_check": logical_radio if logical_radio in ["是", "否"] else None, | |
| "logical_comment": logical_comment if logical_comment else "" | |
| } | |
| elif str(i) in updated_results[current_filepath]: | |
| # 如果两个都没有选择，删除该记录 | |
| del updated_results[current_filepath][str(i)] | |
| return updated_results | |
| # --- CSS --- | |
| custom_css = """ | |
| body, .gradio-container { | |
| font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, "Helvetica Neue", Arial, "Noto Sans SC", "PingFang SC", "Microsoft YaHei", sans-serif !important; | |
| } | |
| .mcq-card { | |
| border: 1px solid #e0e0e0; | |
| border-radius: 8px; | |
| padding: 15px; | |
| margin: 10px 0; | |
| background-color: #fafafa; | |
| } | |
| .nav-footer { | |
| margin-top: 20px; | |
| padding: 15px; | |
| background-color: #f5f5f5; | |
| border-radius: 8px; | |
| } | |
| .nav-info { | |
| color: #666; | |
| font-style: italic; | |
| margin: 5px 0; | |
| } | |
| .save-status { | |
| margin-top: 10px; | |
| padding: 8px; | |
| border-radius: 4px; | |
| } | |
| .caption { | |
| font-size: 0.9em; | |
| color: #666; | |
| margin-top: 5px; | |
| } | |
| .guideline-card { | |
| background: linear-gradient(135deg, #f5f7fa 0%, #c3cfe2 100%); | |
| border: 2px solid #4a90e2; | |
| border-radius: 12px; | |
| padding: 20px; | |
| margin: 15px 0; | |
| box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1); | |
| } | |
| .guideline-header { | |
| color: #2c3e50; | |
| font-size: 1.3em; | |
| font-weight: bold; | |
| margin-bottom: 15px; | |
| text-align: center; | |
| } | |
| .guideline-content { | |
| background: white; | |
| border-radius: 8px; | |
| padding: 15px; | |
| line-height: 1.6; | |
| } | |
| .guideline-section { | |
| margin: 10px 0; | |
| } | |
| .guideline-emphasis { | |
| color: #e74c3c; | |
| font-weight: bold; | |
| } | |
| /* Accordion标题样式优化 */ | |
| .verification-accordion .label-wrap { | |
| font-weight: bold !important; | |
| font-size: 1.1em !important; | |
| color: #2c3e50 !important; | |
| } | |
| /* 如果使用无边框设计 */ | |
| .borderless-accordion { | |
| border: none !important; | |
| box-shadow: none !important; | |
| } | |
| /* 提示文字样式 */ | |
| .small-hint { | |
| font-size: 0.85em !important; | |
| color: #666 !important; | |
| font-style: italic !important; | |
| margin-top: -10px !important; | |
| margin-bottom: 10px !important; | |
| } | |
| /* Accordion容器样式统一 */ | |
| .critic-accordion, .reassessment-accordion, .verification-accordion { | |
| border: 1px solid #e0e0e0 !important; | |
| border-radius: 8px !important; | |
| margin: 5px 0 !important; | |
| } | |
| """ | |
| # --- Gradio App --- | |
| # 先定义MCQ组件，以便其他函数可以引用 | |
| mcq_output_components = [] | |
| verification_inputs_list = [] | |
| with gr.Blocks(theme=gr.themes.Soft(primary_hue="blue", secondary_hue="indigo"), title="医疗案例审查系统", css=custom_css) as app: | |
| # --- State --- | |
| all_json_files_state = gr.State(value=[]) | |
| current_file_index_state = gr.State(value=-1) | |
| current_data_state = gr.State(value={}) | |
| current_filepath_state = gr.State(value="NOT_LOADED") | |
| results_state = gr.State(value={}) | |
| # --- UI Layout --- | |
| gr.Markdown("# 医疗案例审查与问题评估系统") | |
| # 添加审核指南展示 | |
| with gr.Column(elem_classes=["guideline-card"]): | |
| gr.Markdown("## 📋 审核指南", elem_classes=["guideline-header"]) | |
| with gr.Accordion("📖 详细审核指南（点击折叠/展开）", open=False): # open=False 表示默认折叠 | |
| with gr.Column(elem_classes=["guideline-content"]): | |
| gr.Markdown(""" | |
| ### 审核员身份定位 | |
| 您是医学案例问题答案验证员。您的任务是对每个问题进行双重严格验证。 | |
| #### 信息介绍 | |
| * **特权信息 (考生不可见):** 源案例报告 (案例描述 + 讨论)；图片详细说明文字；作为事实验证。 | |
| * **考生可见信息:** 当前问题 + 选项；图片 (无文字说明)；历史题目记录；作为逻辑推理依据。 | |
| #### 验证流程 | |
| * **检查1: 事实验证 (基于特权信息)** | |
| * **核心问题:** “这个题目在医学事实上是否准确？” | |
| * **具体关键:** 将问题内容与源案例报告对比，验证医学数据、前提和答案的准确性。 | |
| * **检查2: 逻辑可解性 (基于考生可见信息)** | |
| * **核心问题:** “仅凭可见信息，能否通过逻辑推理找到答案？” | |
| * **具体关键:** 仅基于临床场景、题目、图片 (无说明)，判断逻辑推理路径是否清晰，答案是否唯一。 | |
| #### 操作要求 | |
| * 每个问题独立完成两项检查。 | |
| * 选择 **“错误”** 或 **“否”** 时，需要详细说明原因和提供具体的改进建议。 | |
| #### 重要提醒 | |
| * 两个检查角度 **完全独立**，避免信息混淆。 | |
| * 以 **考生视角** 为核心评判逻辑可解性。 | |
| * AI 参考建议 **仅供参考**。 | |
| """) | |
| status_message_md = gr.Markdown(visible=False) | |
| with gr.Column(visible=False) as case_display_area: | |
| title_md = gr.Markdown() | |
| level_info_md = gr.Markdown(visible=False) | |
| ids_md = gr.Markdown() | |
| with gr.Accordion("源案例报告(题目依据，被测试者不可见)", open=True): | |
| with gr.Tab("案例描述"): | |
| case_text_md = gr.Markdown() | |
| with gr.Tab("案例讨论"): | |
| discussion_text_md = gr.Markdown() | |
| with gr.Accordion("案例通用图表与图片(总)", open=False): | |
| general_tables_display_md = gr.Markdown(visible=False) | |
| single_general_img = gr.Image(label="案例图片", visible=False, show_label=False, interactive=False, height=300) | |
| single_general_caption_md = gr.Markdown(visible=False, elem_classes=["caption"]) | |
| general_images_gallery_multi = gr.Gallery(label="案例图片 (多张)", show_label=False, columns=3, object_fit="contain", height="auto", visible=False) | |
| general_images_captions_md = gr.Markdown(label="图片说明列表", visible=False) | |
| # 在这里添加新的 Accordion | |
| with gr.Accordion("裁判智能体(o3)(总)", open=False): | |
| quality_assessment_md = gr.Markdown(visible=False) | |
| scoring_summary_md = gr.Markdown(visible=False) | |
| gr.HTML("<hr style='margin: 20px 0;'>") | |
| gr.Markdown("## 临床场景") | |
| with gr.Row(equal_height=False): | |
| with gr.Column(scale=3): | |
| scenario_text_md = gr.Markdown() | |
| with gr.Column(scale=2, min_width=300): | |
| single_scenario_img = gr.Image(label="场景图片", visible=False, show_label=False, interactive=False, height=300) | |
| single_scenario_caption_md = gr.Markdown(visible=False, elem_classes=["caption"]) | |
| scenario_images_gallery = gr.Gallery(label="场景图片", show_label=False, columns=1, object_fit="contain", height="auto", visible=False) | |
| scenario_images_captions_md = gr.Markdown(label="图片说明列表", visible=False) | |
| gr.HTML("<hr style='margin: 20px 0;'>") | |
| gr.Markdown("## 评估问题") | |
| with gr.Column() as mcq_dynamic_container: | |
| for i in range(MAX_MCQS_PER_FILE): | |
| slot_components = {} | |
| with gr.Group(visible=False, elem_classes=["mcq-card"]) as mcq_group: | |
| slot_components["group"] = mcq_group | |
| slot_components["header"] = gr.Markdown(f"### 问题 {i + 1}") | |
| with gr.Row(equal_height=False): | |
| with gr.Column(scale=3): | |
| slot_components["question"] = gr.Markdown("问题文本...") | |
| slot_components["options"] = gr.Markdown("- A. 选项A...") | |
| slot_components["answer"] = gr.Markdown("**正确答案：** A") | |
| slot_components["tables"] = gr.Markdown(visible=False) | |
| with gr.Column(scale=2, min_width=250): | |
| slot_components["single_img"] = gr.Image(show_label=False, height=250, interactive=False, visible=False) | |
| slot_components["single_img_caption"] = gr.Markdown(visible=False, elem_classes=["caption"]) | |
| slot_components["gallery_img"] = gr.Gallery(label="问题图片", show_label=False, columns=1, object_fit="contain", height="auto", visible=False) | |
| slot_components["gallery_img_caption"] = gr.Markdown(visible=False) | |
| slot_components["critic_accordion"] = gr.Accordion( | |
| label="🤖 评论智能体(o4-mini)【AI参考建议】", | |
| open=True | |
| ) | |
| with slot_components["critic_accordion"]: | |
| slot_components["critic_text"] = gr.Markdown("评价详情...") | |
| slot_components["reassessment_accordion"] = gr.Accordion( | |
| "⚖️ 裁判智能体(o3)【AI参考建议】(如果信息不完整请查看'裁判智能体(o3)(总)')", | |
| open=True | |
| ) | |
| with slot_components["reassessment_accordion"]: | |
| slot_components["reassessment_text"] = gr.Markdown("再评估详情...") | |
| # 评论智能体部分 | |
| # gr.Markdown("#### **🤖 评论智能体(o4-mini)【AI参考建议】**") | |
| # slot_components["critic_accordion"] = gr.Accordion("", open=True) | |
| # with slot_components["critic_accordion"]: | |
| # slot_components["critic_text"] = gr.Markdown("评价详情...") | |
| # # 裁判智能体部分 | |
| # gr.Markdown("#### **⚖️ 裁判智能体(o3)【AI参考建议】** <small>*(如果信息不完整请查看'裁判智能体(o3)(总)')*</small>") | |
| # slot_components["reassessment_accordion"] = gr.Accordion("", open=True) | |
| # with slot_components["reassessment_accordion"]: | |
| # slot_components["reassessment_text"] = gr.Markdown("再评估详情...") | |
| # 新的双重验证系统 | |
| # 在Accordion之前添加一个Markdown标题 | |
| gr.Markdown("#### **🔍 审核与验证**") # 独立标题 | |
| slot_components["verification_accordion"] = gr.Accordion( | |
| "", | |
| open=True, | |
| elem_classes=["verification-accordion"] | |
| ) | |
| with slot_components["verification_accordion"]: | |
| with gr.Column(): | |
| # 事实验证 | |
| gr.Markdown("#### 事实验证") | |
| slot_components["factual_radio"] = gr.Radio( | |
| label="事实验证正确或错误", | |
| choices=["正确", "错误"], | |
| value=None | |
| ) | |
| slot_components["factual_comment"] = gr.Textbox( | |
| label="指出事实错误，并依据【源报告】提出修改建议", | |
| placeholder="请详细说明事实性错误的地方以及建议的修改...", | |
| lines=2, | |
| visible=False | |
| ) | |
| gr.HTML("<hr style='margin: 10px 0;'>") | |
| # 逻辑可解性 | |
| gr.Markdown("#### 逻辑可解性") | |
| slot_components["logical_radio"] = gr.Radio( | |
| label="逻辑可解性是或否", | |
| choices=["是", "否"], | |
| value=None | |
| ) | |
| slot_components["logical_comment"] = gr.Textbox( | |
| label="从【考生视角】说明逻辑不通之处及修改建议", | |
| placeholder="请详细说明逻辑上的问题以及建议的修改...", | |
| lines=2, | |
| visible=False | |
| ) | |
| # 添加到输入列表 | |
| verification_inputs_list.extend([ | |
| slot_components["factual_radio"], | |
| slot_components["factual_comment"], | |
| slot_components["logical_radio"], | |
| slot_components["logical_comment"] | |
| ]) | |
| # 添加交互逻辑：条件性显示评论框 | |
| def create_factual_visibility_fn(): | |
| def update_factual_comment_visibility(choice): | |
| return gr.update(visible=(choice == "错误")) | |
| return update_factual_comment_visibility | |
| def create_logical_visibility_fn(): | |
| def update_logical_comment_visibility(choice): | |
| return gr.update(visible=(choice == "否")) | |
| return update_logical_comment_visibility | |
| slot_components["factual_radio"].change( | |
| fn=create_factual_visibility_fn(), | |
| inputs=[slot_components["factual_radio"]], | |
| outputs=[slot_components["factual_comment"]] | |
| ) | |
| slot_components["logical_radio"].change( | |
| fn=create_logical_visibility_fn(), | |
| inputs=[slot_components["logical_radio"]], | |
| outputs=[slot_components["logical_comment"]] | |
| ) | |
| gr.HTML("<hr style='border-top: 1px dashed #ccc; margin: 15px 0 5px 0;'>") | |
| mcq_output_components.append(slot_components) | |
| overall_reassessment_accordion = gr.Accordion("⚖️ 裁判智能体(o3)的总结【AI参考建议】", open=True, visible=False) | |
| with overall_reassessment_accordion: | |
| overall_reassessment_text = gr.Markdown("") | |
| with gr.Row(equal_height=False, elem_classes=["nav-footer"]): | |
| prev_btn = gr.Button("⬅️ 上一个文件", interactive=False) | |
| next_btn = gr.Button("下一个文件 ➡️", interactive=False) | |
| jump_to_file_input = gr.Number(label="跳转到文件 (序号)", minimum=1, step=1, precision=0, scale=1, interactive=False) | |
| jump_to_file_btn = gr.Button("跳转", scale=0, interactive=False) | |
| current_file_info_bottom = gr.Markdown("文件：0 / 0", elem_classes=["nav-info"]) | |
| save_btn = gr.Button("💾 保存所有结果") | |
| save_status_md = gr.Markdown("", elem_classes=["save-status"]) | |
| # --- All updatable outputs --- | |
| all_updatable_outputs = [ | |
| current_file_index_state, current_data_state, current_filepath_state, results_state, | |
| status_message_md, case_display_area, title_md, level_info_md, ids_md, case_text_md, | |
| discussion_text_md, general_tables_display_md, single_general_img, | |
| single_general_caption_md, general_images_gallery_multi, general_images_captions_md, quality_assessment_md, scoring_summary_md, | |
| scenario_text_md, single_scenario_img, single_scenario_caption_md, | |
| scenario_images_gallery, scenario_images_captions_md, save_status_md, | |
| current_file_info_bottom, prev_btn, next_btn, jump_to_file_input, jump_to_file_btn, | |
| overall_reassessment_accordion, overall_reassessment_text | |
| ] | |
| for slot in mcq_output_components: | |
| all_updatable_outputs.extend([ | |
| slot["group"], slot["header"], slot["question"], slot["options"], | |
| slot["answer"], slot["tables"], slot["single_img"], slot["single_img_caption"], | |
| slot["gallery_img"], slot["gallery_img_caption"], slot["critic_accordion"], | |
| slot["critic_text"], slot["reassessment_accordion"], slot["reassessment_text"], | |
| slot["verification_accordion"], slot["factual_radio"], slot["factual_comment"], | |
| slot["logical_radio"], slot["logical_comment"] | |
| ]) | |
| # --- Event Handlers --- | |
| def handle_navigation(direction, current_idx, all_files, current_results, current_filepath, *verification_values): | |
| if not all_files: | |
| return (current_idx, current_data_state.value if hasattr(current_data_state, 'value') else {}, current_filepath, current_results) + tuple([gr.update()] * (len(all_updatable_outputs) - 4)) | |
| updated_results = collect_current_page_answers(current_filepath, current_results, *verification_values) | |
| total_files = len(all_files) | |
| new_idx = (current_idx + direction + total_files) % total_files | |
| new_path = all_files[new_idx] | |
| new_data, loaded_path = load_case_data_from_hf(new_path) | |
| if loaded_path != new_path: | |
| new_path = loaded_path | |
| static_updates_dict = update_static_content(new_data, new_path) | |
| mcq_updates_dict = update_mcq_components(new_data, new_path, updated_results) | |
| nav_info_text = f"文件：{new_idx + 1} / {total_files} - {os.path.basename(new_path)}" if new_path and new_path != "INVALID_PATH" else "未加载文件" | |
| nav_info_update_bottom = gr.update(value=nav_info_text) | |
| combined_updates = [ | |
| new_idx, new_data, new_path, updated_results, | |
| static_updates_dict.get("status_message_md", gr.update()), | |
| static_updates_dict.get("case_display_area", gr.update()), | |
| static_updates_dict.get("title_md", gr.update()), | |
| static_updates_dict.get("level_info_md", gr.update()), | |
| static_updates_dict.get("ids_md", gr.update()), | |
| static_updates_dict.get("case_text_md", gr.update()), | |
| static_updates_dict.get("discussion_text_md", gr.update()), | |
| static_updates_dict.get("general_tables_display_md", gr.update()), | |
| static_updates_dict.get("single_general_img", gr.update()), | |
| static_updates_dict.get("single_general_caption_md", gr.update()), | |
| static_updates_dict.get("general_images_gallery_multi", gr.update()), | |
| static_updates_dict.get("general_images_captions_md", gr.update()), | |
| static_updates_dict.get("quality_assessment_md", gr.update()), | |
| static_updates_dict.get("scoring_summary_md", gr.update()), | |
| static_updates_dict.get("scenario_text_md", gr.update()), | |
| static_updates_dict.get("single_scenario_img", gr.update()), | |
| static_updates_dict.get("single_scenario_caption_md", gr.update()), | |
| static_updates_dict.get("scenario_images_gallery", gr.update()), | |
| static_updates_dict.get("scenario_images_captions_md", gr.update()), | |
| gr.update(value=""), nav_info_update_bottom, | |
| gr.update(interactive=total_files > 1), gr.update(interactive=total_files > 1), | |
| gr.update(maximum=total_files, interactive=total_files > 0), gr.update(interactive=total_files > 0), | |
| static_updates_dict.get("overall_reassessment_accordion", gr.update()), | |
| static_updates_dict.get("overall_reassessment_text", gr.update()), | |
| ] | |
| for slot in mcq_output_components: | |
| combined_updates.extend([ | |
| mcq_updates_dict.get(slot["group"], gr.update()), | |
| mcq_updates_dict.get(slot["header"], gr.update()), | |
| mcq_updates_dict.get(slot["question"], gr.update()), | |
| mcq_updates_dict.get(slot["options"], gr.update()), | |
| mcq_updates_dict.get(slot["answer"], gr.update()), | |
| mcq_updates_dict.get(slot["tables"], gr.update()), | |
| mcq_updates_dict.get(slot["single_img"], gr.update()), | |
| mcq_updates_dict.get(slot["single_img_caption"], gr.update()), | |
| mcq_updates_dict.get(slot["gallery_img"], gr.update()), | |
| mcq_updates_dict.get(slot["gallery_img_caption"], gr.update()), | |
| mcq_updates_dict.get(slot["critic_accordion"], gr.update()), | |
| mcq_updates_dict.get(slot["critic_text"], gr.update()), | |
| mcq_updates_dict.get(slot["reassessment_accordion"], gr.update()), | |
| mcq_updates_dict.get(slot["reassessment_text"], gr.update()), | |
| mcq_updates_dict.get(slot["verification_accordion"], gr.update()), | |
| mcq_updates_dict.get(slot["factual_radio"], gr.update()), | |
| mcq_updates_dict.get(slot["factual_comment"], gr.update()), | |
| mcq_updates_dict.get(slot["logical_radio"], gr.update()), | |
| mcq_updates_dict.get(slot["logical_comment"], gr.update()) | |
| ]) | |
| return tuple(combined_updates) | |
| def handle_jump_to_file(target_file_num_str, current_idx, all_files, current_data, current_results, current_filepath, *verification_values): | |
| if not all_files or target_file_num_str is None: | |
| return (current_idx, current_data, current_filepath, current_results) + tuple([gr.update()] * (len(all_updatable_outputs) - 4)) | |
| try: | |
| target_file_num = int(float(target_file_num_str)) | |
| except (ValueError, TypeError): | |
| error_message = "跳转失败：请输入有效的文件序号。" | |
| return (current_idx, current_data, current_filepath, current_results, gr.update(value=error_message, visible=True)) + tuple([gr.update()] * (len(all_updatable_outputs) - 5)) | |
| total_files = len(all_files) if all_files else 0 | |
| if not (1 <= target_file_num <= total_files) and total_files > 0 : | |
| error_message = f"跳转失败：文件序号必须在 1 到 {total_files} 之间。" | |
| return (current_idx, current_data, current_filepath, current_results, gr.update(value=error_message, visible=True)) + tuple([gr.update()] * (len(all_updatable_outputs) - 5)) | |
| elif total_files == 0: | |
| return (current_idx, current_data, current_filepath, current_results) + tuple([gr.update()] * (len(all_updatable_outputs) - 4)) | |
| updated_results = collect_current_page_answers(current_filepath, current_results, *verification_values) | |
| new_idx = target_file_num - 1 | |
| new_path = all_files[new_idx] | |
| new_data, loaded_path = load_case_data_from_hf(new_path) | |
| if loaded_path != new_path: | |
| new_path = loaded_path | |
| static_updates_dict = update_static_content(new_data, new_path) | |
| mcq_updates_dict = update_mcq_components(new_data, new_path, updated_results) | |
| nav_info_text = f"文件：{new_idx + 1} / {total_files} - {os.path.basename(new_path)}" if new_path and new_path != "INVALID_PATH" else "未加载文件" | |
| nav_info_update_bottom = gr.update(value=nav_info_text) | |
| combined_updates = [ | |
| new_idx, new_data, new_path, updated_results, | |
| static_updates_dict.get("status_message_md", gr.update()), | |
| static_updates_dict.get("case_display_area", gr.update()), | |
| static_updates_dict.get("title_md", gr.update()), | |
| static_updates_dict.get("level_info_md", gr.update()), | |
| static_updates_dict.get("ids_md", gr.update()), | |
| static_updates_dict.get("case_text_md", gr.update()), | |
| static_updates_dict.get("discussion_text_md", gr.update()), | |
| static_updates_dict.get("general_tables_display_md", gr.update()), | |
| static_updates_dict.get("single_general_img", gr.update()), | |
| static_updates_dict.get("single_general_caption_md", gr.update()), | |
| static_updates_dict.get("general_images_gallery_multi", gr.update()), | |
| static_updates_dict.get("general_images_captions_md", gr.update()), | |
| static_updates_dict.get("quality_assessment_md", gr.update()), | |
| static_updates_dict.get("scoring_summary_md", gr.update()), | |
| static_updates_dict.get("scenario_text_md", gr.update()), | |
| static_updates_dict.get("single_scenario_img", gr.update()), | |
| static_updates_dict.get("single_scenario_caption_md", gr.update()), | |
| static_updates_dict.get("scenario_images_gallery", gr.update()), | |
| static_updates_dict.get("scenario_images_captions_md", gr.update()), | |
| gr.update(value=""), nav_info_update_bottom, | |
| gr.update(interactive=total_files > 1), gr.update(interactive=total_files > 1), | |
| gr.update(maximum=total_files, interactive=total_files > 0), gr.update(interactive=total_files > 0), | |
| static_updates_dict.get("overall_reassessment_accordion", gr.update()), | |
| static_updates_dict.get("overall_reassessment_text", gr.update()) | |
| ] | |
| for slot in mcq_output_components: | |
| combined_updates.extend([ | |
| mcq_updates_dict.get(slot["group"], gr.update()), | |
| mcq_updates_dict.get(slot["header"], gr.update()), | |
| mcq_updates_dict.get(slot["question"], gr.update()), | |
| mcq_updates_dict.get(slot["options"], gr.update()), | |
| mcq_updates_dict.get(slot["answer"], gr.update()), | |
| mcq_updates_dict.get(slot["tables"], gr.update()), | |
| mcq_updates_dict.get(slot["single_img"], gr.update()), | |
| mcq_updates_dict.get(slot["single_img_caption"], gr.update()), | |
| mcq_updates_dict.get(slot["gallery_img"], gr.update()), | |
| mcq_updates_dict.get(slot["gallery_img_caption"], gr.update()), | |
| mcq_updates_dict.get(slot["critic_accordion"], gr.update()), | |
| mcq_updates_dict.get(slot["critic_text"], gr.update()), | |
| mcq_updates_dict.get(slot["reassessment_accordion"], gr.update()), | |
| mcq_updates_dict.get(slot["reassessment_text"], gr.update()), | |
| mcq_updates_dict.get(slot["verification_accordion"], gr.update()), | |
| mcq_updates_dict.get(slot["factual_radio"], gr.update()), | |
| mcq_updates_dict.get(slot["factual_comment"], gr.update()), | |
| mcq_updates_dict.get(slot["logical_radio"], gr.update()), | |
| mcq_updates_dict.get(slot["logical_comment"], gr.update()) | |
| ]) | |
| return tuple(combined_updates) | |
| def handle_save(current_filepath, current_results_state_val, *verification_values): | |
| results_gathered_this_call = collect_current_page_answers(current_filepath, current_results_state_val, *verification_values) | |
| message, data_intended_to_be_saved = save_results_to_hf(results_gathered_this_call) | |
| return results_gathered_this_call, gr.update(value=message) | |
| def populate_initial_ui_with_init(): | |
| """首次加载时初始化并更新UI""" | |
| # 初始化 | |
| all_files = find_json_files_from_hf() | |
| results = load_results_from_hf() | |
| if not all_files: | |
| data = create_error_data("未找到数据文件") | |
| path = "NO_FILES" | |
| idx = -1 | |
| else: | |
| idx = 0 | |
| path = all_files[0] | |
| data, path = load_case_data_from_hf(path) | |
| # 更新UI | |
| static_updates_dict = update_static_content(data, path) | |
| mcq_updates_dict = update_mcq_components(data, path, results) | |
| total_files = len(all_files) | |
| nav_info_text = f"文件：{idx + 1} / {total_files} - {os.path.basename(path)}" if idx >= 0 else "文件：0 / 0" | |
| combined_updates = [ | |
| idx, data, path, results, | |
| static_updates_dict.get("status_message_md", gr.update()), | |
| static_updates_dict.get("case_display_area", gr.update()), | |
| static_updates_dict.get("title_md", gr.update()), | |
| static_updates_dict.get("level_info_md", gr.update()), | |
| static_updates_dict.get("ids_md", gr.update()), | |
| static_updates_dict.get("case_text_md", gr.update()), | |
| static_updates_dict.get("discussion_text_md", gr.update()), | |
| static_updates_dict.get("general_tables_display_md", gr.update()), | |
| static_updates_dict.get("single_general_img", gr.update()), | |
| static_updates_dict.get("single_general_caption_md", gr.update()), | |
| static_updates_dict.get("general_images_gallery_multi", gr.update()), | |
| static_updates_dict.get("general_images_captions_md", gr.update()), | |
| static_updates_dict.get("quality_assessment_md", gr.update()), | |
| static_updates_dict.get("scoring_summary_md", gr.update()), | |
| static_updates_dict.get("scenario_text_md", gr.update()), | |
| static_updates_dict.get("single_scenario_img", gr.update()), | |
| static_updates_dict.get("single_scenario_caption_md", gr.update()), | |
| static_updates_dict.get("scenario_images_gallery", gr.update()), | |
| static_updates_dict.get("scenario_images_captions_md", gr.update()), | |
| gr.update(value=""), gr.update(value=nav_info_text), | |
| gr.update(interactive=total_files > 1), gr.update(interactive=total_files > 1), | |
| gr.update(maximum=total_files if total_files > 0 else 1, interactive=total_files > 0), | |
| gr.update(interactive=total_files > 0), | |
| static_updates_dict.get("overall_reassessment_accordion", gr.update()), | |
| static_updates_dict.get("overall_reassessment_text", gr.update()) | |
| ] | |
| for slot in mcq_output_components: | |
| combined_updates.extend([ | |
| mcq_updates_dict.get(slot["group"], gr.update()), | |
| mcq_updates_dict.get(slot["header"], gr.update()), | |
| mcq_updates_dict.get(slot["question"], gr.update()), | |
| mcq_updates_dict.get(slot["options"], gr.update()), | |
| mcq_updates_dict.get(slot["answer"], gr.update()), | |
| mcq_updates_dict.get(slot["tables"], gr.update()), | |
| mcq_updates_dict.get(slot["single_img"], gr.update()), | |
| mcq_updates_dict.get(slot["single_img_caption"], gr.update()), | |
| mcq_updates_dict.get(slot["gallery_img"], gr.update()), | |
| mcq_updates_dict.get(slot["gallery_img_caption"], gr.update()), | |
| mcq_updates_dict.get(slot["critic_accordion"], gr.update()), | |
| mcq_updates_dict.get(slot["critic_text"], gr.update()), | |
| mcq_updates_dict.get(slot["reassessment_accordion"], gr.update()), | |
| mcq_updates_dict.get(slot["reassessment_text"], gr.update()), | |
| mcq_updates_dict.get(slot["verification_accordion"], gr.update()), | |
| mcq_updates_dict.get(slot["factual_radio"], gr.update()), | |
| mcq_updates_dict.get(slot["factual_comment"], gr.update()), | |
| mcq_updates_dict.get(slot["logical_radio"], gr.update()), | |
| mcq_updates_dict.get(slot["logical_comment"], gr.update()) | |
| ]) | |
| # 更新状态 | |
| all_json_files_state.value = all_files | |
| return tuple(combined_updates) | |
| navigation_inputs = [current_file_index_state, all_json_files_state, results_state, current_filepath_state] + verification_inputs_list | |
| jump_inputs = [jump_to_file_input, current_file_index_state, all_json_files_state, current_data_state, results_state, current_filepath_state] + verification_inputs_list | |
| save_inputs = [current_filepath_state, results_state] + verification_inputs_list | |
| next_btn.click(fn=handle_navigation, inputs=[gr.State(value=1)] + navigation_inputs, outputs=all_updatable_outputs, queue=False) | |
| prev_btn.click(fn=handle_navigation, inputs=[gr.State(value=-1)] + navigation_inputs, outputs=all_updatable_outputs, queue=False) | |
| save_btn.click(fn=handle_save, inputs=save_inputs, outputs=[results_state, save_status_md]) | |
| jump_to_file_btn.click(fn=handle_jump_to_file, inputs=jump_inputs, outputs=all_updatable_outputs, queue=False) | |
| app.load( | |
| fn=populate_initial_ui_with_init, | |
| inputs=None, | |
| outputs=all_updatable_outputs | |
| ) | |
| # --- Launch --- | |
| if __name__ == "__main__": | |
| print("准备启动Gradio应用...") | |
| # 从环境变量获取认证信息 | |
| gradio_username = os.environ.get(GRADIO_USERNAME_ENV_VAR) | |
| gradio_password = os.environ.get(GRADIO_PASSWORD_ENV_VAR) | |
| auth = None | |
| if gradio_username and gradio_password: | |
| auth = (gradio_username, gradio_password) | |
| print("Gradio 认证已启用") | |
| else: | |
| print("Warning: 未找到Gradio认证环境变量，应用将在无保护模式下运行") | |
| app.launch(show_error=True, auth=auth) |