Initial draft
This commit is contained in:
1 parent
648dcfb446
commit
c5370642c5
6 files changed
+285
No files matched your search
@@ -0,0 +1,28 @@
|
||||
[tool.ruff]
|
||||
line-length = 120
|
||||
target-version = "py312"
|
||||
[tool.ruff.lint]
|
||||
select = [
|
||||
"E", "F", # pycodestyle errors, pyflakes (unused imports/variables, undefined names)
|
||||
"I", # import sorting (replaces isort)
|
||||
"B", # bugbear: likely bugs
|
||||
"ASYNC", # async mistakes
|
||||
"RUF006", # un-referenced asyncio.create_task (tasks can be garbage-collected)
|
||||
]
|
||||
ignore = ["E501"]
|
||||
|
||||
[tool.ruff.lint.isort]
|
||||
known-first-party = ["db_module", "metrics_server", "rabbitmq_handler", "secret_manager",
|
||||
"simple_logger_handler", "uvicorn_logger_config", "validator"]
|
||||
|
||||
[tool.ruff.format]
|
||||
indent-style = "tab"
|
||||
|
||||
[tool.ruff.lint.flake8-bugbear]
|
||||
extend-immutable-calls = [
|
||||
"fastapi.Depends",
|
||||
"fastapi.Query",
|
||||
"fastapi.Header",
|
||||
"fastapi.Body",
|
||||
"fastapi.Security",
|
||||
]
|
||||
@@ -0,0 +1,2 @@
|
||||
-r requirements.txt
|
||||
ruff==0.17.1
|
||||
@@ -0,0 +1,31 @@
|
||||
--extra-index-url https://git.gansejunge.com/api/packages/notifier/pypi/simple/
|
||||
annotated-doc==0.0.5
|
||||
annotated-types==0.8.0
|
||||
anyio==4.15.1
|
||||
atomicwriter==0.3.0
|
||||
certifi==2026.7.22
|
||||
charset-normalizer==3.5.2
|
||||
click==8.5.0
|
||||
fastapi==0.143.2
|
||||
h11==0.16.0
|
||||
httpcore==1.0.9
|
||||
httpx==0.28.1
|
||||
idna==3.20
|
||||
lib-db-module==0.2.0
|
||||
lib-secret-manager==0.2.0
|
||||
lib-uvicorn-config==0.2.0
|
||||
msgspec==0.22.0
|
||||
mysql-connector-python==9.4.0
|
||||
opentelemetry-api==1.45.1
|
||||
prometheus_client==0.26.0
|
||||
pydantic==2.14.1
|
||||
pydantic_core==2.50.1
|
||||
python-json-logger==4.2.0
|
||||
requests==2.34.2
|
||||
seadex==0.7.3
|
||||
simple-logger-handler==0.1.0
|
||||
starlette==1.8.0
|
||||
typing-inspection==0.4.4
|
||||
typing_extensions==4.16.0
|
||||
urllib3==2.8.0
|
||||
uvicorn==0.54.0
|
||||
+147
@@ -0,0 +1,147 @@
|
||||
import asyncio
|
||||
from contextlib import asynccontextmanager
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import uvicorn
|
||||
from fastapi import Depends, FastAPI, HTTPException, Request
|
||||
from seadex import EntryNotFoundError, SeaDexEntry
|
||||
|
||||
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_notification import send_notification
|
||||
from simple_logger_handler import LOG_LEVEL, setup_logger
|
||||
from uvicorn_logger_config import LOGGING_CONFIG
|
||||
|
||||
logger = setup_logger(__name__)
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
logger.info("[App] Starting application...")
|
||||
|
||||
logger.info("[DB] Creating MySQL connection pool...")
|
||||
create_connection_pool("royalRoadPool")
|
||||
|
||||
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(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}"
|
||||
)
|
||||
send_notification(anime_name, 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
|
||||
|
||||
async def start_servers():
|
||||
logger.info("[Server] Starting main API (port 5000) and metrics server (port 9000)...")
|
||||
config_main = uvicorn.Config(api, host="0.0.0.0", port=5000, log_level=LOG_LEVEL.lower(), log_config=LOGGING_CONFIG)
|
||||
config_metrics = uvicorn.Config(metrics_api, host="0.0.0.0", port=9000, log_level=LOG_LEVEL.lower(), log_config=LOGGING_CONFIG)
|
||||
|
||||
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())
|
||||
@@ -0,0 +1,10 @@
|
||||
from fastapi import FastAPI, Response
|
||||
from prometheus_client import CONTENT_TYPE_LATEST, Counter, generate_latest
|
||||
|
||||
metrics_api = FastAPI(title="Metrics Server", description="Prometheus metrics endpoint")
|
||||
|
||||
REQUEST_COUNTER = Counter("http_requests_total", "Total HTTP Requests", ["method", "endpoint", "status"])
|
||||
|
||||
@metrics_api.get("/metrics")
|
||||
async def metrics():
|
||||
return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
|
||||
@@ -0,0 +1,67 @@
|
||||
import os
|
||||
import time
|
||||
|
||||
import requests
|
||||
from fastapi import HTTPException
|
||||
from requests.exceptions import ConnectionError, HTTPError, RequestException, Timeout
|
||||
|
||||
from secret_manager import return_credentials
|
||||
from simple_logger_handler import setup_logger
|
||||
|
||||
backend_api_url=os.getenv("BACKEND_API_URL","localhost:8101/internal/receive-notifications")
|
||||
api_key= return_credentials("/etc/secrets/api_key")
|
||||
logger = setup_logger(__name__)
|
||||
|
||||
|
||||
def send_notification(anime_name: str,link: str,max_retries: int = 5,timeout: int = 5):
|
||||
"""
|
||||
Sends a notification to the internal backend service when a new Royalroad chapter is released.
|
||||
|
||||
Parameters:
|
||||
anime_name: Name of the anime.
|
||||
link: Direct link to the SeaDex page.
|
||||
"""
|
||||
|
||||
headers = {
|
||||
"X-API-Key-Internal": api_key,
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
data = {
|
||||
"user_id": 1,
|
||||
"message": {
|
||||
"title": "Updated SeaDex entry",
|
||||
"body": anime_name,
|
||||
"link": link,
|
||||
"category":"utility",
|
||||
"timestamp": int(time.time())
|
||||
}
|
||||
}
|
||||
|
||||
logger.debug(f"[Notify] Preparing to send notification: title='{anime_name}', link='{link}'")
|
||||
|
||||
with requests.Session() as session:
|
||||
for attempt in range(1, max_retries + 1):
|
||||
try:
|
||||
logger.debug(f"[Notify] Sending request to backend (attempt {attempt}/{max_retries})")
|
||||
response = session.post(backend_api_url, headers=headers, json=data, timeout=timeout)
|
||||
response.raise_for_status()
|
||||
logger.info(f"[Notify] Notification sent successfully for '{anime_name}')")
|
||||
return response.text
|
||||
|
||||
except (Timeout, ConnectionError) as e:
|
||||
logger.warning(f"[Notify] Attempt {attempt}/{max_retries} failed: {type(e).__name__}")
|
||||
if attempt == max_retries:
|
||||
logger.error(f"[Notify] All retry attempts failed for '{anime_name}'")
|
||||
raise HTTPException(status_code=503, detail=f"Notification service unavailable: {type(e).__name__}") from e
|
||||
sleep_time = 2 ** (attempt - 1)
|
||||
logger.debug(f"[Notify] Retrying in {sleep_time} seconds...")
|
||||
time.sleep(sleep_time)
|
||||
|
||||
except HTTPError as e:
|
||||
logger.error(f"[Notify] HTTP {e.response.status_code}: {e.response.text}")
|
||||
raise HTTPException(status_code=e.response.status_code, detail=e.response.text) from e
|
||||
|
||||
except RequestException as e:
|
||||
logger.error(f"[Notify] Unexpected request failure: {e}")
|
||||
raise HTTPException(status_code=500, detail=f"Request failed: {e!s}") from e
|
||||
Reference in new issue
Block a user