147 lines
4.0 KiB
Python
147 lines
4.0 KiB
Python
"""
|
||
订单合并命令行工具
|
||
--------------
|
||
提供订单合并相关的命令行接口。
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import argparse
|
||
from typing import List, Optional
|
||
|
||
from ..config.settings import ConfigManager
|
||
from ..core.utils.log_utils import get_logger, close_logger
|
||
from ..services.order_service import OrderService
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
def create_parser() -> argparse.ArgumentParser:
|
||
"""
|
||
创建命令行参数解析器
|
||
|
||
Returns:
|
||
参数解析器
|
||
"""
|
||
parser = argparse.ArgumentParser(description='订单合并工具')
|
||
|
||
# 通用选项
|
||
parser.add_argument('--config', type=str, help='配置文件路径')
|
||
|
||
# 子命令
|
||
subparsers = parser.add_subparsers(dest='command', help='子命令')
|
||
|
||
# 合并命令
|
||
merge_parser = subparsers.add_parser('merge', help='合并采购单')
|
||
merge_parser.add_argument('--input', type=str, help='输入采购单文件路径列表，以逗号分隔，如果不指定则合并所有采购单')
|
||
|
||
# 列出采购单命令
|
||
list_parser = subparsers.add_parser('list', help='列出采购单文件')
|
||
|
||
return parser
|
||
|
||
def merge_orders(order_service: OrderService, input_files: Optional[str] = None) -> bool:
|
||
"""
|
||
合并采购单
|
||
|
||
Args:
|
||
order_service: 订单服务
|
||
input_files: 输入文件路径列表，以逗号分隔，如果为None则合并所有采购单
|
||
|
||
Returns:
|
||
合并是否成功
|
||
"""
|
||
if input_files:
|
||
# 分割输入文件列表
|
||
file_paths = [path.strip() for path in input_files.split(',')]
|
||
|
||
# 检查文件是否存在
|
||
for path in file_paths:
|
||
if not os.path.exists(path):
|
||
logger.error(f"输入文件不存在: {path}")
|
||
return False
|
||
|
||
result = order_service.merge_orders(file_paths)
|
||
else:
|
||
# 获取所有采购单文件
|
||
file_paths = order_service.get_purchase_orders()
|
||
if not file_paths:
|
||
logger.warning("未找到采购单文件")
|
||
return False
|
||
|
||
logger.info(f"合并 {len(file_paths)} 个采购单文件")
|
||
result = order_service.merge_orders()
|
||
|
||
if result:
|
||
logger.info(f"合并成功，输出文件: {result}")
|
||
return True
|
||
else:
|
||
logger.error("合并失败")
|
||
return False
|
||
|
||
def list_purchase_orders(order_service: OrderService) -> bool:
|
||
"""
|
||
列出采购单文件
|
||
|
||
Args:
|
||
order_service: 订单服务
|
||
|
||
Returns:
|
||
是否有采购单文件
|
||
"""
|
||
files = order_service.get_purchase_orders()
|
||
|
||
if not files:
|
||
logger.info("未找到采购单文件")
|
||
return False
|
||
|
||
logger.info(f"采购单文件 ({len(files)}):")
|
||
for file in files:
|
||
logger.info(f" {file}")
|
||
|
||
return True
|
||
|
||
def main(args: Optional[List[str]] = None) -> int:
|
||
"""
|
||
订单合并命令行主函数
|
||
|
||
Args:
|
||
args: 命令行参数，如果为None则使用sys.argv
|
||
|
||
Returns:
|
||
退出状态码
|
||
"""
|
||
parser = create_parser()
|
||
parsed_args = parser.parse_args(args)
|
||
|
||
if parsed_args.command is None:
|
||
parser.print_help()
|
||
return 1
|
||
|
||
try:
|
||
# 创建配置管理器
|
||
config = ConfigManager(parsed_args.config) if parsed_args.config else ConfigManager()
|
||
|
||
# 创建订单服务
|
||
order_service = OrderService(config)
|
||
|
||
# 根据命令执行不同功能
|
||
if parsed_args.command == 'merge':
|
||
success = merge_orders(order_service, parsed_args.input)
|
||
elif parsed_args.command == 'list':
|
||
success = list_purchase_orders(order_service)
|
||
else:
|
||
parser.print_help()
|
||
return 1
|
||
|
||
return 0 if success else 1
|
||
|
||
except Exception as e:
|
||
logger.error(f"执行过程中发生错误: {e}")
|
||
return 1
|
||
|
||
finally:
|
||
# 关闭日志
|
||
close_logger(__name__)
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(main()) |