4v5u09Z5a4Yuc/iOSBuildServer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b2982ae8f7affd0207669efb2f54fced90f027f6
iOSBuildServer/backend/routers/config.py
T
shenleiandClaude Opus 4.8 5014376b81 feat: 支持皮肤包就地更新与新建即传 
- 新增皮肤包更新接口，覆盖内容保持文件名不变，避免 APP 配置的主题引用失效
- 上传新皮肤后自动为空主题目录的证书选中该皮肤
- 新建 APP 保存后不关弹窗并回填 upload_key，可直接上传皮肤

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-29 15:14:08 +09:00

739 lines
25 KiB
Python
Raw Blame History

"""配置管理 API"""
import asyncio
import json
import os
import re
import shutil
import zipfile
from datetime import datetime
from pathlib import Path
from fastapi import APIRouter, Depends, HTTPException, Request, UploadFile, File
from sqlalchemy.orm import Session
from ..database import get_db
from ..models import BuildConfig
from ..schemas import BuildConfigUpdate
from ..config import (
BUILD_BASE_DIR,
BOOTSTRAP_CONFIG_PATH,
AUTOMATION_DIR,
CONFIG_JSON_PATH,
DEFAULT_MAX_CONCURRENT_BUILDS,
DEFAULT_BUILD_DIR_RETENTION_HOURS,
SKINS_DIR,
)
from ..deps import get_current_user, require_admin
def _require_config_permission(request: Request, user: dict = Depends(get_current_user)) -> dict:
"""普通用户可管理 Apps 与版本号；其余配置及凭据只允许管理员访问。"""
path = request.url.path.rstrip("/")
if (path == "/api/config/apps" or path.startswith("/api/config/apps/")
or path == "/api/config/versions"):
return user
return require_admin(user)
router = APIRouter(prefix="/api/config", tags=["config"], dependencies=[Depends(_require_config_permission)])
_config_lock = asyncio.Lock()
# 默认上传配置
DEFAULT_UPLOAD = {
"mode": "oss",
"oss": {
"access_key_id": "",
"access_key_secret": "",
"endpoint": "oss-cn-beijing.aliyuncs.com",
"bucket_name": "",
"base_url": "",
},
"webdav": {
"server_url": "",
"username": "",
"password": "",
"base_path": "/ios-builds",
"public_url": "",
},
"dingtalk": {
"enabled": False,
"webhook_url": "",
"secret": "",
},
}
# 默认服务器环境配置
DEFAULT_SERVERS = {
"测试环境": {
"api": "https://api3-dev.readoor.cn",
"assDom": "applinks:dev-data1.readoor.cn",
"universalLink": "https://dev-data1.readoor.cn",
"app_id_prefix": 1,
},
"正式环境": {
"api": "https://api3.readoor.cn",
"assDom": "applinks:data1.readoor.cn",
"universalLink": "https://data1.readoor.cn",
"app_id_prefix": 2,
},
"华师大环境": {
"api": "https://api3.ecnupress.com.cn",
"assDom": "applinks:data1.ecnupress.com.cn",
"universalLink": "https://data1.ecnupress.com.cn",
"app_id_prefix": 3,
},
"外教环境": {
"api": "https://weread-api3.sflep.com/api3",
"assDom": "applinks:wereadossda.sflep.com",
"universalLink": "https://wereadossda.sflep.com",
"app_id_prefix": 4,
}
}
DEFAULT_VERSIONS = {
"app_ver": "2.180.0",
"build_ver": "2.180.0.0",
}
# 为已有环境迁移的固定前缀；后续环境从配置中的 next_app_id_prefix 自动分配。
LEGACY_SERVER_PREFIXES = {
"测试环境": 1,
"正式环境": 2,
"华师大环境": 3,
"外教环境": 4,
}
def _ensure_server_id_prefixes(config: dict) -> bool:
"""为旧配置补齐环境 ID 前缀和下一个可分配前缀。"""
changed = False
servers = config.get("servers", {})
used_prefixes = set()
for name, server in servers.items():
prefix = server.get("app_id_prefix")
if prefix is None and name in LEGACY_SERVER_PREFIXES:
prefix = LEGACY_SERVER_PREFIXES[name]
server["app_id_prefix"] = prefix
changed = True
if isinstance(prefix, int) and prefix > 0:
used_prefixes.add(prefix)
next_prefix = max(used_prefixes, default=0) + 1
for server in servers.values():
prefix = server.get("app_id_prefix")
if not isinstance(prefix, int) or prefix <= 0:
server["app_id_prefix"] = next_prefix
used_prefixes.add(next_prefix)
next_prefix += 1
changed = True
if not isinstance(config.get("next_app_id_prefix"), int) or config["next_app_id_prefix"] < next_prefix:
config["next_app_id_prefix"] = next_prefix
changed = True
return changed
def _ensure_special_app_prefix_overrides(config: dict) -> bool:
"""将英汉大词典的历史 1xx 规则迁移为显式的 App 前缀覆盖。"""
changed = False
for app in config.get("apps", {}).values():
if "英汉大词典" in app.get("name", "") and not app.get("app_id_prefix_override"):
app["app_id_prefix_override"] = 1
changed = True
return changed
def _apply_special_app_prefix_override(app: dict) -> None:
"""英汉大词典沿用历史 1xx 编号段。"""
if "英汉大词典" in app.get("name", "") and not app.get("app_id_prefix_override"):
app["app_id_prefix_override"] = 1
def _next_app_id(apps: dict, servers: dict, app: dict) -> str:
"""根据环境前缀或 App 特殊覆盖生成下一个配置 ID。"""
prefix = app.get("app_id_prefix_override")
if prefix in (None, ""):
server = servers.get(app.get("server", ""), {})
prefix = server.get("app_id_prefix")
try:
prefix = int(prefix)
except (TypeError, ValueError):
prefix = 0
if prefix <= 0:
raise HTTPException(status_code=400, detail="请选择已配置 Apps ID 规则的服务器环境")
prefix_text = str(prefix)
serials = [
int(app_id[len(prefix_text):])
for app_id in apps
if (app_id.isdigit() and app_id.startswith(prefix_text)
and len(app_id) == len(prefix_text) + 2)
]
next_serial = max(serials, default=-1) + 1
if next_serial > 99:
raise HTTPException(status_code=400, detail=f"Apps ID 前缀 {prefix} 的编号已用完")
return f"{prefix}{next_serial:02d}"
def _ensure_upload_keys(config: dict) -> bool:
"""为没有 upload_key 的 app 自动生成，返回是否有变更"""
import uuid
changed = False
for app_id, app in config.get("apps", {}).items():
if not app.get("upload_key"):
app["upload_key"] = uuid.uuid4().hex[:8]
changed = True
return changed
def _migrate_old_themes(config: dict) -> bool:
"""将旧式目录皮肤迁移为 ZIP 存入 data/skins/，返回是否有变更"""
changed = False
themes_dir = AUTOMATION_DIR / "themes"
for app_id, app in config.get("apps", {}).items():
upload_key = app.get("upload_key", "")
if not upload_key:
continue
for cert_type, cert in app.get("certificates", {}).items():
theme = cert.get("theme", "")
if not theme:
continue
# 已经是 ZIP 皮肤（upload_key 目录下有对应 ZIP），跳过
if (SKINS_DIR / upload_key / f"{theme}.zip").exists():
continue
# 在服务自带的默认皮肤目录下查找旧主题
skin_name = Path(theme).name # "AutoPacking/ymh" → "ymh", "ymh" → "ymh"
theme_dir = themes_dir / skin_name
if not theme_dir.is_dir():
continue
app_skins_dir = SKINS_DIR / upload_key
app_skins_dir.mkdir(parents=True, exist_ok=True)
zip_path = app_skins_dir / f"{skin_name}.zip"
# 将目录打包为 ZIP
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
for root, _, files in os.walk(theme_dir):
for f in files:
file_path = Path(root) / f
arcname = file_path.relative_to(theme_dir.parent)
zf.write(file_path, arcname)
# 更新 skins 列表和 theme 引用
skins = app.get("skins", [])
if f"{skin_name}.zip" not in skins:
skins.append(f"{skin_name}.zip")
app["skins"] = skins
if theme != skin_name:
cert["theme"] = skin_name
changed = True
return changed
def load_config() -> dict:
"""读取 config.json"""
if not CONFIG_JSON_PATH.exists():
# 首次部署：优先从项目根目录下的 config.json 导入
if BOOTSTRAP_CONFIG_PATH.exists():
CONFIG_JSON_PATH.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(BOOTSTRAP_CONFIG_PATH, CONFIG_JSON_PATH)
else:
return {
"apps": {}, "schemes": {}, "servers": DEFAULT_SERVERS,
"next_app_id_prefix": 5, "branches": ["main"], "upload": DEFAULT_UPLOAD,
}
with open(CONFIG_JSON_PATH, "r", encoding="utf-8") as f:
config = json.load(f)
# 确保必要字段存在
if "servers" not in config:
config["servers"] = DEFAULT_SERVERS
if "branches" not in config:
config["branches"] = ["main"]
if "upload" not in config:
config["upload"] = DEFAULT_UPLOAD
if "versions" not in config:
config["versions"] = DEFAULT_VERSIONS.copy()
# 自动补齐旧配置的环境前缀、特殊 App 覆盖和 upload_key。
prefixes_changed = _ensure_server_id_prefixes(config)
overrides_changed = _ensure_special_app_prefix_overrides(config)
keys_changed = _ensure_upload_keys(config)
if prefixes_changed or overrides_changed or keys_changed:
save_config(config)
# 迁移旧式目录皮肤为 ZIP
if _migrate_old_themes(config):
save_config(config)
return config
def save_config(config: dict):
"""保存 config.json（自动备份）"""
if CONFIG_JSON_PATH.exists():
backup_path = CONFIG_JSON_PATH.with_suffix(".json.bak")
shutil.copy2(CONFIG_JSON_PATH, backup_path)
with open(CONFIG_JSON_PATH, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2, ensure_ascii=False)
def get_build_config_value(db: Session, key: str, default=None) -> str:
"""获取打包配置值"""
config = db.query(BuildConfig).filter(BuildConfig.key == key).first()
return config.value if config else default
@router.get("")
async def get_full_config():
"""获取完整 config.json"""
return load_config()
@router.put("")
async def update_full_config(config: dict):
"""更新整个 config.json"""
async with _config_lock:
save_config(config)
return {"message": "配置已更新"}
@router.get("/apps")
async def get_apps():
"""获取 apps 配置"""
config = load_config()
return config.get("apps", {})
@router.post("/apps")
async def create_app(app: dict):
"""新增 app"""
import uuid
async with _config_lock:
config = load_config()
apps = config.get("apps", {})
# 按服务环境自动生成三位 ID（例如测试环境 1xx、正式环境 2xx）。
_apply_special_app_prefix_override(app)
new_id = _next_app_id(apps, config.get("servers", {}), app)
# 自动生成 upload_key
if not app.get("upload_key"):
app["upload_key"] = uuid.uuid4().hex[:8]
apps[new_id] = app
config["apps"] = apps
save_config(config)
return {"id": new_id, "upload_key": app["upload_key"], "message": "App 已创建"}
@router.put("/apps/{app_id}")
async def update_app(app_id: str, app: dict):
"""更新指定 app"""
async with _config_lock:
config = load_config()
apps = config.get("apps", {})
if app_id not in apps:
raise HTTPException(status_code=404, detail="App 不存在")
apps[app_id] = app
config["apps"] = apps
save_config(config)
return {"message": "App 已更新"}
@router.delete("/apps/{app_id}")
async def delete_app(app_id: str):
"""删除指定 app"""
async with _config_lock:
config = load_config()
apps = config.get("apps", {})
if app_id not in apps:
raise HTTPException(status_code=404, detail="App 不存在")
del apps[app_id]
config["apps"] = apps
save_config(config)
return {"message": "App 已删除"}
def _find_app_by_upload_key(config: dict, upload_key: str):
"""根据 upload_key 查找 app，返回 (app_id, app) 或 (None, None)"""
for app_id, app in config.get("apps", {}).items():
if app.get("upload_key") == upload_key:
return app_id, app
return None, None
@router.get("/skins/{upload_key}")
async def list_skins(upload_key: str):
"""列出指定 app 的皮肤包"""
config = load_config()
_, app = _find_app_by_upload_key(config, upload_key)
if not app:
raise HTTPException(status_code=404, detail="App 不存在")
app_skins_dir = SKINS_DIR / upload_key
skins = []
if app_skins_dir.exists():
for f in sorted(app_skins_dir.iterdir()):
if f.suffix.lower() == ".zip":
skins.append({
"name": f.name,
"size": f.stat().st_size,
})
return skins
@router.post("/skins/{upload_key}")
async def upload_skin(upload_key: str, file: UploadFile = File(...)):
"""上传皮肤包（ZIP 文件）"""
if not file.filename or not file.filename.lower().endswith(".zip"):
raise HTTPException(status_code=400, detail="仅支持 .zip 格式")
config = load_config()
app_id, _ = _find_app_by_upload_key(config, upload_key)
if not app_id:
raise HTTPException(status_code=404, detail="App 不存在")
app_skins_dir = SKINS_DIR / upload_key
app_skins_dir.mkdir(parents=True, exist_ok=True)
dest = app_skins_dir / file.filename
with open(dest, "wb") as f:
content = await file.read()
f.write(content)
# 更新 config.json 中的 skins 列表
async with _config_lock:
config = load_config()
skins = config["apps"][app_id].get("skins", [])
if file.filename not in skins:
skins.append(file.filename)
config["apps"][app_id]["skins"] = skins
save_config(config)
return {"message": "皮肤包上传成功", "name": file.filename}
@router.put("/skins/{upload_key}/{skin_name}")
async def replace_skin(upload_key: str, skin_name: str, file: UploadFile = File(...)):
"""替换已有皮肤包内容，保持原文件名不变。
APP 配置的 theme 通过文件名（去掉 .zip）引用皮肤包，若上传新文件名会新增一个皮肤而非更新，
导致原引用失效。此接口就地覆盖指定皮肤，确保被引用的皮肤可以更新。
"""
if not file.filename or not file.filename.lower().endswith(".zip"):
raise HTTPException(status_code=400, detail="仅支持 .zip 格式")
config = load_config()
app_id, _ = _find_app_by_upload_key(config, upload_key)
if not app_id:
raise HTTPException(status_code=404, detail="App 不存在")
app_skins_dir = SKINS_DIR / upload_key
skin_path = app_skins_dir / skin_name
if not skin_path.exists():
raise HTTPException(status_code=404, detail="皮肤包不存在")
content = await file.read()
with open(skin_path, "wb") as f:
f.write(content)
return {"message": "皮肤包已更新", "name": skin_name}
@router.delete("/skins/{upload_key}/{skin_name}")
async def delete_skin(upload_key: str, skin_name: str):
"""删除指定皮肤包"""
config = load_config()
app_id, _ = _find_app_by_upload_key(config, upload_key)
if not app_id:
raise HTTPException(status_code=404, detail="App 不存在")
app_skins_dir = SKINS_DIR / upload_key
skin_path = app_skins_dir / skin_name
if not skin_path.exists():
raise HTTPException(status_code=404, detail="皮肤包不存在")
skin_path.unlink()
# 更新 config.json 中的 skins 列表
async with _config_lock:
config = load_config()
skins = config["apps"][app_id].get("skins", [])
if skin_name in skins:
skins.remove(skin_name)
config["apps"][app_id]["skins"] = skins
save_config(config)
return {"message": "皮肤包已删除"}
@router.get("/schemes")
async def get_schemes():
"""获取 schemes 配置"""
config = load_config()
return config.get("schemes", {})
@router.post("/schemes")
async def create_scheme(scheme: dict):
"""新增 scheme"""
async with _config_lock:
config = load_config()
schemes = config.get("schemes", {})
numeric_keys = [int(k) for k in schemes.keys() if k.isdigit()]
new_id = str(max(numeric_keys) + 1) if numeric_keys else "1"
schemes[new_id] = scheme
config["schemes"] = schemes
save_config(config)
return {"id": new_id, "message": "Scheme 已创建"}
@router.put("/schemes/{scheme_id}")
async def update_scheme(scheme_id: str, scheme: dict):
"""更新指定 scheme"""
async with _config_lock:
config = load_config()
schemes = config.get("schemes", {})
if scheme_id not in schemes:
raise HTTPException(status_code=404, detail="Scheme 不存在")
schemes[scheme_id] = scheme
config["schemes"] = schemes
save_config(config)
return {"message": "Scheme 已更新"}
@router.delete("/schemes/{scheme_id}")
async def delete_scheme(scheme_id: str):
"""删除指定 scheme"""
async with _config_lock:
config = load_config()
schemes = config.get("schemes", {})
if scheme_id not in schemes:
raise HTTPException(status_code=404, detail="Scheme 不存在")
del schemes[scheme_id]
config["schemes"] = schemes
save_config(config)
return {"message": "Scheme 已删除"}
# 分支管理 API
@router.get("/branches")
async def get_branches():
"""获取分支列表"""
config = load_config()
return config.get("branches", ["main"])
@router.post("/branches")
async def add_branch(data: dict):
"""新增分支"""
name = data.get("name", "").strip()
if not name:
raise HTTPException(status_code=400, detail="分支名称不能为空")
async with _config_lock:
config = load_config()
branches = config.get("branches", ["main"])
if name in branches:
raise HTTPException(status_code=400, detail="分支已存在")
branches.append(name)
config["branches"] = branches
save_config(config)
return {"message": "分支已添加"}
@router.delete("/branches/{branch_name}")
async def delete_branch(branch_name: str):
"""删除分支"""
async with _config_lock:
config = load_config()
branches = config.get("branches", ["main"])
if branch_name not in branches:
raise HTTPException(status_code=404, detail="分支不存在")
branches.remove(branch_name)
config["branches"] = branches
save_config(config)
return {"message": "分支已删除"}
@router.get("/build")
async def get_build_settings(db: Session = Depends(get_db)):
"""获取打包设置"""
return {
"max_concurrent_builds": int(get_build_config_value(db, "max_concurrent_builds", DEFAULT_MAX_CONCURRENT_BUILDS)),
"build_dir_retention_hours": int(get_build_config_value(db, "build_dir_retention_hours", DEFAULT_BUILD_DIR_RETENTION_HOURS)),
"build_base_dir": get_build_config_value(db, "build_base_dir", str(BUILD_BASE_DIR)),
}
@router.put("/build")
async def update_build_settings(settings: BuildConfigUpdate, db: Session = Depends(get_db)):
"""更新打包设置"""
if settings.max_concurrent_builds is not None:
config = db.query(BuildConfig).filter(BuildConfig.key == "max_concurrent_builds").first()
if config:
config.value = str(settings.max_concurrent_builds)
else:
db.add(BuildConfig(key="max_concurrent_builds", value=str(settings.max_concurrent_builds)))
if settings.build_dir_retention_hours is not None:
config = db.query(BuildConfig).filter(BuildConfig.key == "build_dir_retention_hours").first()
if config:
config.value = str(settings.build_dir_retention_hours)
else:
db.add(BuildConfig(key="build_dir_retention_hours", value=str(settings.build_dir_retention_hours)))
if settings.build_base_dir is not None:
config = db.query(BuildConfig).filter(BuildConfig.key == "build_base_dir").first()
if config:
config.value = settings.build_base_dir
else:
db.add(BuildConfig(key="build_base_dir", value=settings.build_base_dir))
db.commit()
# 运行时更新并发数
if settings.max_concurrent_builds is not None:
from ..services.build_queue import build_queue
build_queue.update_max_concurrent(settings.max_concurrent_builds)
return {"message": "打包设置已更新"}
# 版本号配置 API
@router.get("/versions")
async def get_versions():
"""获取服务端统一管理的 App_Ver 和 Build_Ver"""
config = load_config()
versions = config.get("versions", DEFAULT_VERSIONS.copy())
return {
"app_ver": versions.get("app_ver", DEFAULT_VERSIONS["app_ver"]),
"build_ver": versions.get("build_ver", DEFAULT_VERSIONS["build_ver"]),
}
@router.put("/versions")
async def update_versions(data: dict):
"""更新 App_Ver，并将 Build_Ver 重置为对应的 .0。"""
app_ver = data.get("app_ver", "").strip()
if not re.fullmatch(r"\d+\.\d+\.\d+", app_ver):
raise HTTPException(status_code=400, detail="App_Ver 必须为主版本.次版本.修订号，例如 2.196.0")
build_ver = f"{app_ver}.0"
async with _config_lock:
config = load_config()
config["versions"] = {
"app_ver": app_ver,
"build_ver": build_ver,
}
save_config(config)
return {"message": "版本号已更新", "app_ver": app_ver, "build_ver": build_ver}
# 上传配置 API
@router.get("/upload")
async def get_upload_config():
"""获取上传配置"""
config = load_config()
return config.get("upload", DEFAULT_UPLOAD)
@router.put("/upload")
async def update_upload_config(upload_data: dict):
"""更新上传配置"""
async with _config_lock:
config = load_config()
config["upload"] = upload_data
save_config(config)
return {"message": "上传配置已更新"}
# 服务器环境管理 API
@router.get("/servers")
async def get_servers():
"""获取服务器环境配置"""
config = load_config()
return config.get("servers", DEFAULT_SERVERS)
@router.post("/servers")
async def create_server(server_data: dict):
"""新增服务器环境"""
async with _config_lock:
config = load_config()
servers = config.get("servers", DEFAULT_SERVERS)
name = server_data.get("name", "").strip()
if not name:
raise HTTPException(status_code=400, detail="环境名称不能为空")
if name in servers:
raise HTTPException(status_code=400, detail="环境名称已存在")
prefix = config.get("next_app_id_prefix", 1)
servers[name] = {
"api": server_data.get("api", ""),
"assDom": server_data.get("assDom", ""),
"universalLink": server_data.get("universalLink", ""),
"app_id_prefix": prefix,
}
config["servers"] = servers
config["next_app_id_prefix"] = prefix + 1
save_config(config)
return {"message": "服务器环境已创建"}
@router.put("/servers/{server_name}")
async def update_server(server_name: str, server_data: dict):
"""更新服务器环境"""
async with _config_lock:
config = load_config()
servers = config.get("servers", DEFAULT_SERVERS)
if server_name not in servers:
raise HTTPException(status_code=404, detail="环境不存在")
# 如果名称变了，需要删除旧的
new_name = server_data.get("name", server_name).strip()
if new_name != server_name:
if new_name in servers:
raise HTTPException(status_code=400, detail="新环境名称已存在")
servers[new_name] = servers.pop(server_name)
else:
new_name = server_name
servers[new_name] = {
"api": server_data.get("api", ""),
"assDom": server_data.get("assDom", ""),
"universalLink": server_data.get("universalLink", ""),
"app_id_prefix": servers[new_name].get("app_id_prefix"),
}
config["servers"] = servers
save_config(config)
return {"message": "服务器环境已更新"}
@router.delete("/servers/{server_name}")
async def delete_server(server_name: str):
"""删除服务器环境"""
async with _config_lock:
config = load_config()
servers = config.get("servers", DEFAULT_SERVERS)
if server_name not in servers:
raise HTTPException(status_code=404, detail="环境不存在")
# 检查是否有 App 在使用此环境
apps = config.get("apps", {})
for app in apps.values():
if app.get("server") == server_name:
raise HTTPException(status_code=400, detail=f"无法删除：有 App 正在使用此环境")
del servers[server_name]
config["servers"] = servers
save_config(config)
return {"message": "服务器环境已删除"}
Reference in New Issue View Git Blame Copy Permalink