337 lines
12 KiB
Python
337 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from collections.abc import Iterator, Sequence
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS refresh_runs (
|
|
id TEXT PRIMARY KEY,
|
|
trigger TEXT NOT NULL,
|
|
started_at TEXT NOT NULL,
|
|
finished_at TEXT,
|
|
status TEXT NOT NULL,
|
|
collection_items INTEGER NOT NULL DEFAULT 0,
|
|
unique_items INTEGER NOT NULL DEFAULT 0,
|
|
ready_items INTEGER NOT NULL DEFAULT 0,
|
|
downloads_requested INTEGER NOT NULL DEFAULT 0,
|
|
downloads_succeeded INTEGER NOT NULL DEFAULT 0,
|
|
downloads_failed INTEGER NOT NULL DEFAULT 0,
|
|
bytes_before INTEGER NOT NULL DEFAULT 0,
|
|
bytes_after INTEGER NOT NULL DEFAULT 0,
|
|
bytes_downloaded INTEGER NOT NULL DEFAULT 0,
|
|
cache_bytes INTEGER NOT NULL DEFAULT 0,
|
|
elapsed_seconds REAL NOT NULL DEFAULT 0,
|
|
average_mbps REAL NOT NULL DEFAULT 0,
|
|
error TEXT
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS mods (
|
|
workshop_id TEXT PRIMARY KEY,
|
|
title TEXT NOT NULL DEFAULT '',
|
|
package_id TEXT,
|
|
steam_app_id TEXT,
|
|
display_name TEXT,
|
|
time_updated INTEGER NOT NULL DEFAULT 0,
|
|
metadata_file_size INTEGER NOT NULL DEFAULT 0,
|
|
disk_size INTEGER NOT NULL DEFAULT 0,
|
|
local_path TEXT,
|
|
status TEXT NOT NULL DEFAULT 'pending',
|
|
error TEXT,
|
|
last_checked_at TEXT,
|
|
last_downloaded_at TEXT
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS collection_members (
|
|
collection_id TEXT NOT NULL,
|
|
collection_name TEXT NOT NULL,
|
|
workshop_id TEXT NOT NULL,
|
|
position INTEGER NOT NULL,
|
|
PRIMARY KEY (collection_id, workshop_id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS download_attempts (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
run_id TEXT NOT NULL,
|
|
batch_number INTEGER NOT NULL,
|
|
item_count INTEGER NOT NULL,
|
|
workshop_ids TEXT NOT NULL,
|
|
started_at TEXT NOT NULL,
|
|
finished_at TEXT NOT NULL,
|
|
elapsed_seconds REAL NOT NULL,
|
|
exit_code INTEGER,
|
|
status TEXT NOT NULL,
|
|
output_tail TEXT,
|
|
FOREIGN KEY (run_id) REFERENCES refresh_runs(id)
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_runs_started ON refresh_runs(started_at DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_mods_status ON mods(status);
|
|
CREATE INDEX IF NOT EXISTS idx_attempts_run ON download_attempts(run_id);
|
|
"""
|
|
|
|
|
|
class Database:
|
|
def __init__(self, path: Path):
|
|
self.path = path
|
|
|
|
@contextmanager
|
|
def connect(self) -> Iterator[sqlite3.Connection]:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
connection = sqlite3.connect(self.path, timeout=30)
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA foreign_keys = ON")
|
|
connection.execute("PRAGMA journal_mode = WAL")
|
|
try:
|
|
yield connection
|
|
connection.commit()
|
|
finally:
|
|
connection.close()
|
|
|
|
def initialize(self) -> None:
|
|
with self.connect() as connection:
|
|
connection.executescript(SCHEMA)
|
|
|
|
def execute(self, sql: str, parameters: Sequence[Any] = ()) -> None:
|
|
with self.connect() as connection:
|
|
connection.execute(sql, parameters)
|
|
|
|
def fetch_one(
|
|
self, sql: str, parameters: Sequence[Any] = ()
|
|
) -> dict[str, Any] | None:
|
|
with self.connect() as connection:
|
|
row = connection.execute(sql, parameters).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def fetch_all(
|
|
self, sql: str, parameters: Sequence[Any] = ()
|
|
) -> list[dict[str, Any]]:
|
|
with self.connect() as connection:
|
|
rows = connection.execute(sql, parameters).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def create_run(self, run_id: str, trigger: str, started_at: str) -> None:
|
|
self.execute(
|
|
"INSERT INTO refresh_runs (id, trigger, started_at, status) VALUES (?, ?, ?, 'running')",
|
|
(run_id, trigger, started_at),
|
|
)
|
|
|
|
def finish_run(self, run_id: str, values: dict[str, Any]) -> None:
|
|
allowed = {
|
|
"finished_at",
|
|
"status",
|
|
"collection_items",
|
|
"unique_items",
|
|
"ready_items",
|
|
"downloads_requested",
|
|
"downloads_succeeded",
|
|
"downloads_failed",
|
|
"bytes_before",
|
|
"bytes_after",
|
|
"bytes_downloaded",
|
|
"cache_bytes",
|
|
"elapsed_seconds",
|
|
"average_mbps",
|
|
"error",
|
|
}
|
|
selected = {key: value for key, value in values.items() if key in allowed}
|
|
assignments = ", ".join(f"{key} = ?" for key in selected)
|
|
self.execute(
|
|
f"UPDATE refresh_runs SET {assignments} WHERE id = ?",
|
|
(*selected.values(), run_id),
|
|
)
|
|
|
|
def replace_collection_members(
|
|
self, collections: dict[str, tuple[str, list[str]]]
|
|
) -> None:
|
|
with self.connect() as connection:
|
|
connection.execute("DELETE FROM collection_members")
|
|
for name, (collection_id, workshop_ids) in collections.items():
|
|
connection.executemany(
|
|
"""
|
|
INSERT INTO collection_members
|
|
(collection_id, collection_name, workshop_id, position)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
[
|
|
(collection_id, name, workshop_id, position)
|
|
for position, workshop_id in enumerate(workshop_ids)
|
|
],
|
|
)
|
|
|
|
def _upsert_mod(
|
|
self, connection: sqlite3.Connection, values: dict[str, Any]
|
|
) -> None:
|
|
columns = [
|
|
"workshop_id",
|
|
"title",
|
|
"package_id",
|
|
"steam_app_id",
|
|
"display_name",
|
|
"time_updated",
|
|
"metadata_file_size",
|
|
"disk_size",
|
|
"local_path",
|
|
"status",
|
|
"error",
|
|
"last_checked_at",
|
|
"last_downloaded_at",
|
|
]
|
|
workshop_id = str(values["workshop_id"])
|
|
existing_row = connection.execute(
|
|
"SELECT * FROM mods WHERE workshop_id = ?", (workshop_id,)
|
|
).fetchone()
|
|
existing = dict(existing_row) if existing_row else {}
|
|
defaults = {
|
|
"title": "",
|
|
"time_updated": 0,
|
|
"metadata_file_size": 0,
|
|
"disk_size": 0,
|
|
"status": "pending",
|
|
}
|
|
payload: dict[str, Any] = {"workshop_id": workshop_id}
|
|
for column in columns[1:]:
|
|
candidate = values.get(column)
|
|
if candidate is not None:
|
|
payload[column] = candidate
|
|
elif column in existing:
|
|
payload[column] = existing[column]
|
|
else:
|
|
payload[column] = defaults.get(column)
|
|
updates = ", ".join(
|
|
f"{column} = excluded.{column}"
|
|
for column in columns
|
|
if column != "workshop_id"
|
|
)
|
|
placeholders = ", ".join("?" for _ in columns)
|
|
connection.execute(
|
|
f"""
|
|
INSERT INTO mods ({", ".join(columns)}) VALUES ({placeholders})
|
|
ON CONFLICT(workshop_id) DO UPDATE SET {updates}
|
|
""",
|
|
tuple(payload[column] for column in columns),
|
|
)
|
|
|
|
def upsert_mod(self, values: dict[str, Any]) -> None:
|
|
with self.connect() as connection:
|
|
self._upsert_mod(connection, values)
|
|
|
|
def upsert_mods(self, values: Sequence[dict[str, Any]]) -> None:
|
|
"""Upsert a metadata page in one transaction instead of per-mod commits."""
|
|
with self.connect() as connection:
|
|
for value in values:
|
|
self._upsert_mod(connection, value)
|
|
|
|
def add_download_attempt(self, values: dict[str, Any]) -> None:
|
|
self.execute(
|
|
"""
|
|
INSERT INTO download_attempts
|
|
(run_id, batch_number, item_count, workshop_ids, started_at,
|
|
finished_at, elapsed_seconds, exit_code, status, output_tail)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
values["run_id"],
|
|
values["batch_number"],
|
|
values["item_count"],
|
|
json.dumps(values["workshop_ids"]),
|
|
values["started_at"],
|
|
values["finished_at"],
|
|
values["elapsed_seconds"],
|
|
values.get("exit_code"),
|
|
values["status"],
|
|
values.get("output_tail"),
|
|
),
|
|
)
|
|
|
|
def dashboard(self) -> dict[str, Any]:
|
|
recent_runs = self.runs(1)
|
|
latest = recent_runs[0] if recent_runs else None
|
|
last_update = self.fetch_one(
|
|
"""
|
|
SELECT finished_at
|
|
FROM refresh_runs
|
|
WHERE finished_at IS NOT NULL AND unique_items > 0
|
|
ORDER BY finished_at DESC
|
|
LIMIT 1
|
|
"""
|
|
)
|
|
totals = (
|
|
self.fetch_one(
|
|
"""
|
|
SELECT COUNT(*) AS total_mods,
|
|
SUM(CASE WHEN status = 'ready' THEN 1 ELSE 0 END) AS ready_mods,
|
|
SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) AS failed_mods,
|
|
SUM(CASE WHEN status = 'skipped' THEN 1 ELSE 0 END) AS skipped_mods,
|
|
COALESCE(SUM(disk_size), 0) AS cache_bytes
|
|
FROM mods
|
|
WHERE workshop_id IN (SELECT DISTINCT workshop_id FROM collection_members)
|
|
"""
|
|
)
|
|
or {}
|
|
)
|
|
collections = self.fetch_all(
|
|
"""
|
|
SELECT collection_name AS name, collection_id AS id, COUNT(*) AS item_count
|
|
FROM collection_members
|
|
GROUP BY collection_name, collection_id
|
|
ORDER BY CASE collection_name
|
|
WHEN 'Core' THEN 1 WHEN 'Content' THEN 2 WHEN 'Cosmetics' THEN 3 ELSE 4 END
|
|
"""
|
|
)
|
|
return {
|
|
"latest_run": latest,
|
|
"rml_updated_at": last_update["finished_at"] if last_update else None,
|
|
"totals": totals,
|
|
"collections": collections,
|
|
}
|
|
|
|
def runs(self, limit: int = 48) -> list[dict[str, Any]]:
|
|
return self.fetch_all(
|
|
"""
|
|
SELECT r.*,
|
|
COALESCE(SUM(a.elapsed_seconds), 0) AS download_elapsed_seconds,
|
|
COUNT(a.id) AS download_batches
|
|
FROM refresh_runs r
|
|
LEFT JOIN download_attempts a ON a.run_id = r.id
|
|
GROUP BY r.id
|
|
ORDER BY r.started_at DESC
|
|
LIMIT ?
|
|
""",
|
|
(limit,),
|
|
)
|
|
|
|
def download_attempts(self, run_id: str, limit: int = 100) -> list[dict[str, Any]]:
|
|
return self.fetch_all(
|
|
"""
|
|
SELECT batch_number, item_count, workshop_ids, started_at, finished_at,
|
|
elapsed_seconds, exit_code, status
|
|
FROM download_attempts
|
|
WHERE run_id = ?
|
|
ORDER BY batch_number
|
|
LIMIT ?
|
|
""",
|
|
(run_id, limit),
|
|
)
|
|
|
|
def current_mods(self) -> list[dict[str, Any]]:
|
|
return self.fetch_all(
|
|
"""
|
|
SELECT * FROM mods
|
|
WHERE workshop_id IN (SELECT DISTINCT workshop_id FROM collection_members)
|
|
ORDER BY CAST(workshop_id AS INTEGER)
|
|
"""
|
|
)
|
|
|
|
def mod_map(self, workshop_ids: Sequence[str]) -> dict[str, dict[str, Any]]:
|
|
if not workshop_ids:
|
|
return {}
|
|
placeholders = ",".join("?" for _ in workshop_ids)
|
|
rows = self.fetch_all(
|
|
f"SELECT * FROM mods WHERE workshop_id IN ({placeholders})", workshop_ids
|
|
)
|
|
return {row["workshop_id"]: row for row in rows}
|