import gradio as gr import requests import re # --- 配置 --- # !!! 重要: 请将此 URL 替换为您的 API 服务器的实际地址 API_BASE_URL = "http://134.175.222.87:5006" # 这是一个示例，请务必修改 START_ANALYSIS_ENDPOINT = f"{API_BASE_URL}/start_analysis_session" ANALYZE_NEXT_ENDPOINT = f"{API_BASE_URL}/analyze_next" # --- 核心逻辑函数 --- def format_report(analysis_result): """将API返回的JSON格式化为美观的Markdown文本""" if not analysis_result or "ai_analysis" not in analysis_result: return "### ❌ 错误\n收到的分析结果格式不正确。" analysis = analysis_result["ai_analysis"] def create_tag_list(items): if not items: return "_暂无推断_" return " ".join([f"`{item}`" for item in items]) assessment = analysis.get("评估体系", {}) confidence = assessment.get("画像置信度", {}) identity_level = assessment.get("身份明确度", "未知") badge_map = {"低": "🟢", "中": "🟡", "高": "🔴"} identity_badge = f"{badge_map.get(identity_level, '⚪️')} {identity_level}" report_md = f""" ## 深度人物画像报告 > {analysis.get('核心摘要', 'AI未能生成摘要。')} --- #### 推断画像 - **职业角色**: {create_tag_list(analysis.get("推断画像", {}).get("职业角色"))} - **能力/技能**: {create_tag_list(analysis.get("推断画像", {}).get("能力技能"))} - **兴趣爱好**: {create_tag_list(analysis.get("推断画像", {}).get("兴趣爱好"))} --- #### 关联网络 - **关联人物**: {create_tag_list(analysis.get("关联网络", {}).get("关联人物"))} - **关联组织**: {create_tag_list(analysis.get("关联网络", {}).get("关联组织"))} --- #### 行为模式推断 {analysis.get('行为模式推断', '暂无明确的行为模式推断。')} --- #### 评估体系 - **身份明确度**: {identity_badge} - **画像总置信度**: **{confidence.get('总分', 0) * 100:.0f}%** - `职业置信度: {confidence.get('职业置信度', 0) * 100:.0f}%` - `爱好置信度: {confidence.get('爱好置信度', 0) * 100:.0f}%` - `关联网络置信度: {confidence.get('关联网络置信度', 0) * 100:.0f}%` """ return report_md def start_analysis(phone): """处理“生成画像”按钮点击事件""" if not phone or not re.match(r"^\d{7,15}$", phone): # 返回值对应 outputs 列表 return ( "号码格式不正确哦～", # status_message gr.update(visible=False), # report_area gr.update(value=""), # report_output gr.update(visible=False), # feedback_area gr.update(interactive=False), # correct_btn gr.update(interactive=False), # incorrect_btn None, 0, 0 # states ) status_message = "正在启动分析会话，请稍候..." # 初始加载状态 yield ( status_message, gr.update(visible=False), gr.update(value=""), gr.update(visible=False), gr.update(interactive=False), gr.update(interactive=False), None, 0, 0 ) try: response = requests.post(START_ANALYSIS_ENDPOINT, data={"phone": phone}, timeout=20) response.raise_for_status() data = response.json() if not data.get("success"): raise Exception(data.get("message", "API返回错误但未提供消息")) total_count = data.get("total_count", 0) if total_count == 0: yield ( "恭喜！您开启了隐身魔法呦👉👉", gr.update(visible=False), "", gr.update(visible=False), gr.update(interactive=False), gr.update(interactive=False), [], 0, 0 ) return references = data.get("data_references", []) analysis_result = data.get("analysis_result") report_md = format_report(analysis_result) status_message = f"发现 {total_count} 条线索，正在展示第 1 号情报..." next_btn_interactive = total_count > 1 # 成功获取数据后的最终状态 yield ( status_message, gr.update(visible=True), # report_area gr.update(value=report_md), # report_output gr.update(visible=True), # feedback_area gr.update(interactive=True), # correct_btn gr.update(interactive=next_btn_interactive), # incorrect_btn references, 1, total_count ) except (requests.exceptions.RequestException, Exception) as e: error_message = f"发生错误: {e}" yield ( error_message, gr.update(visible=False), "", gr.update(visible=False), gr.update(interactive=False), gr.update(interactive=False), None, 0, 0 ) def analyze_next_clue(references, current_index, total_count): """处理“换一条”按钮点击事件""" if not references or current_index >= total_count: yield ( "所有线索已分析完毕，没有更多信息了。", # status_message gr.update(), # report_output (保持不变) gr.update(interactive=False), # correct_btn (可设为True或False，这里设为False) gr.update(interactive=False), # incorrect_btn references, current_index, total_count ) return next_list_index = current_index reference_to_analyze = references[next_list_index] display_index = next_list_index + 1 status_message = f"正在分析第 {display_index} / {total_count} 号情报..." # 加载时禁用按钮 yield ( status_message, gr.update(), gr.update(interactive=False), gr.update(interactive=False), references, current_index, total_count ) try: response = requests.post(ANALYZE_NEXT_ENDPOINT, json={"reference": reference_to_analyze}, timeout=20) response.raise_for_status() data = response.json() if not data.get("success"): raise Exception(data.get("message", "API返回分析失败")) analysis_result = data.get("analysis_result") report_md = format_report(analysis_result) new_current_index = current_index + 1 status_message = f"第 {new_current_index} 号情报画像构建完毕！" next_btn_interactive = new_current_index < total_count yield ( status_message, gr.update(value=report_md), gr.update(interactive=True), gr.update(interactive=next_btn_interactive), references, new_current_index, total_count ) except (requests.exceptions.RequestException, Exception) as e: error_message = f"分析出错了: {e}" yield ( error_message, gr.update(), gr.update(interactive=True), gr.update(interactive=True), # 恢复按钮交互 references, current_index, total_count ) def confirm_analysis(): """处理“画像准确”按钮点击事件""" return ( "感谢您的确认！AI酱很开心。🎉", # status_message gr.update(interactive=False), # correct_btn gr.update(interactive=False) # incorrect_btn ) # --- Gradio 界面构建 --- with gr.Blocks(theme=gr.themes.Default(primary_hue="blue")) as demo: # 状态管理 state_references = gr.State([]) state_current_index = gr.State(0) state_total_count = gr.State(0) gr.Markdown("# AI深度人物画像系统") gr.Markdown("本系统利用AI对公开的互联网数据进行深度分析，构建推理性的人物画像...") with gr.Row(): phone_input = gr.Textbox(label="手机号", placeholder="输入11位数字...", scale=3) start_button = gr.Button("生成画像", variant="primary", scale=1) status_message = gr.Textbox(label="状态", value="AI分析引擎待命中...", interactive=False) with gr.Column(visible=False) as report_area: report_output = gr.Markdown() with gr.Row(visible=False) as feedback_area: correct_btn = gr.Button("✅ 画像准确") incorrect_btn = gr.Button("❌ 与事实不符, 换一条") # --- 事件绑定 --- # **核心改动**: 将所有需要更新的组件都列在outputs中 start_button.click( fn=start_analysis, inputs=[phone_input], outputs=[ status_message, report_area, report_output, feedback_area, correct_btn, incorrect_btn, state_references, state_current_index, state_total_count ] ) incorrect_btn.click( fn=analyze_next_clue, inputs=[state_references, state_current_index, state_total_count], outputs=[ status_message, report_output, correct_btn, incorrect_btn, state_references, state_current_index, state_total_count ] ) correct_btn.click( fn=confirm_analysis, inputs=[], outputs=[status_message, correct_btn, incorrect_btn] ) if __name__ == "__main__": demo.launch()