notifier/service-royalroad-chapters
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Feauture/unified path #8

Merged
florian merged 4 commits from feauture/unified-path into main 2026-11-03 12:06:21 +00:00
Conversation 0 Commits 4 Files Changed 5
+39 -143
1 changed files with 39 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 1b451620a7 - Show all commits
src/main.py
+39 -143
View File
@@ -1,6 +1,7 @@
import asyncio import asyncio
import os import os
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import assert_never
import uvicorn import uvicorn
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request
@@ -13,12 +14,10 @@ from feed_handler import (
build_gitea_runner, build_gitea_runner,
ensure_feed_exists, ensure_feed_exists,
fetch_latest, fetch_latest,
grab_gitea_release_information,
grab_royal_road_chapter_information,
) )
from handlers import register_handlers from handlers import register_handlers
from metrics_server import metrics_api from metrics_server import metrics_api
from models import FeedEntry from models import FeedEntry, FeedEntryType
from secret_manager import cleanup_secret_files, get_secret from secret_manager import cleanup_secret_files, get_secret
from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError
from service_auth import UserId from service_auth import UserId
@@ -63,44 +62,30 @@ register_handlers(api)
user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))])
scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))])
@scheduler_router.post("/gitea") @scheduler_router.post("/update/{rss_feed_type}")
def get_release_information(request: Request, db = Depends(get_db)): def update_rss_feeds(
""" request: Request,
Checks for new Gitea releases and updates the database accordingly. rss_feed_type: FeedEntryType,
db=Depends(get_db),
Behaviour: ):
- Fetches all active Gitea feeds and their last known release link.
- Queries each feed for its latest release.
- Updates `last_entry` if the latest release link differs from the stored one.
- Sends a notification for each changed release.
- A failure on one feed is logged and the remaining feeds are still checked.
Returns:
A JSON object with checked/new/error counts. If any feed failed,
a 500 is returned with the same counts as the detail.
"""
logger.info("[FastAPI] Checking for new Gitea releases")
results = {"checked": 0, "new": 0, "errors": 0} results = {"checked": 0, "new": 0, "errors": 0}
with db.cursor() as cursor: with db.cursor() as cursor:
cursor.execute( cursor.execute(
"SELECT id, rss_feed, last_entry " "SELECT id, rss_feed, last_entry "
"FROM rss_feeds WHERE active = 1 AND type = %s", "FROM rss_feeds WHERE type = %s AND active = 1",
("gitea",), (rss_feed_type.value,),
) )
feeds = cursor.fetchall() rss_feeds = cursor.fetchall()
logger.debug(f"[FastAPI] Found {len(feeds)} active Gitea feeds to check.") logger.debug(f"[FastAPI] Found {len(rss_feeds)} active {rss_feed_type.value} RSS feeds to check.")
for feed_id, rss_feed, last_entry in feeds: for feed_id, rss_feed, last_entry in rss_feeds:
results["checked"] += 1 results["checked"] += 1
try: try:
gitea_release_information = grab_gitea_release_information(rss_feed) online_information = fetch_latest(rss_feed_type, rss_feed)
logger.debug(f"[FastAPI] ID:{feed_id},RSS Feed {rss_feed}: db={last_entry}, online={gitea_release_information.entry_id}") logger.debug(f"[FastAPI] ID:{feed_id},RSS Feed {rss_feed}: db={last_entry}, online={online_information.entry_id}")
if online_information.entry_id == last_entry:
if gitea_release_information.entry_id == last_entry:
continue continue
logger.info(f"[FastAPI] Release change for ID {feed_id}: {gitea_release_information.body_title}")
with db.cursor() as cursor: with db.cursor() as cursor:
cursor.execute( cursor.execute(
"SELECT user_id, feed_name " "SELECT user_id, feed_name "
@@ -112,25 +97,31 @@ def get_release_information(request: Request, db = Depends(get_db)):
sent = 0 sent = 0
failed = 0 failed = 0
repo = gitea_release_information.link.split("/releases")[0].removeprefix("https://") match rss_feed_type:
for user_id,feed_name in subscribers: case FeedEntryType.GITEA:
title = online_information.link.split("/releases")[0].removeprefix("https://")
case FeedEntryType.ROYALROAD:
title = online_information.feed_title
case _:
assert_never(rss_feed_type)
for user_id, feed_name in subscribers:
try: try:
request.app.state.notifier.send( request.app.state.notifier.send(
title=feed_name or repo, title=feed_name or title,
body=gitea_release_information.body_title, body=online_information.body_title,
link=gitea_release_information.link, link=online_information.link,
category="utility", category=online_information.notification_type,
user_id=user_id, user_id=user_id,
) )
sent += 1 sent += 1
except NotificationUnavailableError: except NotificationUnavailableError:
failed = len(subscribers) - sent failed = len(subscribers) - sent
logger.error(f"[FastAPI] Notifications backend unavailable, skipping " logger.error (f"[FastAPI] Notifications backend unavailable, skipping "
f"{failed} remaining user(s) for feed_id={feed_id}") f"{failed} remaining user(s) for feed={online_information.feed_title}")
break break
except NotificationError as e: except NotificationError as e:
failed += 1 failed += 1
logger.error(f"[FastAPI] Failed to send notification to user_id={user_id} for feed_id={feed_id} " logger.error(f"[FastAPI] Failed to send notification to user_id={user_id} for feed={online_information.feed_title} "
f"status={getattr(e, 'status_code',None)}") f"status={getattr(e, 'status_code',None)}")
if subscribers and sent == 0: if subscribers and sent == 0:
results["errors"] += 1 results["errors"] += 1
@@ -139,14 +130,14 @@ def get_release_information(request: Request, db = Depends(get_db)):
with db.cursor() as cursor: with db.cursor() as cursor:
cursor.execute( cursor.execute(
"UPDATE rss_feeds SET last_entry = %s WHERE id = %s", "UPDATE rss_feeds SET last_entry = %s WHERE id = %s",
(gitea_release_information.entry_id, feed_id), (online_information.entry_id, feed_id),
) )
db.commit() db.commit()
logger.debug(f"[FastAPI] Notification sent and database updated for Gitea feed ID {feed_id}") logger.debug(f"[FastAPI] Notification sent and database updated for feed ID {feed_id}")
results["new"] += 1 results["new"] += 1
if gitea_release_information.link.startswith("https://gitea.com/gitea/runner/releases/tag/v"): if rss_feed_type is FeedEntryType.GITEA and (online_information.link or "").startswith("https://gitea.com/gitea/runner/releases/tag/v"):
gitea_build_success = build_gitea_runner(gitea_release_information.body_title, request.app.state.gitea_token) gitea_build_success = build_gitea_runner(online_information.body_title, request.app.state.gitea_token)
logger.info(f"[FastAPI] Gitea build ran successfully:{gitea_build_success}") logger.info(f"[FastAPI] Gitea build ran successfully:{gitea_build_success}")
except Exception as e: except Exception as e:
@@ -158,105 +149,10 @@ def get_release_information(request: Request, db = Depends(get_db)):
) )
if results["errors"]: if results["errors"]:
logger.warning(f"[FastAPI] Gitea check finished with errors: {results}") logger.warning(f"[FastAPI] {rss_feed_type.value} check finished with errors: {results}")
raise HTTPException(status_code=500, detail=results) raise HTTPException(status_code=500, detail=results)
logger.info(f"[FastAPI] Gitea check complete: {results}") logger.info(f"[FastAPI] {rss_feed_type.value} check complete: {results}")
return results
@scheduler_router.post("/royalroad")
def get_chapters(request: Request, db = Depends(get_db)):
"""
Checks for new Royalroad chapters and updates the database accordingly.
Behaviour:
- Fetches all active stories and their last known chapter ID.
- Queries Royalroad for each story's latest chapter.
- Updates `last_entry` if a new chapter is found.
- Sends a notification for any newly discovered chapters.
- A failure on one story is logged and the remaining stories are still checked.
Returns:
A JSON object with checked/new/error counts. If any story failed,
a 500 is returned with the same counts as the detail.
"""
logger.info("[FastAPI] Checking for new Royalroad chapters")
results = {"checked": 0, "new": 0, "errors": 0}
with db.cursor() as cursor:
cursor.execute(
"SELECT id, rss_feed, last_entry "
"FROM rss_feeds WHERE active = 1 AND type = %s",
("royalroad",),
)
feeds = cursor.fetchall()
logger.debug(f"[FastAPI] Found {len(feeds)} active stories to check.")
for feed_id, rss_feed, last_entry in feeds:
results["checked"] += 1
try:
story = grab_royal_road_chapter_information(rss_feed)
logger.debug(f"[FastAPI] ID:{feed_id},RSS Feed {rss_feed}: db={last_entry}, online={story.entry_id}")
if story.entry_id == last_entry:
continue
logger.info(f"[FastAPI] New chapter detected for story ID {feed_id}: {story.feed_title}")
with db.cursor() as cursor:
cursor.execute (
"SELECT user_id, feed_name "
"FROM rss_feed_subscriptions "
"WHERE rss_feed_id = %s AND active = 1",
(feed_id,),
)
subscribers = cursor.fetchall()
sent = 0
failed = 0
for user_id, feed_name in subscribers:
try:
request.app.state.notifier.send(
title=feed_name or story.feed_title,
body=story.body_title,
link=story.link,
category="royal-road",
user_id=user_id,
)
sent += 1
except NotificationUnavailableError:
failed = len(subscribers) - sent
logger.error (f"[FastAPI] Notifications backend unavailable, skipping "
f"{failed} remaining user(s) for story={story.feed_title}")
break
except NotificationError as e:
failed += 1
logger.error(f"[FastAPI] Failed to send notification to user_id={user_id} for story={story.feed_title} "
f"status={getattr(e, 'status_code',None)}")
if subscribers and sent == 0:
results["errors"] += 1
logger.error(f"[FastAPI] All notifications failed for Feed ID:{feed_id}, will retry later")
continue
with db.cursor() as cursor:
cursor.execute(
"UPDATE rss_feeds SET last_entry = %s WHERE id = %s",
(story.entry_id, feed_id),
)
db.commit()
logger.debug(f"[FastAPI] Notification sent and database updated for story ID {feed_id}")
results["new"] += 1
except Exception as e:
db.rollback()
results["errors"] += 1
logger.error(
f"[Royalroad] Failed for feed ID {feed_id} ({rss_feed}): {e}",
exc_info=True,
)
if results["errors"]:
logger.warning(f"[FastAPI] Royalroad check finished with errors: {results}")
raise HTTPException(status_code=500, detail=results)
logger.info(f"[FastAPI] Royalroad check complete: {results}")
return results return results
@user_router.post("/add/feed") @user_router.post("/add/feed")
@@ -279,14 +175,14 @@ def add_feed(
try: try:
last_entry_id = fetch_latest(feed_entry.type, url).entry_id last_entry_id = fetch_latest(feed_entry.type, url).entry_id
except (ValueError, RequestException) as e: except (ValueError, RequestException) as e:
logger.warning(f"[FastAPI] Could not read latest entry for {url}, storing none") logger.warning(f"[FastAPI] Could not read latest entry for {url}")
raise FeedCheckFailedError(url) from e raise FeedCheckFailedError(url) from e
try: try:
with db.cursor() as cursor: with db.cursor() as cursor:
cursor.execute( cursor.execute(
"INSERT into rss_feeds (type,rss_feed,last_entry) VALUES (%s,%s,%s) " "INSERT into rss_feeds (type,rss_feed,last_entry) VALUES (%s,%s,%s) "
"ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1", "ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1",
(feed_entry.type, url, last_entry_id), (feed_entry.type.value, url, last_entry_id),
) )
feed_id = cursor.lastrowid feed_id = cursor.lastrowid