Files
config-man/src/config_man/cli/main.py
2026-04-15 14:12:01 +08:00

253 lines
8.6 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import click
import os
from ..core.view_command import ViewCommand
from ..core.suggest_update_command import SuggestUpdateCommand
from ..core.force_update_command import ForceUpdateCommand
from ..utils.mock_rclone import patch_rclone
from ..utils.config import config
@click.group()
@click.version_option(version="0.1.0")
def cli():
"""Config-Man CLI 工具
用于管理存储在对象存储中的加密静态资源配置文件。
支持多版本管理通过Cloudflare CDN进行分发。
"""
pass
@cli.command()
@click.option('--version', required=True, help='版本号,格式为 0.29, 0.30, 0.31 等')
@click.option('--json', 'json_format', is_flag=True, help='以JSON格式输出')
@click.option('--storage-path', help='对象存储路径前缀(覆盖配置文件)')
@click.option('--cdn-url', help='CDN基础URL覆盖配置文件')
@click.option('--mock', is_flag=True, help='使用模拟环境进行测试')
def view(version, json_format, storage_path, cdn_url, mock):
"""查看配置文件
对比显示对象存储和CDN中的配置内容支持 Android、iOS 和 RU 平台。
示例:
config-man view --version 0.30
config-man view --version 0.30 --json
config-man view --version 0.30 --mock
"""
try:
# 如果使用模拟环境启用rclone模拟
if mock:
patch_rclone()
click.echo("使用模拟环境进行测试...")
else:
# 检查rclone是否可用
if not _check_rclone():
click.echo("错误: rclone 未安装或不可用。请先安装 rclone。")
return 1
# 创建查看命令处理器
view_cmd = ViewCommand(storage_path, cdn_url)
# 执行查看命令
success = view_cmd.execute(version, json_format)
return 0 if success else 1
except Exception as e:
click.echo(f"错误: {str(e)}")
return 1
@cli.command()
@click.option('--platform', required=True, type=click.Choice(['android', 'ios', 'ru']),
help='平台类型')
@click.option('--target-version', required=True, help='目标版本号')
@click.option('--update-version', required=True, help='更新版本号')
@click.option('--storage-path', help='对象存储路径前缀(覆盖配置文件)')
@click.option('--cdn-url', help='CDN基础URL覆盖配置文件')
@click.option('--mock', is_flag=True, help='使用模拟环境进行测试')
def suggest_update(platform, target_version, update_version, storage_path, cdn_url, mock):
"""建议更新
在目标版本的配置文件中添加 Ver_New 属性,提示用户有新版本可用。
示例:
config-man suggest-update --platform android --target-version 0.29 --update-version 0.30
config-man suggest-update --platform ios --target-version 0.29 --update-version 0.30 --mock
config-man suggest-update --platform ru --target-version 0.29 --update-version 0.30 --mock
"""
try:
# 如果使用模拟环境启用rclone模拟
if mock:
patch_rclone()
click.echo("使用模拟环境进行测试...")
else:
# 检查rclone是否可用
if not _check_rclone():
click.echo("错误: rclone 未安装或不可用。请先安装 rclone。")
return 1
# 创建建议更新命令处理器
suggest_cmd = SuggestUpdateCommand(storage_path, cdn_url)
# 执行建议更新命令
success = suggest_cmd.execute(platform, target_version, update_version)
return 0 if success else 1
except Exception as e:
click.echo(f"错误: {str(e)}")
return 1
@cli.command()
@click.option('--platform', required=True, type=click.Choice(['android', 'ios', 'ru']),
help='平台类型')
@click.option('--target-version', required=True, help='目标版本号')
@click.option('--update-version', required=True, help='更新版本号')
@click.option('--storage-path', help='对象存储路径前缀(覆盖配置文件)')
@click.option('--cdn-url', help='CDN基础URL覆盖配置文件')
@click.option('--mock', is_flag=True, help='使用模拟环境进行测试')
def force_update(platform, target_version, update_version, storage_path, cdn_url, mock):
"""强制更新
直接更新目标版本的 Ver 属性,强制用户升级到新版本。
示例:
config-man force-update --platform ios --target-version 0.29 --update-version 0.30
config-man force-update --platform android --target-version 0.29 --update-version 0.30 --mock
config-man force-update --platform ru --target-version 0.29 --update-version 0.30 --mock
"""
try:
# 如果使用模拟环境启用rclone模拟
if mock:
patch_rclone()
click.echo("使用模拟环境进行测试...")
else:
# 检查rclone是否可用
if not _check_rclone():
click.echo("错误: rclone 未安装或不可用。请先安装 rclone。")
return 1
# 创建强制更新命令处理器
force_cmd = ForceUpdateCommand(storage_path, cdn_url)
# 执行强制更新命令
success = force_cmd.execute(platform, target_version, update_version)
return 0 if success else 1
except Exception as e:
click.echo(f"错误: {str(e)}")
return 1
@cli.command()
def show_config():
"""显示当前配置
显示当前加载的配置信息。
"""
import json
from rich.console import Console
from rich.panel import Panel
console = Console()
# 获取所有配置
all_config = {
"storage": config.get_storage_config(),
"cdn": config.get_cdn_config(),
"crypto": config.get_crypto_config(),
"display": config.get_display_config(),
"logging": config.get_logging_config()
}
# 格式化输出
config_json = json.dumps(all_config, indent=2, ensure_ascii=False)
panel = Panel(
config_json,
title="当前配置",
border_style="blue"
)
console.print(panel)
# 显示配置文件路径
console.print(f"\n配置文件路径: {config.config_file}")
@cli.command()
@click.option('--api-token', required=True, help='Cloudflare API Token')
@click.option('--zone-id', required=True, help='Cloudflare Zone ID')
def setup_cloudflare(api_token, zone_id):
"""设置Cloudflare CDN配置
设置Cloudflare API Token和Zone ID。
示例:
config-man setup-cloudflare --api-token your_api_token --zone-id your_zone_id
"""
try:
# 设置Cloudflare配置
config.set('cdn.cloudflare.api_token', api_token)
config.set('cdn.cloudflare.zone_id', zone_id)
config.save()
click.echo("✅ Cloudflare配置已更新")
click.echo(f" API Token: {api_token[:10]}...")
click.echo(f" Zone ID: {zone_id}")
click.echo("\n💡 提示: 配置已保存到配置文件,无需重复设置环境变量")
except Exception as e:
click.echo(f"❌ 设置Cloudflare配置失败: {e}")
return 1
@cli.command()
@click.option('--key', required=True, help='配置键,如 storage.path')
@click.option('--value', required=True, help='配置值')
def set_config(key, value):
"""设置配置值
设置指定的配置值并保存到配置文件。
示例:
config-man set-config --key storage.path --value custom_path
config-man set-config --key cdn.base_url --value https://cdn.example.com
config-man set-config --key cdn.cloudflare.api_token --value your_api_token
config-man set-config --key cdn.cloudflare.zone_id --value your_zone_id
"""
try:
# 尝试转换数值类型
if value.isdigit():
value = int(value)
elif value.lower() in ('true', 'false'):
value = value.lower() == 'true'
config.set(key, value)
config.save()
click.echo(f"✅ 配置已更新: {key} = {value}")
except Exception as e:
click.echo(f"❌ 设置配置失败: {e}")
return 1
def _check_rclone():
"""检查rclone是否可用"""
import subprocess
try:
result = subprocess.run(['rclone', 'version'],
capture_output=True, text=True, timeout=5)
return result.returncode == 0
except (subprocess.TimeoutExpired, FileNotFoundError):
return False
def main():
"""主函数"""
cli()
if __name__ == "__main__":
main()