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):
        """
        初始化路径配置

        Args:
            account_name: 账号名称，如果不指定则使用默认账号或环境变量
        """
        # 获取项目根目录
        self.project_root = Path(__file__).parent.parent.parent
        self.config_file = self.project_root / "config" / "accounts.json"

        # 加载配置
        self._load_config()

        # 确定账号名称
        self.account_name = self._determine_account_name(account_name)

        # 构建路径
        account_base = self.config["paths"]["account_base"]
        self.account_dir = self.project_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 _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 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"]
        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"]
        return self.account_dir / rel_path

    @property
    def visualization_dir(self) -> Path:
        """可视化结果目录"""
        rel_path = self.config["paths"]["output"]["visualization"]
        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.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)