137 lines
4.5 KiB
Python
137 lines
4.5 KiB
Python
"""Автообновление (Gitea): публичные поля в БД + админ API."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from typing import Any, Dict
|
||
|
||
from sqlalchemy.exc import IntegrityError
|
||
|
||
from app import db
|
||
from app.models.auto_update_settings import AutoUpdateSettings
|
||
from app.services.gitea_secrets_file import gitea_file_secret_flags
|
||
from config import _resolve_base_dir
|
||
|
||
_ADMIN_KEYS = frozenset(
|
||
{
|
||
"enabled",
|
||
"auto_install",
|
||
"gitea_url",
|
||
"gitea_owner",
|
||
"gitea_repo",
|
||
"repository_url",
|
||
"check_interval_sec",
|
||
}
|
||
)
|
||
|
||
|
||
def ensure_auto_update_row() -> AutoUpdateSettings:
|
||
row = db.session.get(AutoUpdateSettings, 1)
|
||
if row is not None:
|
||
return row
|
||
row = AutoUpdateSettings(id=1)
|
||
db.session.add(row)
|
||
try:
|
||
db.session.commit()
|
||
except IntegrityError:
|
||
db.session.rollback()
|
||
row = db.session.get(AutoUpdateSettings, 1)
|
||
if row is None:
|
||
raise
|
||
return row
|
||
|
||
|
||
def bootstrap_update_environment(app) -> None:
|
||
"""Строка настроек id=1 и data/config.json с version при первом старте."""
|
||
import json
|
||
|
||
from config import get_wesp_data_config_path, read_wesp_data_config
|
||
|
||
ensure_auto_update_row()
|
||
path = get_wesp_data_config_path()
|
||
ver = str(app.config.get("SYNC_CLIENT_VERSION", "2.0.0") or "2.0.0")
|
||
data = read_wesp_data_config()
|
||
if not path.is_file():
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(
|
||
json.dumps({"version": ver}, ensure_ascii=False, indent=2) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
elif not str(data.get("version") or "").strip():
|
||
data["version"] = ver
|
||
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
|
||
|
||
def public_auto_update_settings_payload() -> Dict[str, Any]:
|
||
row = db.session.get(AutoUpdateSettings, 1)
|
||
if row is None:
|
||
interval = 3600
|
||
enabled = auto_install = False
|
||
gitea_url = gitea_owner = gitea_repo = repository_url = ""
|
||
else:
|
||
interval = row.check_interval_sec or 3600
|
||
try:
|
||
interval = int(interval)
|
||
except (TypeError, ValueError):
|
||
interval = 3600
|
||
interval = max(60, min(86400, interval))
|
||
enabled = bool(row.enabled)
|
||
auto_install = bool(row.auto_install)
|
||
gitea_url = str(row.gitea_url or "")
|
||
gitea_owner = str(row.gitea_owner or "")
|
||
gitea_repo = str(row.gitea_repo or "")
|
||
repository_url = str(row.repository_url or "")
|
||
|
||
secrets_file = _resolve_base_dir() / ".secret" / "gitea_secrets.json"
|
||
file_flags = gitea_file_secret_flags(str(_resolve_base_dir()))
|
||
restart_cmd = str(os.getenv("WESP_UPDATE_RESTART_CMD", "") or "").strip()
|
||
payload: Dict[str, Any] = {
|
||
"enabled": enabled,
|
||
"auto_install": auto_install,
|
||
"gitea_url": gitea_url,
|
||
"gitea_owner": gitea_owner,
|
||
"gitea_repo": gitea_repo,
|
||
"repository_url": repository_url,
|
||
"check_interval_sec": interval,
|
||
"gitea_token_via_env": bool(os.getenv("GITEA_TOKEN", "").strip()),
|
||
"gitea_secrets_file": secrets_file.is_file(),
|
||
**file_flags,
|
||
"storage": "sqlite:auto_update_settings",
|
||
"restart_configured": bool(restart_cmd),
|
||
}
|
||
try:
|
||
from app.services.update_api_service import build_updates_status_payload
|
||
|
||
runtime = build_updates_status_payload()
|
||
payload.update(
|
||
{
|
||
"runtime_initialized": runtime.get("initialized"),
|
||
"runtime_update_available": runtime.get("update_available"),
|
||
"runtime_pending_version": runtime.get("pending_version"),
|
||
"runtime_last_check_at": runtime.get("last_check_at"),
|
||
"runtime_current_version": runtime.get("current_version"),
|
||
}
|
||
)
|
||
except Exception:
|
||
pass
|
||
return payload
|
||
|
||
|
||
def apply_auto_update_admin_patch(updates: Dict[str, Any]) -> Dict[str, Any]:
|
||
row = ensure_auto_update_row()
|
||
for k, raw in updates.items():
|
||
if k not in _ADMIN_KEYS:
|
||
continue
|
||
if k in ("enabled", "auto_install"):
|
||
setattr(row, k, bool(raw))
|
||
elif k == "check_interval_sec":
|
||
try:
|
||
n = int(raw)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
row.check_interval_sec = max(60, min(86400, n))
|
||
else:
|
||
setattr(row, k, str(raw).strip() if raw is not None else "")
|
||
db.session.commit()
|
||
return public_auto_update_settings_payload()
|