dev
eadc242173
Docker / docker (push) Successful in 2m50s
FastAPI backend + HTML/JS frontend pour parcourir et télécharger les concerts Arte Concert. Cache 6h, recherche live, historique SQLite, suivi de progression SSE, design sombre Playfair Display + Inter. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
77 lines
2.1 KiB
Python
77 lines
2.1 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
|
|
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, StreamingResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from pydantic import BaseModel
|
|
|
|
from arte_api import fetch_concerts, invalidate_cache
|
|
from downloader import DownloadManager
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
app = FastAPI(title="Arte-dl")
|
|
app.mount("/static", StaticFiles(directory="static"), name="static")
|
|
templates = Jinja2Templates(directory="templates")
|
|
|
|
dm = DownloadManager()
|
|
|
|
|
|
# ------------------------------------------------------------------ pages
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def index(request: Request):
|
|
return templates.TemplateResponse("index.html", {"request": request})
|
|
|
|
|
|
# ------------------------------------------------------------------ API: concerts
|
|
|
|
|
|
@app.get("/api/concerts")
|
|
async def api_concerts(page: int = 1, search: str = "", page_size: int = 24):
|
|
return await fetch_concerts(page=page, search=search, page_size=page_size)
|
|
|
|
|
|
@app.post("/api/refresh")
|
|
async def api_refresh():
|
|
count = await invalidate_cache()
|
|
return {"count": count}
|
|
|
|
|
|
# ------------------------------------------------------------------ API: downloads
|
|
|
|
|
|
class DownloadRequest(BaseModel):
|
|
url: str
|
|
title: str
|
|
|
|
|
|
@app.post("/api/download")
|
|
async def api_download(req: DownloadRequest, bg: BackgroundTasks):
|
|
if not req.url:
|
|
raise HTTPException(status_code=400, detail="url required")
|
|
dl_id = dm.enqueue(req.url, req.title, bg)
|
|
return {"id": dl_id}
|
|
|
|
|
|
@app.get("/api/downloads")
|
|
async def api_downloads():
|
|
return dm.history()
|
|
|
|
|
|
@app.get("/api/progress/{dl_id}")
|
|
async def api_progress(dl_id: str):
|
|
async def stream():
|
|
while True:
|
|
s = dm.status(dl_id)
|
|
yield f"data: {json.dumps(s)}\n\n"
|
|
if s.get("state") in ("done", "error", "unknown"):
|
|
break
|
|
await asyncio.sleep(0.8)
|
|
|
|
return StreamingResponse(stream(), media_type="text/event-stream")
|