288 lines
9.8 KiB
Python
288 lines
9.8 KiB
Python
from __future__ import annotations
|
|
|
|
import secrets
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
from typing import Annotated
|
|
from urllib.parse import urlparse
|
|
|
|
import requests
|
|
from fastapi import (
|
|
FastAPI,
|
|
File,
|
|
Header,
|
|
HTTPException,
|
|
Query,
|
|
UploadFile,
|
|
status,
|
|
)
|
|
from fastapi.responses import FileResponse, RedirectResponse, Response
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from .config import Settings
|
|
from .database import Database
|
|
from .modlist import core_items, generate_rml, parse_active_mods, workshop_items
|
|
from .refresh import RefreshCoordinator, RefreshService
|
|
from .workshop import SteamCmdRunner
|
|
|
|
|
|
DESKTOP_RELEASES_API = (
|
|
"https://git.hudsonriggs.systems/api/v1/repos/HRiggs/ProgressionMods/releases"
|
|
)
|
|
DESKTOP_RELEASE_HOST = "git.hudsonriggs.systems"
|
|
|
|
|
|
def latest_windows_release_url() -> str | None:
|
|
"""Return the newest Windows executable, including prerelease builds."""
|
|
response = requests.get(
|
|
DESKTOP_RELEASES_API,
|
|
headers={"Accept": "application/json"},
|
|
timeout=10,
|
|
)
|
|
response.raise_for_status()
|
|
releases = response.json()
|
|
if not isinstance(releases, list):
|
|
raise ValueError("Unexpected Gitea releases response")
|
|
|
|
published = sorted(
|
|
(release for release in releases if not release.get("draft")),
|
|
key=lambda release: release.get("published_at") or release.get("created_at") or "",
|
|
reverse=True,
|
|
)
|
|
stable = [release for release in published if not release.get("prerelease")]
|
|
candidates = stable or published
|
|
for release in candidates:
|
|
assets = release.get("assets") or []
|
|
executables = [
|
|
asset
|
|
for asset in assets
|
|
if str(asset.get("name") or "").lower().endswith(".exe")
|
|
]
|
|
executables.sort(
|
|
key=lambda asset: (
|
|
str(asset.get("name") or "").lower() != "progressionloader.exe",
|
|
str(asset.get("name") or "").lower(),
|
|
)
|
|
)
|
|
for asset in executables:
|
|
download_url = str(asset.get("browser_download_url") or "")
|
|
parsed = urlparse(download_url)
|
|
if parsed.scheme == "https" and parsed.netloc == DESKTOP_RELEASE_HOST:
|
|
return download_url
|
|
return None
|
|
|
|
|
|
def create_app(settings: Settings | None = None) -> FastAPI:
|
|
settings = settings or Settings.load()
|
|
database = Database(settings.database_path)
|
|
database.initialize()
|
|
runner = SteamCmdRunner(
|
|
settings.steamcmd_path,
|
|
settings.workshop_content_dir,
|
|
settings.steamcmd_timeout_seconds,
|
|
)
|
|
service = RefreshService(settings, database, runner=runner)
|
|
coordinator = RefreshCoordinator(
|
|
service,
|
|
settings.refresh_interval_seconds,
|
|
settings.refresh_on_startup,
|
|
settings.initial_download_timeout_seconds,
|
|
)
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_: FastAPI):
|
|
await coordinator.start()
|
|
yield
|
|
await coordinator.stop()
|
|
|
|
app = FastAPI(
|
|
title="Progression Mods Backend",
|
|
version="0.1.0",
|
|
lifespan=lifespan,
|
|
docs_url="/api/docs",
|
|
redoc_url=None,
|
|
)
|
|
app.state.settings = settings
|
|
app.state.database = database
|
|
app.state.coordinator = coordinator
|
|
app.state.runner = runner
|
|
|
|
@app.get("/api/health")
|
|
async def health() -> dict:
|
|
return {
|
|
"status": "ok",
|
|
"steamcmd_available": runner.available(),
|
|
"workshop_content_dir": str(settings.workshop_content_dir),
|
|
"scheduler": coordinator.state(),
|
|
}
|
|
|
|
@app.get("/api/status")
|
|
async def dashboard_status() -> dict:
|
|
return {
|
|
**database.dashboard(),
|
|
"scheduler": coordinator.state(),
|
|
"steamcmd_available": runner.available(),
|
|
"game_version": settings.game_version,
|
|
}
|
|
|
|
@app.get("/api/analytics")
|
|
async def analytics(limit: int = Query(24, ge=1, le=168)) -> dict:
|
|
return {"runs": database.runs(limit)}
|
|
|
|
@app.get("/api/analytics/downloads")
|
|
async def download_analytics(
|
|
run_id: str | None = None,
|
|
limit: int = Query(100, ge=1, le=500),
|
|
) -> dict:
|
|
if not run_id:
|
|
runs = database.runs(1)
|
|
run_id = runs[0]["id"] if runs else None
|
|
return {
|
|
"run_id": run_id,
|
|
"attempts": database.download_attempts(run_id, limit) if run_id else [],
|
|
}
|
|
|
|
@app.get("/api/mods")
|
|
async def mods(
|
|
offset: int = Query(0, ge=0),
|
|
limit: int = Query(100, ge=1, le=500),
|
|
state: str | None = Query(None, pattern="^(ready|pending|failed)$"),
|
|
) -> dict:
|
|
clauses = [
|
|
"workshop_id IN (SELECT DISTINCT workshop_id FROM collection_members)"
|
|
]
|
|
parameters: list[object] = []
|
|
if state:
|
|
clauses.append("status = ?")
|
|
parameters.append(state)
|
|
where = " AND ".join(clauses)
|
|
total = database.fetch_one(
|
|
f"SELECT COUNT(*) AS count FROM mods WHERE {where}", parameters
|
|
) or {"count": 0}
|
|
rows = database.fetch_all(
|
|
f"""
|
|
SELECT workshop_id, title, package_id, display_name, time_updated,
|
|
metadata_file_size, disk_size, status, error, last_checked_at,
|
|
last_downloaded_at
|
|
FROM mods WHERE {where}
|
|
ORDER BY CAST(workshop_id AS INTEGER)
|
|
LIMIT ? OFFSET ?
|
|
""",
|
|
[*parameters, limit, offset],
|
|
)
|
|
return {
|
|
"total": total["count"],
|
|
"offset": offset,
|
|
"limit": limit,
|
|
"items": rows,
|
|
}
|
|
|
|
@app.post("/api/refresh", status_code=status.HTTP_202_ACCEPTED)
|
|
async def refresh(authorization: str | None = Header(None)) -> dict:
|
|
if settings.admin_token:
|
|
expected = f"Bearer {settings.admin_token}"
|
|
if not authorization or not secrets.compare_digest(authorization, expected):
|
|
raise HTTPException(status_code=401, detail="Invalid admin token")
|
|
run_id, started = await coordinator.trigger("manual")
|
|
return {"run_id": run_id, "started": started}
|
|
|
|
def ready_workshop_rows(allow_partial: bool) -> list[dict]:
|
|
rows = database.current_mods()
|
|
failed = [row for row in rows if row.get("status") != "ready"]
|
|
if failed and not allow_partial:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail={
|
|
"message": "The mod list is incomplete",
|
|
"not_ready": len(failed),
|
|
"hint": "Run a refresh or request allow_partial=true",
|
|
},
|
|
)
|
|
if not rows:
|
|
raise HTTPException(
|
|
status_code=409, detail="No collection data is available yet"
|
|
)
|
|
return rows
|
|
|
|
@app.get("/api/modlists/vanilla")
|
|
async def vanilla_modlist(
|
|
dlc: Annotated[list[str] | None, Query()] = None,
|
|
allow_partial: Annotated[bool, Query()] = False,
|
|
) -> Response:
|
|
rows = ready_workshop_rows(allow_partial)
|
|
rml = generate_rml(
|
|
[*core_items(dlc or []), *workshop_items(rows)], settings.game_version
|
|
)
|
|
return Response(
|
|
rml,
|
|
media_type="application/xml",
|
|
headers={
|
|
"Content-Disposition": 'attachment; filename="ProgressionVanilla.rml"'
|
|
},
|
|
)
|
|
|
|
@app.get("/api/downloads/windows", include_in_schema=False)
|
|
def latest_windows_download() -> RedirectResponse:
|
|
try:
|
|
download_url = latest_windows_release_url()
|
|
except (requests.RequestException, TypeError, ValueError) as error:
|
|
raise HTTPException(
|
|
status_code=502,
|
|
detail="The Windows download could not be resolved from Gitea",
|
|
) from error
|
|
if not download_url:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="No Windows executable is attached to a published release",
|
|
)
|
|
return RedirectResponse(download_url, status_code=307)
|
|
|
|
@app.post("/api/modlists/homebrew")
|
|
async def homebrew_modlist(
|
|
mods_config: Annotated[UploadFile, File()],
|
|
allow_partial: Annotated[bool, Query()] = False,
|
|
) -> Response:
|
|
if not mods_config.filename or not mods_config.filename.lower().endswith(
|
|
".xml"
|
|
):
|
|
raise HTTPException(status_code=400, detail="Upload ModsConfig.xml")
|
|
content = await mods_config.read(5 * 1024 * 1024 + 1)
|
|
try:
|
|
current = parse_active_mods(content)
|
|
except ValueError as error:
|
|
raise HTTPException(status_code=400, detail=str(error)) from error
|
|
rows = ready_workshop_rows(allow_partial)
|
|
by_package = {
|
|
str(row.get("package_id") or "").lower(): row
|
|
for row in rows
|
|
if row.get("package_id")
|
|
}
|
|
hydrated = []
|
|
for item in current:
|
|
match = by_package.get(item.package_id.lower())
|
|
if match:
|
|
hydrated.extend(workshop_items([match]))
|
|
else:
|
|
hydrated.append(item)
|
|
rml = generate_rml([*hydrated, *workshop_items(rows)], settings.game_version)
|
|
return Response(
|
|
rml,
|
|
media_type="application/xml",
|
|
headers={
|
|
"Content-Disposition": 'attachment; filename="ProgressionHomebrew.rml"'
|
|
},
|
|
)
|
|
|
|
@app.get("/", include_in_schema=False)
|
|
async def index() -> FileResponse:
|
|
return FileResponse(settings.static_dir / "index.html")
|
|
|
|
app.mount("/static", StaticFiles(directory=settings.static_dir), name="static")
|
|
art_dir = Path(settings.static_dir).parents[1] / "art"
|
|
if art_dir.is_dir():
|
|
app.mount("/art", StaticFiles(directory=art_dir), name="art")
|
|
return app
|
|
|
|
|
|
app = create_app()
|