#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 路径配置管理工具 提供统一的路径管理，支持多账号批量处理 """ import json from pathlib import Path from typing import Dict, Optional, List import os class PathConfig: """路径配置管理类""" def __init__(self, account_name: Optional[str] = None, output_version: Optional[str] = None): """ 初始化路径配置 Args: account_name: 账号名称，如果不指定则使用默认账号或环境变量 output_version: 输出版本，如果不指定则使用项目根目录名称 """ # 获取项目根目录 self.project_root = Path(__file__).parent.parent.parent self.config_file = self.project_root / "config" / "accounts.json" # 加载配置 self._load_config() # 获取数据根目录 self.data_root = self._get_data_root() # 确定账号名称 self.account_name = self._determine_account_name(account_name) # 确定输出版本（默认使用项目根目录名） self.output_version = self._determine_output_version(output_version) # 构建路径 account_base = self.config["paths"]["account_base"] self.account_dir = self.data_root / account_base / self.account_name def _load_config(self): """加载配置文件""" if not self.config_file.exists(): raise FileNotFoundError(f"配置文件不存在: {self.config_file}") with open(self.config_file, "r", encoding="utf-8") as f: self.config = json.load(f) def _get_data_root(self) -> Path: """ 获取数据根目录 优先级： 1. 环境变量 DATA_ROOT 2. 配置文件 data_root 3. 默认值 project_root/data（向后兼容） """ # 1. 环境变量 data_root = os.environ.get("DATA_ROOT") if data_root: return Path(os.path.expanduser(data_root)) # 2. 配置文件 data_root_config = self.config.get("data_root") if data_root_config: # 支持 ~ 和环境变量 expanded = os.path.expandvars(os.path.expanduser(data_root_config)) path = Path(expanded) if path.is_absolute(): return path else: return self.project_root / path # 3. 默认值（向后兼容） return self.project_root / "data" def _determine_account_name(self, account_name: Optional[str]) -> str: """ 确定要使用的账号名称 优先级： 1. 函数参数指定的账号名 2. 环境变量 ACCOUNT_NAME 3. 配置文件中的默认账号 Args: account_name: 参数指定的账号名 Returns: 最终确定的账号名称 """ # 1. 参数指定 if account_name: return account_name # 2. 环境变量 env_account = os.environ.get("ACCOUNT_NAME") if env_account: return env_account # 3. 配置文件默认值 default_account = self.config.get("default_account") if default_account: return default_account # 4. 如果都没有，抛出错误 raise ValueError( "未指定账号名称！请通过以下方式之一指定：\n" "1. 参数: PathConfig(account_name='账号名')\n" "2. 环境变量: export ACCOUNT_NAME='账号名'\n" "3. 配置文件: 在 config/accounts.json 中设置 default_account" ) def _determine_output_version(self, output_version: Optional[str]) -> str: """ 确定输出版本 优先级： 1. 函数参数 2. 环境变量 OUTPUT_VERSION 3. 配置文件中的 output_version 4. 项目根目录名称（默认） """ # 1. 参数指定 if output_version: return output_version # 2. 环境变量 env_version = os.environ.get("OUTPUT_VERSION") if env_version: return env_version # 3. 配置文件指定 config_version = self.config.get("output_version") if config_version: return config_version # 4. 使用项目根目录名称（默认） project_dir_name = self.project_root.name return project_dir_name def _replace_version_var(self, path_template: str) -> str: """替换路径模板中的 {version} 变量""" return path_template.replace("{version}", self.output_version) def get_enabled_accounts(self) -> List[str]: """获取所有启用的账号列表""" accounts = self.config.get("accounts", []) return [acc["name"] for acc in accounts if acc.get("enabled", True)] def get_all_accounts(self) -> List[str]: """获取所有账号列表（包括未启用的）""" accounts = self.config.get("accounts", []) return [acc["name"] for acc in accounts] @property def filter_mode(self) -> str: """ 获取过滤模式 Returns: 过滤模式名称： - "exclude_current_posts": 过滤当前帖子ID（默认，推荐） - "time_based": 基于时间过滤 - "none": 不过滤 """ return self.config.get("filter_mode", "exclude_current_posts") # ===== 输入路径 ===== @property def current_posts_dir(self) -> Path: """当前帖子what解构结果目录""" rel_path = self.config["paths"]["input"]["current_posts"] return self.account_dir / rel_path @property def historical_posts_dir(self) -> Path: """过去帖子what解构结果目录""" rel_path = self.config["paths"]["input"]["historical_posts"] return self.account_dir / rel_path @property def pattern_cluster_file(self) -> Path: """pattern聚合结果文件""" rel_path = self.config["paths"]["input"]["pattern_cluster"] return self.account_dir / rel_path # ===== 输出路径 ===== @property def intermediate_dir(self) -> Path: """中间结果目录""" rel_path = self.config["paths"]["output"]["intermediate"] rel_path = self._replace_version_var(rel_path) return self.account_dir / rel_path @property def feature_category_mapping_file(self) -> Path: """特征名称_分类映射.json""" return self.intermediate_dir / "特征名称_分类映射.json" @property def category_hierarchy_file(self) -> Path: """分类层级映射.json""" return self.intermediate_dir / "分类层级映射.json" @property def feature_source_mapping_file(self) -> Path: """特征名称_帖子来源.json""" return self.intermediate_dir / "特征名称_帖子来源.json" @property def task_list_file(self) -> Path: """当前帖子_解构任务列表.json""" return self.intermediate_dir / "当前帖子_解构任务列表.json" @property def how_results_dir(self) -> Path: """how解构结果目录""" rel_path = self.config["paths"]["output"]["how_results"] rel_path = self._replace_version_var(rel_path) return self.account_dir / rel_path @property def visualization_dir(self) -> Path: """可视化结果目录""" rel_path = self.config["paths"]["output"]["visualization"] rel_path = self._replace_version_var(rel_path) return self.account_dir / rel_path @property def visualization_file(self) -> Path: """可视化HTML文件""" return self.visualization_dir / "how解构结果_可视化.html" # ===== 工具方法 ===== def ensure_dirs(self): """确保所有输出目录存在""" self.intermediate_dir.mkdir(parents=True, exist_ok=True) self.how_results_dir.mkdir(parents=True, exist_ok=True) self.visualization_dir.mkdir(parents=True, exist_ok=True) def validate_input_paths(self) -> Dict[str, bool]: """ 验证输入路径是否存在 Returns: 验证结果字典 """ results = { "当前帖子目录": self.current_posts_dir.exists(), "过去帖子目录": self.historical_posts_dir.exists(), "pattern聚合文件": self.pattern_cluster_file.exists(), } return results def print_paths(self): """打印所有路径信息（用于调试）""" print("="*60) print(f"项目根目录: {self.project_root}") print(f"项目名称: {self.project_root.name}") print(f"数据根目录: {self.data_root}") print(f"输出版本: {self.output_version}") print(f"账号: {self.account_name}") print(f"过滤模式: {self.filter_mode}") print(f"账号根目录: {self.account_dir}") print("\n输入路径:") print(f" 当前帖子目录: {self.current_posts_dir}") print(f" 过去帖子目录: {self.historical_posts_dir}") print(f" pattern聚合文件: {self.pattern_cluster_file}") print("\n输出路径:") print(f" 中间结果目录: {self.intermediate_dir}") print(f" how解构结果目录: {self.how_results_dir}") print(f" 可视化结果目录: {self.visualization_dir}") print("="*60) def check_and_print_status(self): """检查并打印路径状态""" self.print_paths() print("\n输入路径验证:") validation = self.validate_input_paths() for name, exists in validation.items(): status = "✓ 存在" if exists else "✗ 不存在" print(f" {name}: {status}") if not all(validation.values()): print("\n⚠️ 警告: 部分输入路径不存在！") return False else: print("\n✓ 所有输入路径验证通过") return True def get_path_config(account_name: Optional[str] = None) -> PathConfig: """ 获取路径配置对象（便捷函数） Args: account_name: 账号名称，可选 Returns: PathConfig对象 """ return PathConfig(account_name) if __name__ == "__main__": # 测试代码 import sys account = sys.argv[1] if len(sys.argv) > 1 else None try: config = PathConfig(account) config.check_and_print_status() print("\n所有启用的账号:") for acc in config.get_enabled_accounts(): print(f" - {acc}") except Exception as e: print(f"错误: {e}") sys.exit(1)