59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
import json
|
|
|
|
from fastapi import HTTPException
|
|
|
|
|
|
def policy_row_to_dict(row) -> dict:
|
|
if not row:
|
|
return {
|
|
"policy_seq": 1,
|
|
"force_update": False,
|
|
"allow_rollback": False,
|
|
"offline_allowed": True,
|
|
"valid_until": "2099-12-31T23:59:59Z",
|
|
"min_supported_version": "",
|
|
"disabled_versions": [],
|
|
"git_tags_enabled": False,
|
|
"message": "",
|
|
}
|
|
try:
|
|
disabled = json.loads(row["disabled_versions"] or "[]")
|
|
except (TypeError, json.JSONDecodeError):
|
|
disabled = []
|
|
return {
|
|
"policy_seq": int(row["policy_seq"]),
|
|
"force_update": bool(row["force_update"]),
|
|
"allow_rollback": bool(row["allow_rollback"]),
|
|
"offline_allowed": bool(row["offline_allowed"]),
|
|
"valid_until": row["valid_until"],
|
|
"min_supported_version": row["min_supported_version"] or "",
|
|
"disabled_versions": disabled if isinstance(disabled, list) else [],
|
|
"git_tags_enabled": bool(row["git_tags_enabled"]) if "git_tags_enabled" in row.keys() else False,
|
|
"message": row["message"] or "",
|
|
}
|
|
|
|
|
|
def validate_channel_code(code: str) -> bool:
|
|
return bool(code) and len(code) <= 32 and code[0].isalnum() and all(c.isalnum() or c in "_-" for c in code)
|
|
|
|
|
|
def require_channel(conn, app_id: str, channel: str, require_enabled: bool = True):
|
|
row = conn.execute("SELECT * FROM channels WHERE app_id=? AND channel_code=?", (app_id, channel)).fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=400, detail={"error": "channel_not_found", "msg": f"渠道 {channel} 不存在"})
|
|
if require_enabled and not row["enabled"]:
|
|
raise HTTPException(status_code=403, detail={"error": "channel_disabled", "msg": f"渠道 {channel} 已停用"})
|
|
return row
|
|
|
|
|
|
def is_executable_path(path: str) -> bool:
|
|
return path.lower().endswith((".exe", ".dll"))
|
|
|
|
|
|
def version_key(version: str):
|
|
parts = []
|
|
for part in version.split("."):
|
|
digits = "".join(ch for ch in part if ch.isdigit())
|
|
parts.append(int(digits) if digits else 0)
|
|
return tuple(parts)
|