main.py - dev - vaskaz/flexin_flow

main.py
264 строки9 KB
vaskaz
newborn
10 июн 2026, 00:16
10 июн 2026, 00:16dc9da65
"""Retro Library — local web service for browsing game ROMs."""

import os
import sys
import json
import threading
import time
from pathlib import Path

from fastapi import FastAPI, Query, HTTPException, BackgroundTasks
from fastapi.responses import HTMLResponse, FileResponse, JSONResponse, StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
import uvicorn

from database import get_all_roms, get_rom_by_id, get_platforms, search_roms_by_filters, get_stats, init_db
from metadata import format_size
from metadata_fetcher import enrich_game, enrich_all_games

app = FastAPI(title="Retro Library", version="1.0.0")

# CORS — allow any origin for local use
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Resolve base directory
BASE_DIR = Path(__file__).parent.resolve()

# Create static and covers directories
STATIC_DIR = BASE_DIR / "static"
COVERS_DIR = BASE_DIR / "covers_cache"

STATIC_DIR.mkdir(exist_ok=True)
COVERS_DIR.mkdir(exist_ok=True)

# Mount static files (for frontend)
if STATIC_DIR.exists():
    app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
if COVERS_DIR.exists():
    app.mount("/covers", StaticFiles(directory=str(COVERS_DIR)), name="covers")

# ── Configuration ─────────────────────────────────────────────────────

SCAN_ROOT = os.environ.get("RETRO_LIBRARY_PATH", "")
SCAN_INTERVAL = int(os.environ.get("RETRO_LIBRARY_SCAN_INTERVAL", "0"))  # auto-rescan hours (0=manual)

scan_lock = threading.Lock()
last_scan_result = {"status": "idle", "stats": {}, "timestamp": 0}


# ── API Endpoints ─────────────────────────────────────────────────────

@app.on_event("startup")
def startup():
    """Initialize database on startup."""
    init_db()
    # If a scan path is configured, auto-scan on boot
    if SCAN_ROOT and os.path.isdir(SCAN_ROOT):
        threading.Thread(target=run_scan, daemon=True).start()


@app.get("/api/stats")
def api_stats():
    """Get library statistics."""
    return get_stats()


@app.get("/api/platforms")
def api_platforms():
    """Get all platforms with game counts."""
    return get_platforms()


@app.get("/api/roms")
def api_roms(
    platform: str = Query("", description="Filter by platform ID"),
    search: str = Query("", description="Search query"),
    page: int = Query(1, ge=1),
    per_page: int = Query(100, ge=1, le=500),
):
    """Get all ROMs with optional filtering and pagination."""
    all_roms = search_roms_by_filters(platform=platform, search=search)
    total = len(all_roms)
    start = (page - 1) * per_page
    end = start + per_page
    page_roms = all_roms[start:end]

    return {
        "total": total,
        "page": page,
        "per_page": per_page,
        "total_pages": max(1, (total + per_page - 1) // per_page),
        "results": page_roms,
    }


@app.get("/api/roms/{rom_id}")
def api_rom_detail(rom_id: int):
    """Get a single ROM with full details."""
    rom = get_rom_by_id(rom_id)
    if not rom:
        raise HTTPException(status_code=404, detail="ROM not found")
    rom["size_str"] = format_size(rom["size_bytes"])
    rom["cover_exists"] = bool(rom["cover_url"]) and (COVERS_DIR / rom["cover_url"]).exists()
    return rom


@app.get("/api/download/{rom_id}")
def api_download(rom_id: int):
    """Download a ROM file."""
    rom = get_rom_by_id(rom_id)
    if not rom:
        raise HTTPException(status_code=404, detail="ROM not found")
    rom_path = rom["path"]
    if not os.path.isfile(rom_path):
        raise HTTPException(status_code=404, detail="File no longer exists on disk")
    return FileResponse(
        path=rom_path,
        filename=rom["filename"],
        media_type="application/octet-stream",
        headers={
            "Content-Disposition": f'attachment; filename="{rom["filename"]}"',
        },
    )


@app.post("/api/scan")
def api_scan():
    """Trigger a scan of the configured directory."""
    global last_scan_result
    if not SCAN_ROOT:
        raise HTTPException(status_code=400, detail="RETRO_LIBRARY_PATH not configured. Set env var or restart.")
    if not os.path.isdir(SCAN_ROOT):
        raise HTTPException(status_code=400, detail=f"Scan path does not exist: {SCAN_ROOT}")

    scan_lock.acquire()
    try:
        run_scan()
        return last_scan_result
    finally:
        scan_lock.release()


@app.get("/api/scan/status")
def api_scan_status():
    """Get last scan result."""
    return last_scan_result


# ── Enrichment endpoints ────────────────────────────────────────────────

enrich_lock = threading.Lock()
last_enrich_result = {"status": "idle", "stats": {}, "timestamp": 0}


@app.post("/api/enrich")
def api_enrich_all():
    """Enrich all games without metadata from Wikipedia."""
    global last_enrich_result

    if not enrich_lock.acquire(blocking=False):
        raise HTTPException(status_code=429, detail="Enrichment already in progress")

    try:
        last_enrich_result["status"] = "running"
        last_enrich_result["timestamp"] = time.time()

        stats = enrich_all_games()

        last_enrich_result = {
            "status": "done",
            "stats": stats,
            "timestamp": time.time(),
        }
        return last_enrich_result
    except Exception as e:
        last_enrich_result = {
            "status": "error",
            "error": str(e),
            "timestamp": time.time(),
        }
        raise HTTPException(status_code=500, detail=str(e))
    finally:
        enrich_lock.release()


@app.post("/api/enrich/{rom_id}")
def api_enrich_one(rom_id: int):
    """Enrich a single ROM with metadata."""
    result = enrich_game(rom_id)
    if result.get("error"):
        raise HTTPException(status_code=404, detail=result["error"])
    return result


@app.get("/api/enrich/status")
def api_enrich_status():
    """Get enrichment progress."""
    return last_enrich_result


# ── Frontend ──────────────────────────────────────────────────────────

@app.get("/", response_class=HTMLResponse)
def index():
    """Serve the SPA."""
    index_path = STATIC_DIR / "index.html"
    if index_path.exists():
        return HTMLResponse(index_path.read_text(encoding="utf-8"))
    return HTMLResponse("<h1>Retro Library</h1><p>Frontend not found. Create static/index.html</p>")


# ── Scan worker ───────────────────────────────────────────────────────

def run_scan():
    """Run scan and update global state."""
    global last_scan_result
    from scanner import scan_and_update

    last_scan_result["status"] = "scanning"
    last_scan_result["timestamp"] = time.time()

    try:
        stats = scan_and_update(SCAN_ROOT)
        last_scan_result = {
            "status": "done",
            "stats": stats,
            "timestamp": time.time(),
        }
    except Exception as e:
        last_scan_result = {
            "status": "error",
            "error": str(e),
            "timestamp": time.time(),
        }


# ── Entry point ───────────────────────────────────────────────────────

def main():
    port = int(os.environ.get("RETRO_LIBRARY_PORT", "8080"))
    host = os.environ.get("RETRO_LIBRARY_HOST", "0.0.0.0")

    print(f"╔══════════════════════════════════════════════╗")
    print(f"║         🎮 Retro Library Server             ║")
    print(f"╠══════════════════════════════════════════════╣")
    print(f"║  URL:      http://localhost:{port}              ║")
    if SCAN_ROOT:
        print(f"║  Scan:     {SCAN_ROOT}")
    else:
        print(f"║  Scan:     NOT SET (set RETRO_LIBRARY_PATH)")
    print(f"║  Library:  {BASE_DIR / 'library.db'}")
    print(f"╚══════════════════════════════════════════════╝")

    uvicorn.run(app, host=host, port=port, log_level="info")


if __name__ == "__main__":
    main()