import asyncio import os from contextlib import asynccontextmanager from datetime import UTC, datetime import uvicorn from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request from seadex import EntryNotFoundError, SeaDexEntry from service_auth import UserId from anilist_query import get_anime_name from auth_config import Caller, auth from db_module import ( close_connection_pool, create_connection_pool, get_db, start_healthcheck_thread, ) from handlers import register_handlers from metrics_server import metrics_api from models import AnimeCreate, AnimeEntry, AnimeList, AnimeRemove from secret_manager import cleanup_secret_files from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError from simple_logger_handler import LOG_LEVEL, setup_logger logger = setup_logger(__name__) @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, dependencies=[Depends(auth.authenticate)], ) register_handlers(api) user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) 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 @scheduler_router.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() logger.debug("[FastAPI] Found %s active anime to check.", len(anime)) if not anime: logger.info("[FastAPI] No active Seadex entries") return results with SeaDexEntry() as seadex: for seadex_state_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("[FastAPI] No entry yet for Anime: %s on SeaDex", anime_name) continue if current == last_updated: continue logger.info( "[FastAPI] Anime: %s updated, Anilist ID: %s", anime_name,anilist_id, ) with db.cursor() as cursor: cursor.execute( "SELECT user_id, custom_name " "FROM seadex_state_subscriptions " "WHERE seadex_state_id = %s AND active = 1", (seadex_state_id,), ) subscribers = cursor.fetchall() sent = 0 failed = 0 for user_id, custom_name in subscribers: try: request.app.state.notifier.send( title=custom_name or anime_name, body="SeaDex entry updated", link=f"https://seadex.moe/{anilist_id}/", user_id=user_id, ) sent += 1 except NotificationUnavailableError: failed = len(subscribers) - sent logger.error("[FastAPI] Notifications backend unavailable, skipping " "%s remaining user(s) for Anime: %s", failed, anime_name, ) break except NotificationError as e: failed += 1 logger.error( "[FastAPI] Failed to send notification to user_id=%s for Anime: %s, status=%s", user_id, anime_name, getattr(e, 'status_code',None), ) if subscribers and sent == 0: results["errors"] += 1 logger.error("[FastAPI] All notifications failed for Anime: %s, will retry later", anime_name) continue with db.cursor() as cursor: cursor.execute( "UPDATE seadex_state SET last_updated = %s WHERE id = %s", (current, seadex_state_id), ) db.commit() logger.debug("[FastAPI] Notification sent and database updated for Anime: %s", anime_name) results["new"] += 1 except Exception: db.rollback() results["errors"] += 1 logger.exception( "[FastAPI] Failed for id: %s Anilist: %s", seadex_state_id, anilist_id, ) if results["errors"]: logger.warning("[FastAPI] SeaDex check finished with errors: %s", results) raise HTTPException(status_code=500, detail=results) logger.info("[FastAPI] SeaDex check complete: %s", results) return results @user_router.post("/anime/add") def add_anime( anime: AnimeCreate, user_id: UserId, db=Depends(get_db), ): logger.info("[FastAPI] Received request to track anilist_id=%s, user_id=%s", anime.anilist_id, user_id) with db.cursor() as cursor: cursor.execute( "SELECT id, active FROM seadex_state WHERE anilist_id = %s", (anime.anilist_id,), ) known = cursor.fetchone() last_updated = None with SeaDexEntry() as seadex: if known is None: anime_name = get_anime_name(anime.anilist_id) last_updated = check_for_update(seadex, anime.anilist_id) else: anime_name, active = known if not active: last_updated = check_for_update(seadex, anime.anilist_id) with db.cursor() as cursor: try: cursor.execute( "INSERT INTO seadex_state (anilist_id, anime_name, last_updated) " "VALUES (%s, %s, %s) " "ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1, " "last_updated = COALESCE(VALUES(last_updated), last_updated)", (anime.anilist_id, anime_name, last_updated), ) seadex_state_id = cursor.lastrowid cursor.execute( "INSERT INTO seadex_state_subscriptions (user_id, seadex_state_id, custom_name) " "VALUES (%s, %s, %s) " "ON DUPLICATE KEY UPDATE active = 1, custom_name = VALUES(custom_name)", (user_id, seadex_state_id, anime.custom_name or None), ) db.commit() except Exception as e: db.rollback() logger.exception( "[FastAPI] Failed for anilist_id=%s, name=%s, user_id=%s", anime.anilist_id, anime_name, user_id, ) raise HTTPException(status_code=500) from e logger.info( "[FastAPI] Tracking anilist_id=%s name=%r for user_id=%s (state_id=%s)", anime.anilist_id, anime_name, user_id, seadex_state_id, ) return {"status": "ok", "detail": f"Tracking name={anime.custom_name or anime_name}, Anilist ID={anime.anilist_id}"} @user_router.post("/anime/delete") def remove_anime( anime: AnimeRemove, user_id: UserId, db=Depends(get_db) ): logger.info( "[FastAPI] Received request to deactivate anilist_id=%s for user_id=%s", anime.anilist_id, user_id, ) with db.cursor() as cursor: try: cursor.execute( "SELECT id FROM seadex_state " "WHERE anilist_id = %s FOR UPDATE", (anime.anilist_id,), ) known = cursor.fetchone() if known is None: db.rollback() logger.info("[FastAPI] Anime anilist_id=%s is not tracked, nothing to do", anime.anilist_id) return {"status": "ok", "detail": f"Stopped tracking Anilist ID={anime.anilist_id}"} seadex_state_id = known[0] cursor.execute( "UPDATE seadex_state_subscriptions SET active = 0 " "WHERE user_id = %s AND seadex_state_id = %s", (user_id, seadex_state_id), ) cursor.execute( "SELECT 1 from seadex_state_subscriptions " "WHERE seadex_state_id = %s AND active = 1 LIMIT 1", (seadex_state_id,), ) if cursor.fetchone() is None: cursor.execute( "UPDATE seadex_state SET active = 0 WHERE id = %s", (seadex_state_id,), ) logger.info("[FastAPI] No active subscribers left, deactivated state_id=%s", seadex_state_id) db.commit() except Exception as e: db.rollback() logger.exception( "[FastAPI] Failed to unsubscribe anime anilist_id=%s for user_id=%s", anime.anilist_id, user_id, ) raise HTTPException(status_code=500) from e logger.info( "[FastAPI] Successfully deactivated tracking for anilist_id=%s for user_id=%s", anime.anilist_id, user_id, ) return {"status": "ok", "detail": f"Stopped tracking Anilist ID={anime.anilist_id}"} @user_router.get("/anime/list", response_model=AnimeList) def list_active_and_inactive_anime( user_id: UserId, db=Depends(get_db), ): logger.info("[FastAPI] Received request to list anime for User ID = %s", user_id) try: with db.cursor() as cursor: cursor.execute( "SELECT s.custom_name AS custom_name, st.anime_name AS anime_name, " "st.anilist_id, s.active, st.last_updated from seadex_state_subscriptions AS s " "JOIN seadex_state AS st on s.seadex_state_id = st.id " "WHERE s.user_id = %s ORDER BY st.anilist_id", (user_id,), ) rows = cursor.fetchall() anime_list=AnimeList() for custom_name, anime_name, anilist_id, is_active, last_updated in rows: entry = AnimeEntry( name=custom_name or anime_name, anilist_id=anilist_id, last_updated=None if last_updated is None else last_updated, ) (anime_list.active if is_active else anime_list.inactive).append(entry) except Exception as e: logger.exception("[FastAPI] Failed to list anime") raise HTTPException(status_code=500) from e return anime_list api.include_router(user_router) api.include_router(scheduler_router) async def start_servers(): logger.info("[Server] Starting main API (port 5000) and metrics server (port 9000)...") logger.info("[Server] Build commit id=%s", os.getenv("COMMIT_ID", "unknown")) auth.load_keys(api, Caller.GATEWAY, Caller.SCHEDULER) cleanup_secret_files(auth.key_paths(Caller.GATEWAY, Caller.SCHEDULER)) 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())