Mustapha/config-man
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: restructure project to modern Python layout, separate src, tests, docs, examples

Browse Source
This commit is contained in:
tech
2025-08-07 16:57:41 +08:00
commit 4cb6c4eab7
27 changed files with 2580 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/config_man/__init__.py Normal file
View File

@@ -0,0 +1,10 @@
"""
Config-Man CLI 工具
用于管理存储在对象存储中的加密静态资源配置文件。
支持多版本管理，通过Cloudflare CDN进行分发。
"""
__version__ = "0.1.0"
__author__ = "Config-Man Team"
__description__ = "CLI tool for managing encrypted configuration files"

9
src/config_man/cli/__init__.py Normal file
View File

@@ -0,0 +1,9 @@
"""
CLI 模块
包含命令行界面相关的功能。
"""
from .main import cli, main
__all__ = ["cli", "main"]

109
src/config_man/cli/main.py Normal file
View File

@@ -0,0 +1,109 @@
import click
import os
from ..core.view_command import ViewCommand
from ..utils.mock_rclone import patch_rclone
@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', default='ab', help='对象存储路径前缀')
@click.option('--cdn-url', default='', help='CDN基础URL')
@click.option('--mock', is_flag=True, help='使用模拟环境进行测试')
def view(version, json_format, storage_path, cdn_url, mock):
"""查看配置文件
对比显示对象存储和CDN中的配置内容，支持Android和iOS平台。
示例:
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']),
help='平台类型')
@click.option('--target-version', required=True, help='目标版本号')
@click.option('--update-version', required=True, help='更新版本号')
def suggest_update(platform, target_version, update_version):
"""建议更新
在目标版本的配置文件中添加 Ver_New 属性，提示用户有新版本可用。
示例:
config-man suggest-update --platform android --target-version 0.29 --update-version 0.30
"""
click.echo("建议更新功能尚未实现")
return 1
@cli.command()
@click.option('--platform', required=True, type=click.Choice(['android', 'ios']),
help='平台类型')
@click.option('--target-version', required=True, help='目标版本号')
@click.option('--update-version', required=True, help='更新版本号')
def force_update(platform, target_version, update_version):
"""强制更新
直接更新目标版本的 Ver 属性，强制用户升级到新版本。
示例:
config-man force-update --platform ios --target-version 0.29 --update-version 0.30
"""
click.echo("强制更新功能尚未实现")
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()

11
src/config_man/core/__init__.py Normal file
View File

@@ -0,0 +1,11 @@
"""
核心功能模块
包含配置管理、加密解密、显示管理等核心功能。
"""
from .config_manager import ConfigManager
from .display_manager import DisplayManager
from .view_command import ViewCommand
__all__ = ["ConfigManager", "DisplayManager", "ViewCommand"]

119
src/config_man/core/config_manager.py Normal file
View File

@@ -0,0 +1,119 @@
import json
import os
import subprocess
import requests
from typing import Dict, Optional, Tuple
from ..utils.crypto import encrypt, decrypt
class ConfigManager:
"""配置文件管理器"""
def __init__(self, storage_path: str = "ab", cdn_base_url: str = ""):
self.storage_path = storage_path
self.cdn_base_url = cdn_base_url
def version_to_path(self, version: str) -> str:
"""将版本号转换为存储路径格式"""
# 0.29 -> 0_29
return version.replace('.', '_')
def get_config_path(self, version: str, platform: str) -> str:
"""获取配置文件在对象存储中的路径"""
version_path = self.version_to_path(version)
filename = f"{platform}config.json"
return f"{self.storage_path}/{version_path}/{filename}"
def get_cdn_url(self, version: str, platform: str) -> str:
"""获取配置文件在CDN中的URL"""
if not self.cdn_base_url:
return ""
version_path = self.version_to_path(version)
filename = f"{platform}config.json"
return f"{self.cdn_base_url}/{self.storage_path}/{version_path}/{filename}"
def download_from_storage(self, version: str, platform: str) -> Optional[str]:
"""从对象存储下载并解密配置文件"""
try:
config_path = self.get_config_path(version, platform)
# 使用rclone下载文件
result = subprocess.run(
["rclone", "cat", config_path],
capture_output=True,
text=True,
timeout=30
)
if result.returncode != 0:
print(f"警告: 无法从对象存储下载 {config_path}: {result.stderr}")
return None
# 解密文件内容
encrypted_content = result.stdout.strip()
if not encrypted_content:
print(f"警告: 文件 {config_path} 为空")
return None
decrypted_content = decrypt(encrypted_content)
return decrypted_content
except subprocess.TimeoutExpired:
print(f"错误: 下载 {config_path} 超时")
return None
except Exception as e:
print(f"错误: 下载 {config_path} 失败: {str(e)}")
return None
def get_from_cdn(self, version: str, platform: str) -> Optional[str]:
"""从CDN获取配置文件内容"""
try:
cdn_url = self.get_cdn_url(version, platform)
if not cdn_url:
print(f"警告: CDN URL未配置，跳过CDN内容获取")
return None
response = requests.get(cdn_url, timeout=10)
response.raise_for_status()
# CDN中的内容应该是解密后的JSON
return response.text
except requests.RequestException as e:
print(f"警告: 无法从CDN获取 {cdn_url}: {str(e)}")
return None
except Exception as e:
print(f"错误: 获取CDN内容失败: {str(e)}")
return None
def parse_config(self, content: str) -> Optional[Dict]:
"""解析配置文件内容"""
try:
return json.loads(content)
except json.JSONDecodeError as e:
print(f"错误: 配置文件格式错误: {str(e)}")
return None
def compare_configs(self, storage_config: Dict, cdn_config: Dict) -> Dict:
"""比较存储配置和CDN配置的差异"""
all_keys = set(storage_config.keys()) | set(cdn_config.keys())
differences = {}
for key in sorted(all_keys):
storage_value = storage_config.get(key, "N/A")
cdn_value = cdn_config.get(key, "N/A")
if storage_value != cdn_value:
differences[key] = {
"storage": storage_value,
"cdn": cdn_value,
"different": True
}
else:
differences[key] = {
"storage": storage_value,
"cdn": cdn_value,
"different": False
}
return differences

109
src/config_man/core/display_manager.py Normal file
View File

@@ -0,0 +1,109 @@
from typing import Dict, List
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.text import Text
import json
class DisplayManager:
"""显示管理器"""
def __init__(self):
self.console = Console()
def display_config_comparison(self, version: str, platform: str,
storage_config: Dict, cdn_config: Dict,
differences: Dict):
"""显示配置对比结果"""
# 创建标题
title = f"版本: {version} | 平台: {platform.title()}"
self.console.print(f"\n[bold blue]{title}[/bold blue]")
# 创建表格
table = Table(show_header=True, header_style="bold magenta")
table.add_column("配置项", style="cyan", width=20)
table.add_column("对象存储内容", style="green", width=30)
table.add_column("CDN内容", style="yellow", width=30)
table.add_column("状态", style="red", width=10)
# 添加数据行
for key, data in differences.items():
storage_value = str(data["storage"])
cdn_value = str(data["cdn"])
status = "❌ 不同" if data["different"] else "✅ 一致"
# 截断过长的值
if len(storage_value) > 25:
storage_value = storage_value[:22] + "..."
if len(cdn_value) > 25:
cdn_value = cdn_value[:22] + "..."
table.add_row(key, storage_value, cdn_value, status)
self.console.print(table)
def display_json_format(self, version: str, platform: str,
storage_config: Dict, cdn_config: Dict):
"""以JSON格式显示配置"""
title = f"版本: {version} | 平台: {platform.title()}"
self.console.print(f"\n[bold blue]{title}[/bold blue]")
# 创建对比JSON
comparison = {
"version": version,
"platform": platform,
"storage_config": storage_config,
"cdn_config": cdn_config
}
# 格式化JSON输出
json_str = json.dumps(comparison, indent=2, ensure_ascii=False)
# 使用面板显示
panel = Panel(
json_str,
title="JSON格式",
border_style="blue"
)
self.console.print(panel)
def display_summary(self, version: str, results: Dict):
"""显示总结信息"""
self.console.print(f"\n[bold green]配置查看总结[/bold green]")
self.console.print(f"版本: {version}")
for platform in ["android", "ios"]:
if platform in results:
result = results[platform]
if result["success"]:
storage_keys = len(result["storage_config"])
cdn_keys = len(result["cdn_config"])
differences = sum(1 for data in result["differences"].values()
if data["different"])
status = "✅ 正常" if differences == 0 else f"⚠️ {differences} 项不同"
self.console.print(
f" {platform.title()}: {status} "
f"(存储: {storage_keys}项, CDN: {cdn_keys}项)"
)
else:
self.console.print(f" {platform.title()}: ❌ 失败")
else:
self.console.print(f" {platform.title()}: ⚠️ 未检查")
def display_error(self, message: str):
"""显示错误信息"""
self.console.print(f"[bold red]错误: {message}[/bold red]")
def display_warning(self, message: str):
"""显示警告信息"""
self.console.print(f"[bold yellow]警告: {message}[/bold yellow]")
def display_info(self, message: str):
"""显示信息"""
self.console.print(f"[bold blue]信息: {message}[/bold blue]")

142
src/config_man/core/view_command.py Normal file
View File

@@ -0,0 +1,142 @@
from typing import Dict, Optional
from .config_manager import ConfigManager
from .display_manager import DisplayManager
class ViewCommand:
"""查看命令处理器"""
def __init__(self, storage_path: str = "ab", cdn_base_url: str = ""):
self.config_manager = ConfigManager(storage_path, cdn_base_url)
self.display_manager = DisplayManager()
def execute(self, version: str, json_format: bool = False) -> bool:
"""执行查看命令"""
# 验证版本号格式
if not self._validate_version(version):
self.display_manager.display_error(f"无效的版本号格式: {version}")
return False
self.display_manager.display_info(f"正在查看版本 {version} 的配置文件...")
results = {}
success = True
# 检查Android和iOS平台
for platform in ["android", "ios"]:
result = self._check_platform_config(version, platform)
results[platform] = result
if not result["success"]:
success = False
# 显示结果
if json_format:
self._display_json_results(version, results)
else:
self._display_table_results(version, results)
# 显示总结
self.display_manager.display_summary(version, results)
return success
def _validate_version(self, version: str) -> bool:
"""验证版本号格式"""
try:
# 检查版本号格式 (如 0.29, 0.30, 1.00)
parts = version.split('.')
if len(parts) != 2:
return False
major, minor = parts
if not major.isdigit() or not minor.isdigit():
return False
return True
except:
return False
def _check_platform_config(self, version: str, platform: str) -> Dict:
"""检查指定平台的配置"""
result = {
"success": False,
"storage_config": {},
"cdn_config": {},
"differences": {},
"error": None
}
try:
# 从对象存储获取配置
storage_content = self.config_manager.download_from_storage(version, platform)
if storage_content is None:
result["error"] = f"无法从对象存储获取 {platform} 配置"
return result
storage_config = self.config_manager.parse_config(storage_content)
if storage_config is None:
result["error"] = f"解析对象存储中的 {platform} 配置失败"
return result
result["storage_config"] = storage_config
# 从CDN获取配置
cdn_content = self.config_manager.get_from_cdn(version, platform)
if cdn_content is None:
# CDN获取失败，使用空配置
result["cdn_config"] = {}
result["differences"] = self.config_manager.compare_configs(
storage_config, {}
)
else:
cdn_config = self.config_manager.parse_config(cdn_content)
if cdn_config is None:
result["error"] = f"解析CDN中的 {platform} 配置失败"
return result
result["cdn_config"] = cdn_config
result["differences"] = self.config_manager.compare_configs(
storage_config, cdn_config
)
result["success"] = True
except Exception as e:
result["error"] = f"检查 {platform} 配置时发生错误: {str(e)}"
return result
def _display_table_results(self, version: str, results: Dict):
"""以表格格式显示结果"""
for platform in ["android", "ios"]:
if platform in results:
result = results[platform]
if result["success"]:
self.display_manager.display_config_comparison(
version, platform,
result["storage_config"],
result["cdn_config"],
result["differences"]
)
else:
self.display_manager.display_error(
f"{platform.title()} 配置检查失败: {result['error']}"
)
def _display_json_results(self, version: str, results: Dict):
"""以JSON格式显示结果"""
for platform in ["android", "ios"]:
if platform in results:
result = results[platform]
if result["success"]:
self.display_manager.display_json_format(
version, platform,
result["storage_config"],
result["cdn_config"]
)
else:
self.display_manager.display_error(
f"{platform.title()} 配置检查失败: {result['error']}"
)

10
src/config_man/utils/__init__.py Normal file
View File

@@ -0,0 +1,10 @@
"""
工具模块
包含加密解密、模拟环境等工具功能。
"""
from .crypto import encrypt, decrypt
from .mock_rclone import MockRclone, patch_rclone
__all__ = ["encrypt", "decrypt", "MockRclone", "patch_rclone"]

23
src/config_man/utils/crypto.py Normal file
View File

@@ -0,0 +1,23 @@
import json
import os
from Crypto.Cipher import DES
from Crypto.Util.Padding import pad,unpad
import base64
def encrypt(text):
key = 'tbambooz'.encode('ascii')
des = DES.new(key, DES.MODE_CBC, IV=key)
padded_text = pad(text.encode('utf-8'), DES.block_size)
encrypted_text = des.encrypt(padded_text)
return base64.b64encode(encrypted_text).decode('utf-8')
def decrypt(encrypted_text):
key = 'tbambooz'
key = key.encode('ascii')
des = DES.new(key, DES.MODE_CBC, IV=key)
decoded_text = base64.b64decode(encrypted_text)
decrypted_text = des.decrypt(decoded_text)
unpadded_text = unpad(decrypted_text, DES.block_size)
return unpadded_text.decode('utf-8')

153
src/config_man/utils/mock_rclone.py Normal file
View File

@@ -0,0 +1,153 @@
import os
import json
import subprocess
from .crypto import encrypt, decrypt
class MockRclone:
"""模拟rclone环境，用于测试"""
def __init__(self, test_data_path: str = "test_data"):
self.test_data_path = test_data_path
self._setup_mock_environment()
def _setup_mock_environment(self):
"""设置模拟环境"""
# 确保测试数据目录存在
os.makedirs(self.test_data_path, exist_ok=True)
# 创建测试配置文件
self._create_test_configs()
def _create_test_configs(self):
"""创建测试配置文件"""
test_configs = {
"0.29": {
"android": {
"EventApiURL": "https://n3backend.azurewebsites.net/",
"Ver": "0.29.1abc123",
"PlayFabTitle": "B066F",
"Google_Play_URL": "https://play.google.com/store/apps/details?id=com.arkgame.ft",
"APP_Store_URL": "https://apps.apple.com/us/app/id6505145935",
"HeartBeat": "60",
"RTMPid": "80000586",
"RTMServerEndpoint": "rtm-intl-frontgate.ilivedata.com:13321",
"RTMHmacSecret": "57047697437f4f2c97a835e8d9a53358",
"FuncUrl": "https://leaderboardcreate.azurewebsites.net",
"FuncKey": "R5kU45fNBRd52Eqp3tEqfpZqrbFw53uSSEo7wraUSqIfAzFuRmLm_w=="
},
"ios": {
"EventApiURL": "https://n3backend.azurewebsites.net/",
"Ver": "0.29.1def456",
"PlayFabTitle": "B066F",
"Google_Play_URL": "https://play.google.com/store/apps/details?id=com.arkgame.ft",
"APP_Store_URL": "https://apps.apple.com/us/app/id6505145935",
"HeartBeat": "60",
"RTMPid": "80000586",
"RTMServerEndpoint": "rtm-intl-frontgate.ilivedata.com:13321",
"RTMHmacSecret": "57047697437f4f2c97a835e8d9a53358",
"FuncUrl": "https://leaderboardcreate.azurewebsites.net",
"FuncKey": "R5kU45fNBRd52Eqp3tEqfpZqrbFw53uSSEo7wraUSqIfAzFuRmLm_w=="
}
},
"0.30": {
"android": {
"EventApiURL": "https://n3backend.azurewebsites.net/",
"Ver": "0.30.2ghi789",
"PlayFabTitle": "B066F",
"Google_Play_URL": "https://play.google.com/store/apps/details?id=com.arkgame.ft",
"APP_Store_URL": "https://apps.apple.com/us/app/id6505145935",
"HeartBeat": "60",
"RTMPid": "80000586",
"RTMServerEndpoint": "rtm-intl-frontgate.ilivedata.com:13321",
"RTMHmacSecret": "57047697437f4f2c97a835e8d9a53358",
"FuncUrl": "https://leaderboardcreate.azurewebsites.net",
"FuncKey": "R5kU45fNBRd52Eqp3tEqfpZqrbFw53uSSEo7wraUSqIfAzFuRmLm_w==",
"NewFeature": "enabled"
},
"ios": {
"EventApiURL": "https://n3backend.azurewebsites.net/",
"Ver": "0.30.2jkl012",
"PlayFabTitle": "B066F",
"Google_Play_URL": "https://play.google.com/store/apps/details?id=com.arkgame.ft",
"APP_Store_URL": "https://apps.apple.com/us/app/id6505145935",
"HeartBeat": "60",
"RTMPid": "80000586",
"RTMServerEndpoint": "rtm-intl-frontgate.ilivedata.com:13321",
"RTMHmacSecret": "57047697437f4f2c97a835e8d9a53358",
"FuncUrl": "https://leaderboardcreate.azurewebsites.net",
"FuncKey": "R5kU45fNBRd52Eqp3tEqfpZqrbFw53uSSEo7wraUSqIfAzFuRmLm_w==",
"NewFeature": "enabled"
}
}
}
# 创建测试目录结构
for version, platforms in test_configs.items():
version_path = f"{self.test_data_path}/ab/{version.replace('.', '_')}"
os.makedirs(version_path, exist_ok=True)
for platform, config in platforms.items():
filename = f"{platform}config.json"
filepath = os.path.join(version_path, filename)
# 加密配置并保存
config_json = json.dumps(config, indent=2, ensure_ascii=False)
encrypted_config = encrypt(config_json)
with open(filepath, 'w', encoding='utf-8') as f:
f.write(encrypted_config)
print(f"创建测试文件: {filepath}")
def mock_rclone_cat(self, path: str) -> str:
"""模拟rclone cat命令"""
# 将路径转换为本地文件路径
local_path = os.path.join(self.test_data_path, path)
if not os.path.exists(local_path):
raise FileNotFoundError(f"文件不存在: {path}")
with open(local_path, 'r', encoding='utf-8') as f:
return f.read().strip()
def patch_rclone():
"""修补rclone调用，使用模拟环境"""
import subprocess
original_run = subprocess.run
def mock_run(cmd, *args, **kwargs):
if len(cmd) >= 2 and cmd[0] == 'rclone' and cmd[1] == 'cat':
# 模拟rclone cat命令
mock_rclone = MockRclone()
try:
content = mock_rclone.mock_rclone_cat(cmd[2])
# 返回模拟的subprocess结果
from types import SimpleNamespace
result = SimpleNamespace()
result.returncode = 0
result.stdout = content
result.stderr = ""
return result
except FileNotFoundError as e:
# 返回错误结果
from types import SimpleNamespace
result = SimpleNamespace()
result.returncode = 1
result.stdout = ""
result.stderr = str(e)
return result
else:
# 对于其他命令，使用原始实现
return original_run(cmd, *args, **kwargs)
subprocess.run = mock_run
print("已启用rclone模拟环境")
if __name__ == "__main__":
# 创建测试环境
mock = MockRclone()
print("模拟rclone环境设置完成！")