yuanzhen869/MobileModels
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve manual catalog UX

Browse Source
This commit is contained in:
yuanzhen869
2026-04-15 09:15:48 +08:00
parent d29cbed1fa
commit f6541d291e
2 changed files with 184 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/web_server.py
+63 -11
View File
@@ -33,11 +33,19 @@ SYNC_LOCK = threading.Lock()
SCHEDULE_LOCK = threading.Lock()
MYSQL_CONFIG_LOCK = threading.Lock()
INDEX_ALIAS_LOCK = threading.Lock()
MANUAL_REBUILD_LOCK = threading.Lock()
NORMALIZE_RE = re.compile(r"[^0-9a-z\u4e00-\u9fff]+")
SCHEDULE_TIME_RE = re.compile(r"^(?:[01]?\d|2[0-3]):[0-5]\d$")
SCHEDULER_POLL_SECONDS = 20
INDEX_DEVICE_NAME_ALIAS_MAP: dict[str, list[str]] | None = None
DEVICE_TYPES = {"phone", "tablet", "wear", "tv", "computer", "other"}
LAST_MANUAL_MYSQL_LOAD_STATUS: dict[str, object] = {
"running": False,
"last_started_at": None,
"last_finished_at": None,
"last_status": None,
"last_message": None,
}
def truthy_env(name: str, default: str = "0") -> bool:
@@ -471,7 +479,34 @@ def count_manual_alias_conflicts(device: dict[str, object]) -> int:
return len(conflicts)
def rebuild_generated_outputs() -> dict[str, object]:
def _manual_mysql_loader_task() -> None:
with MANUAL_REBUILD_LOCK:
LAST_MANUAL_MYSQL_LOAD_STATUS["running"] = True
LAST_MANUAL_MYSQL_LOAD_STATUS["last_started_at"] = local_now().isoformat(timespec="seconds")
LAST_MANUAL_MYSQL_LOAD_STATUS["last_status"] = "running"
LAST_MANUAL_MYSQL_LOAD_STATUS["last_message"] = "手动补录触发的 MySQL 刷新进行中。"
try:
load_proc = run_command(["python3", str(MYSQL_LOADER)])
message = "\n".join(part for part in [load_proc.stdout.strip(), load_proc.stderr.strip()] if part).strip() or "MySQL 已刷新。"
LAST_MANUAL_MYSQL_LOAD_STATUS["last_status"] = "success" if load_proc.returncode == 0 else "failed"
LAST_MANUAL_MYSQL_LOAD_STATUS["last_message"] = message
except Exception as err:
LAST_MANUAL_MYSQL_LOAD_STATUS["last_status"] = "failed"
LAST_MANUAL_MYSQL_LOAD_STATUS["last_message"] = str(err)
finally:
LAST_MANUAL_MYSQL_LOAD_STATUS["running"] = False
LAST_MANUAL_MYSQL_LOAD_STATUS["last_finished_at"] = local_now().isoformat(timespec="seconds")
def start_manual_mysql_loader() -> bool:
if LAST_MANUAL_MYSQL_LOAD_STATUS.get("running"):
return False
thread = threading.Thread(target=_manual_mysql_loader_task, name="manual-mysql-loader", daemon=True)
thread.start()
return True
def rebuild_generated_outputs(*, defer_mysql_load: bool = False) -> dict[str, object]:
build_proc = run_command(
[
"python3",
@@ -504,16 +539,26 @@ def rebuild_generated_outputs() -> dict[str, object]:
mysql_loaded = False
mysql_message = "MySQL 未刷新。"
if mysql_auto_load_enabled():
load_proc = run_command(["python3", str(MYSQL_LOADER)])
mysql_message = "\n".join(part for part in [load_proc.stdout.strip(), load_proc.stderr.strip()] if part).strip() or "MySQL 已刷新。"
if load_proc.returncode != 0:
raise RuntimeError(mysql_message)
mysql_loaded = True
if defer_mysql_load:
started = start_manual_mysql_loader()
mysql_message = "MySQL 后台刷新中。" if started else "MySQL 后台刷新已在进行中。"
else:
load_proc = run_command(["python3", str(MYSQL_LOADER)])
mysql_message = "\n".join(part for part in [load_proc.stdout.strip(), load_proc.stderr.strip()] if part).strip() or "MySQL 已刷新。"
if load_proc.returncode != 0:
raise RuntimeError(mysql_message)
mysql_loaded = True
return {
"index_updated": True,
"mysql_seed_updated": True,
"mysql_loaded": mysql_loaded,
"message": "本地覆盖库已保存，索引与 MySQL seed 已刷新。" if mysql_loaded else "本地覆盖库已保存，索引与 MySQL seed 已刷新，MySQL 未自动装载。",
"message": (
"本地覆盖库已保存，索引与 MySQL seed 已刷新，MySQL 正在后台刷新。"
if defer_mysql_load and mysql_auto_load_enabled()
else "本地覆盖库已保存，索引与 MySQL seed 已刷新。"
if mysql_loaded
else "本地覆盖库已保存，索引与 MySQL seed 已刷新，MySQL 未自动装载。"
),
"build_output": build_output,
"mysql_seed_output": seed_output,
"mysql_message": mysql_message,
@@ -532,6 +577,13 @@ def manual_catalog_payload() -> dict[str, object]:
"device_count": len(devices),
},
"catalog_file": str(MANUAL_CATALOG_PATH.relative_to(PROJECT_ROOT)),
"mysql_refresh": {
"running": bool(LAST_MANUAL_MYSQL_LOAD_STATUS.get("running")),
"last_started_at": LAST_MANUAL_MYSQL_LOAD_STATUS.get("last_started_at"),
"last_finished_at": LAST_MANUAL_MYSQL_LOAD_STATUS.get("last_finished_at"),
"last_status": LAST_MANUAL_MYSQL_LOAD_STATUS.get("last_status"),
"last_message": LAST_MANUAL_MYSQL_LOAD_STATUS.get("last_message"),
},
}
@@ -552,7 +604,7 @@ def upsert_manual_brand(payload: dict[str, object]) -> dict[str, object]:
catalog["brands"].append(incoming)
validated = validate_manual_catalog(catalog)
write_manual_catalog(validated)
rebuild_result = rebuild_generated_outputs()
rebuild_result = rebuild_generated_outputs(defer_mysql_load=True)
return {
"saved_brand": incoming,
"catalog": manual_catalog_payload(),
@@ -576,7 +628,7 @@ def upsert_manual_device(payload: dict[str, object]) -> dict[str, object]:
catalog["devices"].append(device_payload)
validated = validate_manual_catalog(catalog)
write_manual_catalog(validated)
rebuild_result = rebuild_generated_outputs()
rebuild_result = rebuild_generated_outputs(defer_mysql_load=True)
return {
"saved_device": device_payload,
"alias_conflict_count": count_manual_alias_conflicts(device_payload),
@@ -602,7 +654,7 @@ def delete_manual_brand(payload: dict[str, object]) -> dict[str, object]:
if len(next_brands) == len(catalog["brands"]):
raise RuntimeError(f"未找到品牌: {brand_name}")
write_manual_catalog({"brands": next_brands, "devices": catalog["devices"]})
rebuild_result = rebuild_generated_outputs()
rebuild_result = rebuild_generated_outputs(defer_mysql_load=True)
return {
"deleted_brand": brand_name,
"catalog": manual_catalog_payload(),
@@ -624,7 +676,7 @@ def delete_manual_device(payload: dict[str, object]) -> dict[str, object]:
if len(next_devices) == len(catalog["devices"]):
raise RuntimeError(f"未找到设备: {device_id}")
write_manual_catalog({"brands": catalog["brands"], "devices": next_devices})
rebuild_result = rebuild_generated_outputs()
rebuild_result = rebuild_generated_outputs(defer_mysql_load=True)
return {
"deleted_device": device_id,
"catalog": manual_catalog_payload(),