251 lines
7.7 KiB
Python
251 lines
7.7 KiB
Python
import asyncio
|
|
import os
|
|
from contextlib import asynccontextmanager
|
|
from datetime import UTC, datetime
|
|
|
|
import uvicorn
|
|
from fastapi import Depends, FastAPI, HTTPException, Request
|
|
from pydantic import BaseModel, Field
|
|
from seadex import EntryNotFoundError, SeaDexEntry
|
|
|
|
from anilist_query import AniListError, AnimeNotFoundError, MissingAnimeName, get_anime_name
|
|
from db_module import (
|
|
close_connection_pool,
|
|
create_connection_pool,
|
|
get_db,
|
|
start_healthcheck_thread,
|
|
)
|
|
from metrics_server import REQUEST_COUNTER, metrics_api
|
|
from secret_manager import cleanup_secret_files
|
|
from send_notifications import NotificationClient
|
|
from simple_logger_handler import LOG_LEVEL, setup_logger
|
|
|
|
logger = setup_logger(__name__)
|
|
|
|
class AnimeCreate(BaseModel):
|
|
anilist_id: int = Field(gt=0)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
logger.info("[App] Starting application...")
|
|
app.state.notifier = NotificationClient()
|
|
logger.info("[DB] Creating MySQL connection pool...")
|
|
create_connection_pool("seadexPool")
|
|
|
|
start_healthcheck_thread()
|
|
logger.info("[DB] MySQL healthcheck thread started.")
|
|
|
|
SECRET_PATHS = frozenset({
|
|
"/etc/secrets/api_key",
|
|
"/etc/secrets/db_username",
|
|
"/etc/secrets/db_password"
|
|
})
|
|
cleanup_secret_files(SECRET_PATHS)
|
|
|
|
yield
|
|
logger.info("[App] Closing MySQL connection pool...")
|
|
close_connection_pool()
|
|
logger.info("[App] Shutdown complete.")
|
|
|
|
|
|
|
|
api = FastAPI(
|
|
title="Service SeaDex",
|
|
description="Queries SeaDex for new torrent information",
|
|
version="1.0.0",
|
|
lifespan=lifespan
|
|
)
|
|
|
|
@api.middleware("http")
|
|
async def prometheus_middleware(request: Request, call_next):
|
|
logger.debug(f"[Metrics] Incoming request: {request.method} {request.url.path}")
|
|
status = 500
|
|
try:
|
|
response = await call_next(request)
|
|
status = response.status_code
|
|
logger.debug(f"[Metrics] Request processed with status {status}")
|
|
except Exception:
|
|
logger.exception("[Metrics] Exception occurred:")
|
|
raise
|
|
finally:
|
|
REQUEST_COUNTER.labels(request.method, request.url.path, status).inc()
|
|
logger.debug(f"[Metrics] Counter incremented for {request.method} {request.url.path} [{status}]")
|
|
return response
|
|
|
|
|
|
def to_iso(dt: datetime) -> str:
|
|
return dt.astimezone(UTC).isoformat(timespec="microseconds")
|
|
|
|
|
|
def check_for_update(seadex: SeaDexEntry, anilist_id: int) -> str | None:
|
|
try:
|
|
return to_iso(seadex.from_id(anilist_id).updated_at)
|
|
except EntryNotFoundError:
|
|
return None
|
|
|
|
|
|
@api.post("/update")
|
|
def check_seadex(request:Request, db=Depends(get_db)):
|
|
results = {"checked": 0, "new": 0, "errors": 0}
|
|
|
|
with db.cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT id, anilist_id, anime_name, last_updated "
|
|
"FROM seadex_state WHERE active = 1"
|
|
)
|
|
anime = cursor.fetchall()
|
|
|
|
with SeaDexEntry() as seadex:
|
|
for row_id, anilist_id, anime_name, last_updated in anime:
|
|
results["checked"] += 1
|
|
try:
|
|
current = check_for_update(seadex, anilist_id)
|
|
|
|
if current is None:
|
|
logger.debug(f"[FastAPI] No entry yet for {anime_name} on SeaDex")
|
|
continue
|
|
|
|
if current != last_updated:
|
|
logger.info(
|
|
f"[FastAPI] {anime_name} updated, Anilist ID: {anilist_id}"
|
|
)
|
|
request.app.state.notifier.send(
|
|
title=anime_name,
|
|
body="SeaDex entry updated",
|
|
link=f"https://seadex.moe/{anilist_id}/",
|
|
)
|
|
|
|
with db.cursor() as cursor:
|
|
cursor.execute(
|
|
"UPDATE seadex_state SET last_updated = %s WHERE id = %s",
|
|
(current, row_id),
|
|
)
|
|
db.commit()
|
|
|
|
logger.debug(
|
|
f"[FastAPI] Notified and updated db for id={row_id}, "
|
|
f"Anilist ID={anilist_id}"
|
|
)
|
|
results["new"] += 1
|
|
|
|
except Exception:
|
|
db.rollback()
|
|
results["errors"] += 1
|
|
logger.exception(f"[FastAPI] Failed for id={row_id} Anilist={anilist_id}",)
|
|
|
|
if results["errors"]:
|
|
logger.warning(f"[FastAPI] SeaDex check finished with errors: {results}")
|
|
raise HTTPException(status_code=500, detail=results)
|
|
|
|
logger.info(f"[FastAPI] SeaDex check complete: {results}")
|
|
return results
|
|
|
|
@api.post("/add/anime")
|
|
def add_anime(anime: AnimeCreate, db=Depends(get_db)):
|
|
logger.info(f"[FastAPI] Received request to track anilist_id={anime.anilist_id}")
|
|
|
|
try:
|
|
anime_name = get_anime_name(anime.anilist_id)
|
|
except AnimeNotFoundError as e:
|
|
raise HTTPException(status_code=404, detail="Not found on AniList") from e
|
|
except MissingAnimeName as e:
|
|
raise HTTPException(status_code=404, detail="The anime exists but has no usable title") from e
|
|
except AniListError as e:
|
|
logger.exception("[FastAPI] AniList request failed")
|
|
raise HTTPException(status_code=502, detail="The AniList request failed") from e
|
|
logger.debug(f"[FastAPI] Found name={anime_name} for Anilist id={anime.anilist_id}")
|
|
|
|
with db.cursor() as cursor:
|
|
try:
|
|
cursor.execute(
|
|
"SELECT id FROM seadex_state WHERE anilist_id = %s",
|
|
(anime.anilist_id,),
|
|
)
|
|
|
|
if cursor.fetchone():
|
|
logger.debug("[FastAPI] Found existing entry, updating name and setting to active")
|
|
cursor.execute(
|
|
"UPDATE seadex_state SET anime_name = %s, active = 1 "
|
|
"WHERE anilist_id = %s",
|
|
(anime_name, anime.anilist_id),
|
|
)
|
|
db.commit()
|
|
return {
|
|
"status": "ok",
|
|
"detail": f"Updated existing entry name={anime_name}, "
|
|
f"Anilist ID={anime.anilist_id}",
|
|
}
|
|
|
|
with SeaDexEntry() as seadex:
|
|
current = check_for_update(seadex, anime.anilist_id)
|
|
logger.debug("[FastAPI] Inserting new anime with last_updated={current}")
|
|
|
|
cursor.execute(
|
|
"INSERT INTO seadex_state (anilist_id, anime_name, last_updated) "
|
|
"VALUES (%s, %s, %s)",
|
|
(anime.anilist_id, anime_name, current),
|
|
)
|
|
db.commit()
|
|
return {
|
|
"status": "ok",
|
|
"detail": f"Created new entry name={anime_name}, "
|
|
f"Anilist ID={anime.anilist_id}",
|
|
}
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.exception(f"[FastAPI] Failed for anilist_id={anime.anilist_id} name={anime_name}")
|
|
raise HTTPException(status_code=500) from e
|
|
|
|
@api.post("/delete/anime")
|
|
def remove_anime(anime: AnimeCreate, db=Depends(get_db)):
|
|
logger.info(f"[FastAPI] Received request to deactivate tracking for anilist_id={anime.anilist_id}")
|
|
with db.cursor() as cursor:
|
|
try:
|
|
cursor.execute(
|
|
"SELECT active FROM seadex_state WHERE anilist_id = %s",
|
|
(anime.anilist_id,),
|
|
)
|
|
row = cursor.fetchone()
|
|
if row and row[0]:
|
|
cursor.execute(
|
|
"UPDATE seadex_state SET active = 0 "
|
|
"WHERE anilist_id = %s",
|
|
(anime.anilist_id,),
|
|
)
|
|
db.commit()
|
|
logger.debug(f"[FastAPI] Successfully deactivated tracking for anilist_id={anime.anilist_id}")
|
|
return {
|
|
"status": "ok",
|
|
"detail": f"Deactivated entry {anime.anilist_id}",
|
|
}
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.exception(f"[FastAPI] Failed removeing tracking for id={anime.anilist_id}")
|
|
raise HTTPException(status_code=500) from e
|
|
|
|
if row is None:
|
|
logger.debug(f"[FastAPI] No tracked entry for anilist_id={anime.anilist_id}")
|
|
raise HTTPException(status_code=404, detail="No tracked entry for this AniList ID")
|
|
if not row[0]:
|
|
logger.debug(f"[FastAPI] Tracking was already deactivated for anilist_id={anime.anilist_id}")
|
|
return {"status": "ok", "detail": f"Entry {anime.anilist_id} was already inactive"}
|
|
|
|
|
|
|
|
async def start_servers():
|
|
logger.info("[Server] Starting main API (port 5000) and metrics server (port 9000)...")
|
|
logger.info(f"[Server] Build commit id={os.getenv("COMMIT_ID","unknown")}")
|
|
config_main = uvicorn.Config(api, host="0.0.0.0", port=5000, log_level=LOG_LEVEL.lower(),log_config=None)
|
|
config_metrics = uvicorn.Config(metrics_api, host="0.0.0.0", port=9000, log_level=LOG_LEVEL.lower(),log_config=None)
|
|
|
|
server_main = uvicorn.Server(config_main)
|
|
server_metrics = uvicorn.Server(config_metrics)
|
|
|
|
await asyncio.gather(server_main.serve(), server_metrics.serve())
|
|
logger.info("[Server] Both servers have stopped.")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(start_servers()) |