feat: 皮肤包上传管理功能,持久化数据目录
- 新增皮肤包上传/列表/删除 API(基于 upload_key) - 构建时自动解压 ZIP 皮肤包应用到项目 - 配置和皮肤包迁移到 backend/data/ 目录,切分支不会丢失 - 旧式目录皮肤自动迁移为 ZIP 格式 - 前端皮肤包管理 UI(上传、删除、下拉选择) - .gitignore 排除运行时数据目录
This commit is contained in:
+145
-3
@@ -1,15 +1,18 @@
|
||||
"""配置管理 API"""
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import zipfile
|
||||
from datetime import datetime
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from pathlib import Path
|
||||
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ..database import get_db
|
||||
from ..models import BuildConfig
|
||||
from ..schemas import BuildConfigUpdate
|
||||
from ..config import CONFIG_JSON_PATH, DEFAULT_MAX_CONCURRENT_BUILDS, DEFAULT_BUILD_DIR_RETENTION_HOURS
|
||||
from ..config import CONFIG_JSON_PATH, DEFAULT_MAX_CONCURRENT_BUILDS, DEFAULT_BUILD_DIR_RETENTION_HOURS, SKINS_DIR, AUTOPACKING_DIR
|
||||
from ..deps import get_current_user
|
||||
|
||||
router = APIRouter(prefix="/api/config", tags=["config"], dependencies=[Depends(get_current_user)])
|
||||
@@ -76,10 +79,60 @@ def _ensure_upload_keys(config: dict) -> bool:
|
||||
return changed
|
||||
|
||||
|
||||
def _migrate_old_themes(config: dict) -> bool:
|
||||
"""将旧式目录皮肤迁移为 ZIP 存入 data/skins/,返回是否有变更"""
|
||||
changed = False
|
||||
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
|
||||
# 在 AutoPacking 目录下查找皮肤目录
|
||||
skin_name = Path(theme).name # "AutoPacking/ymh" → "ymh", "ymh" → "ymh"
|
||||
theme_dir = AUTOPACKING_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():
|
||||
return {"apps": {}, "schemes": {}, "servers": DEFAULT_SERVERS, "branches": ["main"], "upload": DEFAULT_UPLOAD}
|
||||
# 迁移:从旧的 AutoPacking/config.json 迁移到持久化目录
|
||||
old_config = AUTOPACKING_DIR / "config.json"
|
||||
if old_config.exists():
|
||||
CONFIG_JSON_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(old_config, CONFIG_JSON_PATH)
|
||||
# 迁移旧的 skins 目录
|
||||
old_skins = AUTOPACKING_DIR / "skins"
|
||||
if old_skins.exists():
|
||||
shutil.copytree(old_skins, SKINS_DIR, dirs_exist_ok=True)
|
||||
else:
|
||||
return {"apps": {}, "schemes": {}, "servers": DEFAULT_SERVERS, "branches": ["main"], "upload": DEFAULT_UPLOAD}
|
||||
with open(CONFIG_JSON_PATH, "r", encoding="utf-8") as f:
|
||||
config = json.load(f)
|
||||
# 确保必要字段存在
|
||||
@@ -92,6 +145,9 @@ def load_config() -> dict:
|
||||
# 自动为缺少 upload_key 的 app 生成唯一标识
|
||||
if _ensure_upload_keys(config):
|
||||
save_config(config)
|
||||
# 迁移旧式目录皮肤为 ZIP
|
||||
if _migrate_old_themes(config):
|
||||
save_config(config)
|
||||
return config
|
||||
|
||||
|
||||
@@ -185,6 +241,92 @@ async def delete_app(app_id: str):
|
||||
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.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 配置"""
|
||||
|
||||
Reference in New Issue
Block a user