138 lines
4.4 KiB
Python
138 lines
4.4 KiB
Python
"""
|
|
Чтение/запись настроек автообновления в recipes.db без Flask (update.py, миграции).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
AUTO_UPDATE_TABLE = "auto_update_settings"
|
|
|
|
_MERGE_KEYS = frozenset(
|
|
{
|
|
"enabled",
|
|
"auto_install",
|
|
"gitea_url",
|
|
"gitea_owner",
|
|
"gitea_repo",
|
|
"repository_url",
|
|
"check_interval_sec",
|
|
}
|
|
)
|
|
|
|
|
|
def recipes_db_path(base_dir: str) -> Path:
|
|
return Path(base_dir).resolve() / "data" / "recipes.db"
|
|
|
|
|
|
def load_auto_update_dict(base_dir: str) -> Optional[Dict[str, Any]]:
|
|
path = recipes_db_path(base_dir)
|
|
if not path.is_file():
|
|
return None
|
|
conn = sqlite3.connect(str(path))
|
|
try:
|
|
conn.row_factory = sqlite3.Row
|
|
row = conn.execute(
|
|
f"""
|
|
SELECT enabled, auto_install, gitea_url, gitea_owner, gitea_repo,
|
|
repository_url, check_interval_sec
|
|
FROM {AUTO_UPDATE_TABLE}
|
|
WHERE id = 1
|
|
"""
|
|
).fetchone()
|
|
except sqlite3.OperationalError:
|
|
return None
|
|
finally:
|
|
conn.close()
|
|
if row is None:
|
|
return None
|
|
try:
|
|
interval = int(row["check_interval_sec"] or 3600)
|
|
except (TypeError, ValueError):
|
|
interval = 3600
|
|
interval = max(60, min(86400, interval))
|
|
return {
|
|
"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 ""),
|
|
"check_interval_sec": interval,
|
|
}
|
|
|
|
|
|
def merge_auto_update_settings(base_dir: str, updates: Dict[str, Any]) -> None:
|
|
"""Частично обновить публичные поля id=1 в recipes.db (CLI, без Flask)."""
|
|
path = recipes_db_path(base_dir)
|
|
if not path.is_file():
|
|
raise FileNotFoundError(str(path))
|
|
cur = load_auto_update_dict(base_dir)
|
|
if cur is None:
|
|
cur = {
|
|
"enabled": False,
|
|
"auto_install": False,
|
|
"gitea_url": "",
|
|
"gitea_owner": "",
|
|
"gitea_repo": "",
|
|
"repository_url": "",
|
|
"check_interval_sec": 3600,
|
|
}
|
|
for k, raw in updates.items():
|
|
if k not in _MERGE_KEYS:
|
|
continue
|
|
if k == "check_interval_sec":
|
|
try:
|
|
n = int(raw)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
cur["check_interval_sec"] = max(60, min(86400, n))
|
|
elif k in ("enabled", "auto_install"):
|
|
cur[k] = bool(raw)
|
|
else:
|
|
cur[k] = str(raw).strip() if raw is not None else ""
|
|
upsert_auto_update_dict(base_dir, cur)
|
|
|
|
|
|
def upsert_auto_update_dict(base_dir: str, data: Dict[str, Any]) -> None:
|
|
path = recipes_db_path(base_dir)
|
|
if not path.is_file():
|
|
raise OSError(f"recipes.db не найден: {path}")
|
|
try:
|
|
interval = int(data.get("check_interval_sec", 3600) or 3600)
|
|
except (TypeError, ValueError):
|
|
interval = 3600
|
|
interval = max(60, min(86400, interval))
|
|
conn = sqlite3.connect(str(path))
|
|
try:
|
|
conn.execute(
|
|
f"""
|
|
INSERT INTO {AUTO_UPDATE_TABLE} (
|
|
id, enabled, auto_install, gitea_url, gitea_owner, gitea_repo,
|
|
repository_url, check_interval_sec
|
|
) VALUES (1, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
enabled = excluded.enabled,
|
|
auto_install = excluded.auto_install,
|
|
gitea_url = excluded.gitea_url,
|
|
gitea_owner = excluded.gitea_owner,
|
|
gitea_repo = excluded.gitea_repo,
|
|
repository_url = excluded.repository_url,
|
|
check_interval_sec = excluded.check_interval_sec
|
|
""",
|
|
(
|
|
1 if data.get("enabled") else 0,
|
|
1 if data.get("auto_install") else 0,
|
|
str(data.get("gitea_url") or "")[:512],
|
|
str(data.get("gitea_owner") or "")[:255],
|
|
str(data.get("gitea_repo") or "")[:255],
|
|
str(data.get("repository_url") or "")[:512],
|
|
interval,
|
|
),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|