import json import os from datetime import datetime from pathlib import Path from typing import Dict, List, Optional def _build_selection_points_from_decode(decode_data: Dict) -> List[Dict]: """将 examples 解构内容 JSON 转为可视化「帖子选题表」所需的选题点行。""" result: List[Dict] = [] point_types = ["灵感点", "目的点", "关键点"] for point_type in point_types: points = decode_data.get(point_type, []) if not isinstance(points, list): continue for item in points: if not isinstance(item, dict): continue title = item.get("选题点") or item.get("点") or "" essence: List[str] = [] form: List[str] = [] intent: List[str] = [] for el in item.get("选题点元素") or []: if not isinstance(el, dict): continue name = el.get("元素名称") if not name: continue et = el.get("元素类型") or "" if et == "实质": essence.append(name) elif et == "形式": form.append(name) elif et == "意图": intent.append(name) else: essence.append(name) result.append( { "类型": point_type, "选题点": title, "实质": essence, "形式": form, "意图": intent, } ) return result def load_post_detail_for_visualization(account_name: str, post_id: str) -> Optional[Dict]: """ 从 Agent 示例目录读取原始帖子与解构内容，供「待解构帖子详情」弹窗与侧边栏使用。 - post_data: input/{account}/原始数据/post_data/{post_id}.json - 解构: input/{account}/原始数据/解构内容/{post_id}.json """ base = Path(__file__).resolve().parent post_path = base / "input" / account_name / "原始数据" / "post_data" / f"{post_id}.json" decode_path = base / "input" / account_name / "原始数据" / "解构内容" / f"{post_id}.json" try: with open(post_path, "r", encoding="utf-8") as f: post_data = json.load(f) except Exception: return None decode_data: Dict = {} try: with open(decode_path, "r", encoding="utf-8") as f: decode_data = json.load(f) except Exception: pass out = dict(post_data) out["选题点"] = _build_selection_points_from_decode(decode_data) if decode_data else [] pid = out.get("channel_content_id") or decode_data.get("帖子ID") if pid and not out.get("id"): out["id"] = pid return out def generate_all_in_one_visualization( data_map: Dict[str, dict], output_path: str, account_name: str, derivation_data: Dict[str, list] = None, post_detail_map: Dict[str, dict] = None, dimension_analyze_map: Dict[str, dict] = None, ): """ 将所有帖子的数据整合到一个 HTML 中，支持动态切换 data_map: { "文件名": json_data, ... } derivation_data: { "文件名": 推导结果列表, ... } post_detail_map: { "文件名": 帖子详情(含选题点)，来自 load_post_detail_for_visualization } dimension_analyze_map: { post_id: 整体推导维度分析 JSON（含 rounds.derived_dims 等）} """ # 提取第一个帖子的数据作为默认展示 first_key = list(data_map.keys())[0] # 将整个 data_map 转换为 JS 对象 json_data_js = json.dumps(data_map, ensure_ascii=False) # 将推导数据转换为 JS 对象 if derivation_data is None: derivation_data = {} derivation_data_js = json.dumps(derivation_data, ensure_ascii=False) # 将帖子详情数据转换为 JS 对象（供「待解构帖子」弹窗使用） if post_detail_map is None: post_detail_map = {} post_detail_map_js = json.dumps(post_detail_map, ensure_ascii=False) if dimension_analyze_map is None: dimension_analyze_map = {} dimension_analyze_data_js = json.dumps(dimension_analyze_map, ensure_ascii=False) account_name_js = json.dumps(account_name, ensure_ascii=False) html_content = rf''' 多源数据流可视化 - 完整全景版

多源数据流可视化 - 完整全景版

推导进度
未点亮
当前轮次点亮
之前已点亮
待解构帖子详情
维度 patterns
''' with open(output_path, 'w', encoding='utf-8') as f: f.write(html_content) print(f"最终结果可视化已生成: {output_path}") def main(account_name) -> None: name = account_name base = Path(__file__).resolve().parent output_base = base / "output" / name data_dir = output_base / "整体推导路径可视化" if not data_dir.exists(): print(f"错误: 找不到数据目录 {data_dir}") return json_files = sorted(f for f in os.listdir(data_dir) if f.endswith(".json")) if not json_files: print(f"在目录 {data_dir} 中未找到 .json 文件。") return data_map: Dict[str, dict] = {} print("\n" + "=" * 50) print(f"账号: {name}") print(f"数据目录: {data_dir}") print(f"正在读取 {len(json_files)} 个帖子数据...") for filename in json_files: json_path = data_dir / filename try: with open(json_path, "r", encoding="utf-8") as f: data_map[filename] = json.load(f) print(f" -> 已读取: {filename}") except Exception as e: print(f" [错误] 读取 {filename} 时出错: {e}") if not data_map: print("没有成功读取到任何数据。") return post_detail_map: Dict[str, dict] = {} for filename in data_map.keys(): post_id = Path(filename).stem try: detail = load_post_detail_for_visualization(name, post_id) if detail is not None: post_detail_map[filename] = detail except Exception as e: print(f" [警告] 加载帖子详情 {filename} 时出错: {e}") derivation_dir = output_base / "整体推导结果" derivation_data: Dict[str, list] = {} if derivation_dir.exists(): print("\n正在读取推导进度数据...") for json_file in derivation_dir.glob("*.json"): try: with open(json_file, "r", encoding="utf-8") as f: derivation_data[json_file.stem] = json.load(f) print(f" -> 已加载推导进度: {json_file.name}") except Exception as e: print(f" [警告] 读取推导进度 {json_file.name} 时出错: {e}") else: print(f" [提示] 推导结果目录不存在: {derivation_dir}") dimension_analyze_dir = output_base / "整体推导维度分析" dimension_analyze_map: Dict[str, dict] = {} if dimension_analyze_dir.exists(): print("\n正在读取整体推导维度分析...") suf = "_pattern_dimension_analyze" for json_file in sorted(dimension_analyze_dir.glob(f"*{suf}.json")): stem = json_file.stem if not stem.endswith(suf): continue post_id_key = stem[: -len(suf)] try: with open(json_file, "r", encoding="utf-8") as f: dimension_analyze_map[post_id_key] = json.load(f) print(f" -> 已加载维度分析: {json_file.name}") except Exception as e: print(f" [警告] 读取维度分析 {json_file.name} 时出错: {e}") else: print(f" [提示] 整体推导维度分析目录不存在: {dimension_analyze_dir}") output_base.mkdir(parents=True, exist_ok=True) ts = datetime.now().strftime("%Y%m%d%H%M%S") output_path = output_base / f"{name}_how推导可视化_{ts}.html" generate_all_in_one_visualization( data_map, str(output_path), name, derivation_data=derivation_data, post_detail_map=post_detail_map, dimension_analyze_map=dimension_analyze_map, ) print("\n" + "=" * 50) print("处理完成！") print(f"输出文件: {output_path}") print("=" * 50 + "\n") if __name__ == "__main__": main(account_name="空间点阵设计研究室")