how/script/data_processing/path_config.py

#!/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_account_config(self, account_name: str) -> Optional[Dict]:
        """获取特定账号的配置"""
        accounts = self.config.get("accounts", [])
        for acc in accounts:
            if acc["name"] == account_name:
                return acc
        return None

    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")

    # ===== 输入路径 =====

    def _get_input_path(self, path_key: str) -> str:
        """
        获取输入路径配置，支持账号级别的自定义路径

        优先级：
        1. 账号特定配置 (accounts[x].paths.input.path_key)
        2. 全局默认配置 (paths.input.path_key)
        """
        # 1. 检查账号特定配置
        account_config = self._get_account_config(self.account_name)
        if account_config and "paths" in account_config:
            account_paths = account_config["paths"]
            if "input" in account_paths and path_key in account_paths["input"]:
                return account_paths["input"][path_key]

        # 2. 使用全局默认配置
        return self.config["paths"]["input"][path_key]

    @property
    def current_posts_dir(self) -> Path:
        """当前帖子what解构结果目录"""
        rel_path = self._get_input_path("current_posts")
        return self.account_dir / rel_path

    @property
    def historical_posts_dir(self) -> Path:
        """过去帖子what解构结果目录"""
        rel_path = self._get_input_path("historical_posts")
        return self.account_dir / rel_path

    @property
    def pattern_cluster_file(self) -> Path:
        """pattern聚合结果文件"""
        rel_path = self._get_input_path("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)