{advice}"
# ================== LLM 教学建议---测试版 ==================
def generate_teaching_advice(sankey_fig, cluster_stats):
# 临时替换LLM调用（30秒内必完成）
prompt = """
你是一名GIS实验教学专家，基于以下分析结果，生成教学优化方案：
【桑基图分析】
- 学生反馈从 s1→s4 的主要流向：核密度 → 参数设置 → 应用场景
- 最粗路径：核密度分析 → 搜索半径选择 → 城市规划应用
【聚类分析】
"""
for s in cluster_stats[:3]:
prompt += f"- 聚类 {s['cluster_id']}：{s['keyword']}（{s['size']}条，占{s['ratio']:.1%}）\n"
prompt += f" 代表句：{s['rep_sentence'][:100]}\n"
prompt += """
【要求】
1. 诊断核心教学痛点（3条）
2. 提出针对性优化措施（微课/演示/作业）
3. 设计 1 个 2 分钟微课脚本（标题+3步演示）
4. 建议 1 个课后作业（验证学生掌握）
【输出格式】
# 教学优化方案
## 1. 核心痛点
## 2. 优化措施
## 3. 微课脚本
## 4. 课后作业
"""
return prompt, f"{prompt}"
# ================== 主分析函数 ==================
def analyze_report(excel_path):
if not excel_path:
return "", "请上传 Excel 文件", None, None, None, "", ""
try:
sankey_fig, sb64 = analyze_teacher_dashboard(excel_path=excel_path)
cluster_fig, cb64, cluster_stats = cluster_and_visualize(excel_path=excel_path)
prompt, msgs = generate_teaching_advice(sankey_fig, cluster_stats)
print(f"{sankey_fig}和{sb64}生成成功！")
print(f"{cluster_fig}和{cb64}生成成功！")
# 统计表格
stats_table = """
| 聚类 | 主题关键词 | 反馈数 | 占比 | 代表句 |
|---|---|---|---|---|
| {s['cluster_id']} | {s['keyword']} | {s['size']} | {s['ratio']:.1%} | {s['rep_sentence'][:80]}... |
GIS 教学智能体 · 循证教学优化
EGISInsight © 2025 | 从数据到教学内容改革
分析失败：{str(e)}
" return "", error_html, None, None, None, "", "" def analyze_report_plugin(excel_path): if not excel_path: return "", "请上传 Excel 文件", None, "", "" try: sankey_fig, sb64 = analyze_teacher_dashboard(excel_path=excel_path) cluster_fig, cb64, cluster_stats = cluster_and_visualize(excel_path=excel_path) prompt, msgs = generate_teaching_advice(sankey_fig, cluster_stats) print(f"{sankey_fig}和{sb64}生成成功！") print(f"{cluster_fig}和{cb64}生成成功！") # 统计表格 stats_table = """| 聚类 | 主题关键词 | 反馈数 | 占比 | 代表句 |
|---|---|---|---|---|
| {s['cluster_id']} | {s['keyword']} | {s['size']} | {s['ratio']:.1%} | {s['rep_sentence'][:80]}... |
GIS 教学智能体 · 循证教学优化
EGISInsight © 2025 | 从数据到教学内容改革
分析失败：{str(e)}
" return "", error_html, None, "", "" # 核心工具函数：把平台传递的纯文本 → DataFrame（不变） def parse_text_to_df(raw_text: str): """ 步骤： 1. 按制表符\t分割所有单元格（你的原有逻辑） 2. 过滤空单元格，得到纯数据列表 3. 按每行5列重新分组，不足5列的补None 4. 手动添加表头，生成DataFrame """ # 1. 清理原始文本（避免多余空字符干扰） cleaned_text = re.sub(r'\n+', '', raw_text.strip()) # 去掉所有换行（因为你按\t分割，换行没用） cleaned_text = re.sub(r'\t+', '\t', cleaned_text) # 合并连续制表符 cleaned_text = re.sub(r'^\t|\t$', '', cleaned_text) # 去掉首尾制表符 # 2. 按制表符分割，过滤空单元格（你的原有逻辑，稍作优化） lines = [] for cell in cleaned_text.split('\t'): cell_stripped = cell.strip() # 跳过空单元格（仅保留有内容的） if cell_stripped or cell_stripped == '0': # 避免把"0"当成空（如果有编号为0的情况） lines.append(cell_stripped) else: lines.append(None) # 空单元格用None填充，方便后续分组 # 关键修改：去掉第一个元素（sheet名，如ex01） if len(lines) > 0: lines = lines[1:] # 切片：保留从索引1开始的所有元素，去掉索引0（第一个元素） print(f"已去掉sheet名，剩余元素数量：{len(lines)}") else: raise Exception("分割后无有效数据，无法解析") # 关键修改2：再去掉第一个元素（原始表头 no/s1/s2/s3/s4） if len(lines) > 5 and lines[0] == 'no': # 确认第一个元素是表头，再删除 lines = lines[5:] else: # 若没有原始表头（极端情况），直接继续（后续用我们的固定表头） pass # 3. 核心：按每行5列重新分组（关键逻辑） header = ['no', 's1', 's2', 's3', 's4'] # 固定表头 data_rows = [] # 存储分组后的每行数据 # 计算需要分多少组（向上取整，避免遗漏最后几个元素） total_cells = len(lines[1:]) total_rows = (total_cells + 4) // 5 # 向上取整（比如11个元素→3行：5+5+1） # 按5个元素一组拆分，不足5列的补None for i in range(total_rows): # 截取当前行的5个元素（左闭右开区间） start_idx = i * 5 end_idx = start_idx + 5 row = lines[start_idx:end_idx] # 不足5列的补None（确保每行都是5列） while len(row) < 5: row.append(None) data_rows.append(row) # 4. 生成DataFrame，清理无效数据 df = pd.DataFrame(data_rows, columns=header) # 清理：去掉no列为空的行、s1-s4全为空的行 df = df.dropna(subset=['no'], how='any') df = df.dropna(subset=['s1', 's2', 's3', 's4'], how='all') return df # ================== 完美兼容版界面（老版本也能居中 + 控制高度）================== def create_demo(): with gr.Blocks(theme=gr.themes.Soft(), title="EGISInsight") as demo: gr.Markdown("# GIS实验报告智能分析系统") gr.Markdown("**上传学生反馈 Excel → 1秒生成教学决策图 + AI教案**") file_input = gr.File(label="上传 ex02.xlsx（需含 s1-s4 列）", file_types=[".xlsx"]) # 标题 + 桑基图 gr.HTML('通义千问大模型实时生成教学优化方案
") return demo def create_demo_plugin(): with gr.Blocks(theme=gr.themes.Soft(), title="EGISInsight") as demo: gr.Markdown("# GIS实验报告智能分析系统") gr.Markdown("**上传学生反馈 Excel → 1秒生成教学决策图 + AI教案**") file_input = gr.File(label="上传 ex02.xlsx（需含 s1-s4 列）", file_types=[".xlsx"]) # 标题 + 桑基图（完美居中 + 固定高度） gr.HTML('通义千问大模型实时生成教学优化方案
") return demo # if __name__ == "__main__": # demo.launch(server_name="0.0.0.0", share=True) # ================== 大赛平台专用 API（最终修正版）================== from fastapi import FastAPI, File, UploadFile, Form from fastapi.responses import FileResponse import uvicorn from io import BytesIO app = FastAPI() # 配置临时图片目录（确保存在） IMAGE_DIR = "./temp_images" os.makedirs(IMAGE_DIR, exist_ok=True) # ================== 1. 图片访问接口（修复路径匹配）================== @app.get("/temp_images/{img_name}") async def get_temp_image(img_name: str): """通过 URL 访问图片（确保路径正确）""" # 用绝对路径查找图片（避免相对路径混乱） img_path = os.path.abspath(os.path.join(IMAGE_SAVE_DIR, img_name)) print(f"🔍 正在查找图片：{img_path}") # 检查图片是否存在 if not os.path.exists(img_path): print(f"❌ 图片不存在：{img_path}") return Response(status_code=404, content="图片不存在") # 检查文件是否是有效图片（避免目录遍历攻击） if not img_name.endswith((".png", ".jpg", ".jpeg")): return Response(status_code=403, content="无效的图片格式") # 返回图片文件 return FileResponse(img_path, media_type="image/png") # ================== 2. 辅助函数：base64转图片并生成URL（修复URL拼接）================== def base64_to_public_url(base64_str: str, img_name: str) -> str: """ 功能：将base64字符串保存为图片，返回可公网访问的URL 修复：正确拼接端口和路径分隔符 """ if not base64_str: print("⚠️ 空的base64字符串，跳过图片生成") return "" try: # 去掉base64前缀（如果有） if base64_str.startswith("data:image/png;base64,"): base64_str = base64_str.split(",")[1] # 解码base64（处理可能的空格/换行） # base64_str = base64_str.strip().replace("\n", "").replace(" ", "") # img_data = base64.b64decode(base64_str) processed = ( base64_str.strip().replace("\n", "").replace("\r", "").replace(" ", "").replace("-", "+").replace("_", "/") ) # 兼容 URL 安全 Base64 # 补全 Base64 填充符（避免 Incorrect padding 错误） padding = 4 - (len(processed) % 4) if padding != 4: processed += "=" * padding img_data = base64.b64decode(processed) # 保存图片到绝对路径 img_path = os.path.abspath(os.path.join(IMAGE_SAVE_DIR, img_name)) with open(img_path, "wb") as f: f.write(img_data) print(f"✅ 图片保存成功：{img_path}（大小：{len(img_data)/1024:.1f}KB）") # 生成正确的公网URL（含端口和路径分隔符） public_url = f"{IMAGE_BASE_URL}{img_name}" print(f"✅ 图片公网URL：{public_url}") return public_url except Exception as e: print(f"❌ 图片生成失败：{str(e)}") return "" # ================== 桑基图转公网URL函数（无需传参，读取全局配置）================== def plotly_to_url(fig: go.Figure,IMG_N: str) -> str: """ 无需传参！读取全局配置，将Plotly桑基图转公网可访问URL 自动处理：中文字体修复、图片保存、URL生成 参数： fig: Plotly桑基图对象（go.Figure） 返回： 成功：公网URL字符串；失败：None """ # 1. 校验输入 if not isinstance(fig, go.Figure): print(f"❌ 输入不是Plotly Figure对象（类型：{type(fig)}）") return None try: # 2. 生成唯一图片名（防重复，用时间戳+固定前缀） import time timestamp = time.strftime("%Y%m%d%H%M%S", time.localtime()) img_name = f"{IMG_N}_{timestamp}.png" # 示例：sankey_gis_20251123153020.png # 3. 强制设置中文字体（解决中文方块问题） fig.update_layout( title=dict( text=fig.layout.title.text if fig.layout.title else "GIS实践教学改革方向捕捉", font=dict(family="SimHei", size=22) ), font=dict(family="SimHei", size=18), # 全局字体 autosize=False, width=1186, # 匹配日志中的宽度 height=798 # 匹配日志中的高度 ) # 单独设置桑基图节点标签字体（防止节点中文失效） for trace in fig.data: if hasattr(trace, "node"): trace.node.font = dict(family="SimHei", size=18) # 4. 保存图片（使用全局配置的目录） img_path = os.path.abspath(os.path.join(IMAGE_SAVE_DIR, img_name)) # 转绝对路径更稳妥 fig.write_image(img_path, engine="kaleido") print(f"✅ 图片保存成功：{img_path}（大小：{os.path.getsize(img_path)/1024:.1f}KB）") # 5. 生成公网URL（读取全局配置的前缀，处理 '/' 拼接） public_url = f"{IMAGE_BASE_URL.rstrip('/')}/{img_name.lstrip('/')}" print(f"✅ 图片公网URL：{public_url}") return public_url except Exception as e: print(f"❌ 桑基图转URL失败：{str(e)}") # 针对常见错误给出提示 if "font" in str(e).lower(): print(f"📌 建议：尝试更换中文字体为 'WenQuanYi Zen Hei'（适配Linux环境）") elif "engine" in str(e).lower(): print(f"📌 建议：将 engine 改为 'orca'，并执行 conda install -c plotly plotly-orca 安装") return None @app.post("/api/plugin") async def plugin_api( file: str = Form(...), token: str = Form(...), timestamp: str = Form(...), signature: str = Form(...) ): # print(file) try: # print("file 前10个字符:", file[:10]) # re.match(r'^\s*\{"file"', file) if signature=='doc': print(f"走到这步了！！！！！！！！！！") eval, prob1, prob2 = calcDocSims(file) prompt = f" 你是一名GIS实验教学专家/名师，对于当前教学优化效果（优化效果值{eval}也要反馈给用户）作出评价：" if eval>0.45 and eval <= 1: prompt += """ 当前教学优化效果显著，说点鼓励教师的话。""" elif eval >0.25 and eval <= 0.45: prompt += f" 当前教学优化效果尚可，但最好结合{prob1}的内容，再凝练3条可调整优化内容，比如措施--除了微课、演示以外的方式。" else: prompt = prompt + f" 当前教学优化效果一般，请根据{prob1}的内容，做进一步调整优化方案。" print(prompt) return { "code": "200", # 插件定义为 String 类型 "message": "success", # 插件定义为 String 类型 "prompt": prompt } else: if file == '1': temp_file_path = './ex01.xlsx' else: # print(type(file), file) df = parse_text_to_df(file) print(f"解析出 {len(df)} 条有效反馈") # 关键：创建本地临时Excel文件（自动生成真实路径，用完删除） with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as temp_file: df.to_excel(temp_file, index=False) # 写入临时文件 temp_file_path = temp_file.name # 拿到临时文件的真实路径（比如：C:\Users\XXX\AppData\Local\Temp\tmpxxxx.xlsx） # print(df) # 调用分析函数（传临时文件的真实路径） prompt, html_report, stats, sb64, cb64 = analyze_report_plugin(temp_file_path) # prompt, html_report, sankey_fig, cluster_fig, stats, sb64, cb64 = analyze_report(temp_file_path) # print(f"{sankey_fig}生成成功！") # print(f"{cluster_fig}生成成功！") # 用完删除临时文件（避免占用空间） if file == '1': print('本地测试！') else: os.remove(temp_file_path) print(f"临时文件已删除：{temp_file_path}") # 生成图片公网URL（关键修复：用正确的URL拼接） timestamp = int(time.time() * 1000) sankey_name = f"sankey_{timestamp}.png" sankey_url = base64_to_public_url(sb64, sankey_name) timestamp = int(time.time() * 1000) cluster_name = f"cluster_{timestamp}.png" cluster_url = base64_to_public_url(cb64, cluster_name) # sankey_url = plotly_to_url(sankey_fig, "sankey") # cluster_url = plotly_to_url(cluster_fig, "cluster") print(f"✅ 图片URL生成：桑基图={sankey_url}，聚类图={cluster_url}") # 按平台要求返回结果 # 按平台要求返回结果（用 sb64/cb64 替换 to_json()） print(f"{sb64}生成成功！") print(f"{cb64}生成成功！") # 3. 格式化 stats 字段（严格匹配插件定义的子字段） formatted_stats = [] if stats and isinstance(stats, list): # 确保 stats 是数组 for idx, s in enumerate(stats): # 严格匹配平台定义的5个字段：cluster_id、keyword、size、ratio、rep_sentence stat_item = { "cluster_id": str(s.get("cluster_id", idx)), # String 类型 "keyword": str(s.get("keyword", "无")), # String 类型 "size": int(s.get("size", 0)), # Number 类型 "ratio": f"{s.get('ratio', 0):.1%}" if s.get('ratio') is not None else "0.0%", # String 类型（如 "30.5%"） "rep_sentence": str(s.get("rep_sentence", "无")[:200]) # String 类型，截取200字 } # 过滤空值（必填字段确保有默认值） for key in list(stat_item.keys()): if stat_item[key] == "无" and key in ["cluster_id", "keyword", "rep_sentence"]: stat_item[key] = "无数据" elif stat_item[key] == 0 and key == "size": stat_item[key] = 0 formatted_stats.append(stat_item) else: formatted_stats = [] return { "code": "200", # 插件定义为 String 类型 "message": "success", # 插件定义为 String 类型 "data": { "html_report": html_report.strip() if html_report else "", # 对应 html_report 字段 "sankey_fig": sankey_url, # 对应 sankey_fig 字段（图片URL，String类型） "cluster_fig": cluster_url, # 对应 cluster_fig 字段（图片URL，String类型） "stats": formatted_stats, # 对应 stats 字段（Array/Object类型） "prompt":prompt } } except Exception as e: error_msg = f"分析失败：{str(e)}" print(error_msg) return { "code": 500, "message": error_msg, "data": { "images": [], "html_report": f"{error_msg}
", "stats": [], "teaching_advice": "" } } # ================== 以下是你的原有函数（不变，确保正常调用）================== # LLM教学建议函数（generate_teaching_advice1 / generate_teaching_advice） # 主分析函数（analyze_report） # 数据解析函数（parse_text_to_df） # 桑基图生成函数（plot_sankey_from_df） # Gradio界面配置（demo） # if DEBUG: # # 本地模式：仅启动 Gradio 服务（不用 uvicorn） # if __name__ == "__main__": # demo = create_demo() # demo.launch(server_name="0.0.0.0", share=True) # else: # # 线上 HF 模式：启动 FastAPI+Gradio 整合服务（用 uvicorn） # demo = create_demo_plugin() # app = gr.mount_gradio_app(app, demo, path="/gradio") # if __name__ == "__main__": # print(f"🚀 服务启动：{PLUGIN_DOMAIN}:{PLUGIN_PORT}") # print(f"📌 API路径：{PLUGIN_DOMAIN}:{PLUGIN_PORT}/api/plugin") # print(f"🖼️ 图片访问路径：{PLUGIN_DOMAIN}:{PLUGIN_PORT}/temp_images/") # uvicorn.run(app, host="0.0.0.0", port=PLUGIN_PORT) # 保留原界面 demo = create_demo_plugin() app = gr.mount_gradio_app(app, demo, path="/hidden-gradio") if __name__ == "__main__": # 启动 Gradio 原型（端口 7860，仅本地/你自己访问） # demo = create_demo() # demo.launch(server_name="0.0.0.0", share=True) # gradio_thread = threading.Thread( # target=lambda: demo.launch( # server_name="0.0.0.0", # server_port=7860, # 非大赛端口，避免冲突 # share=False # 若需要公网访问，可设为 True，但会生成独立链接 # ), # daemon=True # 主程序退出时自动关闭 # ) # gradio_thread.start() print(f"🚀 服务启动：{PLUGIN_DOMAIN}:{PLUGIN_PORT}") print(f"📌 API路径：{PLUGIN_DOMAIN}:{PLUGIN_PORT}/api/plugin") print(f"🖼️ 图片访问路径：{PLUGIN_DOMAIN}:{PLUGIN_PORT}/temp_images/") uvicorn.run(app, host="0.0.0.0", port=PLUGIN_PORT)