APP配置MAP提取器.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
APP配置MAP提取器

功能描述：
从JSON配置文件中提取APP配置映射信息，并生成格式化的文本文件。
主要用于处理drpy-node项目中的APP配置数据，将JSON格式的配置转换为
便于使用的文本映射格式。

主要功能：
1. 读取JSON配置文件（App_PY.json）
2. 提取每个APP的api路径、名称和扩展配置
3. 格式化输出为文本映射文件（appMap.txt）
4. 支持嵌套JSON对象的序列化处理

输出格式：
{filename}@@{value}@@{key}[{name}]

使用示例：
python APP配置MAP提取器.py

作者: drpy-node
版本: 2.0.0
创建时间: 2024
最后修改: 2024
"""

import argparse
import json
import logging
import os
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional, Union


class AppConfigMapExtractor:
    """
    APP配置MAP提取器类
    
    负责从JSON配置文件中提取APP配置映射信息，并生成格式化的文本文件。
    """
    
    def __init__(self, separator: str = "@@") -> None:
        """
        初始化提取器
        
        Args:
            separator: 字段分隔符，默认为"@@"
        """
        self.separator = separator
        self.logger = self._setup_logger()
    
    def _setup_logger(self) -> logging.Logger:
        """
        设置日志记录器
        
        Returns:
            配置好的日志记录器实例
        """
        logger = logging.getLogger(__name__)
        logger.setLevel(logging.INFO)
        
        # 避免重复添加处理器
        if not logger.handlers:
            # 创建控制台处理器
            console_handler = logging.StreamHandler(sys.stdout)
            console_handler.setLevel(logging.INFO)
            
            # 设置日志格式
            formatter = logging.Formatter(
                '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                datefmt='%Y-%m-%d %H:%M:%S'
            )
            console_handler.setFormatter(formatter)
            
            logger.addHandler(console_handler)
        
        return logger
    
    def _validate_json_structure(self, data: Any) -> bool:
        """
        验证JSON数据结构的有效性
        
        Args:
            data: 待验证的JSON数据
            
        Returns:
            bool: 数据结构是否有效
        """
        if not isinstance(data, list):
            self.logger.error("JSON数据必须是一个数组")
            return False
        
        for i, item in enumerate(data):
            if not isinstance(item, dict):
                self.logger.error(f"第{i+1}个元素必须是对象类型")
                return False
            
            required_fields = ['api', 'name']
            for field in required_fields:
                if field not in item:
                    self.logger.warning(f"第{i+1}个元素缺少必需字段: {field}")
        
        return True
    
    def _extract_filename_from_api(self, api_path: str) -> str:
        """
        从API路径中提取文件名（不含扩展名）
        
        Args:
            api_path: API文件路径
            
        Returns:
            str: 提取的文件名
        """
        if not api_path:
            return ""
        
        try:
            # 使用pathlib处理路径，更加健壮
            path = Path(api_path)
            return path.stem  # 获取不含扩展名的文件名
        except Exception as e:
            self.logger.warning(f"解析API路径失败: {api_path}, 错误: {e}")
            return os.path.splitext(os.path.basename(api_path))[0]
    
    def _process_extension_value(self, value: Any) -> str:
        """
        处理扩展配置值，将复杂对象转换为JSON字符串
        
        Args:
            value: 扩展配置值
            
        Returns:
            str: 处理后的字符串值
        """
        if isinstance(value, dict):
            try:
                return json.dumps(value, ensure_ascii=False, separators=(',', ':'))
            except (TypeError, ValueError) as e:
                self.logger.warning(f"JSON序列化失败: {e}")
                return str(value)
        elif isinstance(value, (list, tuple)):
            try:
                return json.dumps(value, ensure_ascii=False, separators=(',', ':'))
            except (TypeError, ValueError) as e:
                self.logger.warning(f"JSON序列化失败: {e}")
                return str(value)
        else:
            return str(value)
    
    def _process_single_item(self, item: Dict[str, Any]) -> List[str]:
        """
        处理单个配置项，生成映射行
        
        Args:
            item: 单个APP配置项
            
        Returns:
            List[str]: 生成的映射行列表
        """
        results = []
        
        # 提取基本信息
        api_path = item.get('api', '')
        name = item.get('name', '')
        filename = self._extract_filename_from_api(api_path)
        
        if not filename:
            self.logger.warning(f"无法提取文件名，跳过项目: {item}")
            return results
        
        # 处理扩展配置
        exts = item.get('exts', {})
        if not isinstance(exts, dict):
            self.logger.warning(f"exts字段必须是对象类型，跳过项目: {item}")
            return results
        
        # 生成映射行
        for key, value in exts.items():
            try:
                processed_value = self._process_extension_value(value)
                line = f"{filename}{self.separator}{processed_value}{self.separator}{key}[{name}]"
                results.append(line)
            except Exception as e:
                self.logger.error(f"处理扩展配置失败: key={key}, value={value}, 错误: {e}")
                continue
        
        return results
    
    def _load_json_data(self, json_path: Path) -> Optional[List[Dict[str, Any]]]:
        """
        加载并验证JSON数据
        
        Args:
            json_path: JSON文件路径
            
        Returns:
            Optional[List[Dict[str, Any]]]: 加载的数据，失败时返回None
        """
        try:
            # 检查输入文件是否存在
            if not json_path.exists():
                self.logger.error(f"输入文件不存在: {json_path}")
                return None
            
            # 读取JSON文件
            self.logger.info(f"开始读取JSON文件: {json_path}")
            with open(json_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
            
            # 验证数据结构
            if not self._validate_json_structure(data):
                return None
                
            return data
            
        except json.JSONDecodeError as e:
            self.logger.error(f"JSON文件格式错误: {e}")
            return None
        except FileNotFoundError as e:
            self.logger.error(f"文件未找到: {e}")
            return None
        except Exception as e:
            self.logger.error(f"读取JSON文件时发生错误: {e}")
            return None
    
    def _process_all_items(self, data: List[Dict[str, Any]]) -> List[str]:
        """
        处理所有配置项
        
        Args:
            data: 配置数据列表
            
        Returns:
            List[str]: 处理后的映射行列表
        """
        self.logger.info("开始处理配置数据...")
        all_results = []
        
        for i, item in enumerate(data):
            try:
                item_results = self._process_single_item(item)
                all_results.extend(item_results)
            except Exception as e:
                self.logger.error(f"处理第{i+1}个配置项失败: {e}")
                continue
        
        return all_results
    
    def _write_output_file(self, output_path: Path, results: List[str]) -> bool:
        """
        写入输出文件
        
        Args:
            output_path: 输出文件路径
            results: 要写入的结果列表
            
        Returns:
            bool: 写入是否成功
        """
        try:
            # 确保输出目录存在
            output_path.parent.mkdir(parents=True, exist_ok=True)
            
            # 写入输出文件
            self.logger.info(f"开始写入输出文件: {output_path}")
            with open(output_path, 'w', encoding='utf-8') as f:
                f.write('\n'.join(results))
            
            self.logger.info(f"处理完成！共生成 {len(results)} 行映射数据")
            self.logger.info(f"输出文件: {output_path.absolute()}")
            
            return True
            
        except PermissionError as e:
            self.logger.error(f"文件权限错误: {e}")
            return False
        except Exception as e:
            self.logger.error(f"写入文件时发生错误: {e}")
            return False

    def extract_config_map(self, json_file_path: Union[str, Path], 
                          output_txt_path: Union[str, Path]) -> bool:
        """
        从JSON文件提取配置映射并保存到文本文件
        
        Args:
            json_file_path: 输入的JSON文件路径
            output_txt_path: 输出的文本文件路径
            
        Returns:
            bool: 处理是否成功
        """
        json_path = Path(json_file_path)
        output_path = Path(output_txt_path)
        
        # 加载JSON数据
        data = self._load_json_data(json_path)
        if data is None:
            return False
        
        # 处理所有配置项
        all_results = self._process_all_items(data)
        if not all_results:
            self.logger.warning("没有生成任何映射数据")
            return False
        
        # 写入输出文件
        return self._write_output_file(output_path, all_results)


def parse_arguments() -> argparse.Namespace:
    """
    解析命令行参数
    
    Returns:
        argparse.Namespace: 解析后的命令行参数
    """
    parser = argparse.ArgumentParser(
        description="APP配置MAP提取器 - 从JSON配置文件中提取APP配置映射信息",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
使用示例:
  %(prog)s                                    # 使用默认文件名
  %(prog)s -i config.json -o output.txt      # 指定输入输出文件
  %(prog)s -s "||" -v                        # 使用自定义分隔符和详细输出
  %(prog)s --input-file App_PY.json --quiet  # 静默模式运行
        """
    )
    
    parser.add_argument(
        '-i', '--input-file',
        type=str,
        default='App_PY.json',
        help='输入的JSON配置文件路径 (默认: App_PY.json)'
    )
    
    parser.add_argument(
        '-o', '--output-file',
        type=str,
        default='appMap.txt',
        help='输出的文本映射文件路径 (默认: appMap.txt)'
    )
    
    parser.add_argument(
        '-s', '--separator',
        type=str,
        default='@@',
        help='字段分隔符 (默认: @@)'
    )
    
    parser.add_argument(
        '-v', '--verbose',
        action='store_true',
        help='启用详细输出模式'
    )
    
    parser.add_argument(
        '-q', '--quiet',
        action='store_true',
        help='静默模式，只输出错误信息'
    )
    
    parser.add_argument(
        '--version',
        action='version',
        version='%(prog)s 2.0.0'
    )
    
    return parser.parse_args()


def setup_logging_level(verbose: bool, quiet: bool) -> int:
    """
    根据命令行参数设置日志级别
    
    Args:
        verbose: 是否启用详细输出
        quiet: 是否启用静默模式
        
    Returns:
        int: 日志级别
    """
    if quiet:
        return logging.ERROR
    elif verbose:
        return logging.DEBUG
    else:
        return logging.INFO


def validate_arguments(args: argparse.Namespace) -> None:
    """
    验证命令行参数的有效性
    
    Args:
        args: 解析后的命令行参数
        
    Raises:
        SystemExit: 参数验证失败时退出程序
    """
    if args.verbose and args.quiet:
        print("错误: --verbose 和 --quiet 参数不能同时使用", file=sys.stderr)
        sys.exit(1)


def setup_extractor(args: argparse.Namespace) -> AppConfigMapExtractor:
    """
    创建并配置提取器实例
    
    Args:
        args: 解析后的命令行参数
        
    Returns:
        AppConfigMapExtractor: 配置好的提取器实例
    """
    # 设置日志级别
    log_level = setup_logging_level(args.verbose, args.quiet)
    
    # 创建提取器实例
    extractor = AppConfigMapExtractor(separator=args.separator)
    
    # 更新日志级别
    extractor.logger.setLevel(log_level)
    for handler in extractor.logger.handlers:
        handler.setLevel(log_level)
    
    return extractor


def print_startup_info(extractor: AppConfigMapExtractor, args: argparse.Namespace) -> None:
    """
    输出程序启动信息
    
    Args:
        extractor: 提取器实例
        args: 解析后的命令行参数
    """
    if not args.quiet:
        extractor.logger.info("=" * 50)
        extractor.logger.info("APP配置MAP提取器 v2.0.0")
        extractor.logger.info("=" * 50)
        extractor.logger.info(f"输入文件: {args.input_file}")
        extractor.logger.info(f"输出文件: {args.output_file}")
        extractor.logger.info(f"分隔符: {args.separator}")
        extractor.logger.info("=" * 50)


def print_result_info(extractor: AppConfigMapExtractor, args: argparse.Namespace, success: bool) -> None:
    """
    输出处理结果信息
    
    Args:
        extractor: 提取器实例
        args: 解析后的命令行参数
        success: 处理是否成功
    """
    if not args.quiet:
        if success:
            extractor.logger.info("✅ 处理成功完成！")
        else:
            extractor.logger.error("❌ 处理失败！")


def main() -> None:
    """
    主函数 - 程序入口点
    """
    try:
        # 解析和验证命令行参数
        args = parse_arguments()
        validate_arguments(args)
        
        # 创建并配置提取器
        extractor = setup_extractor(args)
        
        # 输出启动信息
        print_startup_info(extractor, args)
        
        # 执行提取操作
        success = extractor.extract_config_map(args.input_file, args.output_file)
        
        # 输出结果信息
        print_result_info(extractor, args, success)
        
        # 根据结果设置退出码
        sys.exit(0 if success else 1)
        
    except KeyboardInterrupt:
        print("\n用户中断操作", file=sys.stderr)
        sys.exit(130)
    except Exception as e:
        print(f"程序执行出错: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()