nas/Auto_Bangumi
1
0
Fork 0
mirror of https://github.com/EstrellaXD/Auto_Bangumi.git synced 2026-11-06 07:29:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: remove dead code before the 3.3 architecture work

Browse Source 
Delete the legacy notification stack (notification.py + plugin/, five
providers duplicated by the active providers/ registry, including a broken
copy-pasted slack plugin), the unused SIGALRM timeout decorator, the unused
ConflictError, the unused bgm_parser, the 0-byte transmission client stub,
unused Lock/Event state in ProgramStatus, the fully-skipped legacy RSS
engine test module, and the dev_server/test_passkey_server scratch entry
points that duplicated app assembly.

The zh notification message template now exists in exactly one place
(NotificationProvider._format_message) — the other seven copies lived in
the deleted legacy plugins.

No behavior change; full suite stays green (726 passed).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014w1Z6Nxy6XTRgkFXqPr9Zh
This commit is contained in:
Estrella Pan
2026-07-02 11:48:37 +02:00
parent fa24ec4e2a
commit f3634baaee
17 changed files with 1 additions and 376 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
backend/src/dev_server.py
View File

@@ -1,47 +0,0 @@
"""Minimal dev server that skips downloader check for UI testing."""
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi import APIRouter
from module.database.combine import Database
from module.database.engine import engine
# Initialize DB + migrations + default user
with Database(engine) as db:
db.create_table()
db.user.add_default_user()
# Build v1 router without program router (which blocks on downloader check)
from module.api.auth import router as auth_router
from module.api.bangumi import router as bangumi_router
from module.api.config import router as config_router
from module.api.log import router as log_router
from module.api.rss import router as rss_router
from module.api.search import router as search_router
v1 = APIRouter(prefix="/v1")
v1.include_router(auth_router)
v1.include_router(bangumi_router)
v1.include_router(config_router)
v1.include_router(log_router)
v1.include_router(rss_router)
v1.include_router(search_router)
# Stub status endpoint (real one lives in program router which blocks on downloader)
@v1.get("/status")
async def stub_status():
return {"status": True, "version": "dev"}
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(v1, prefix="/api")
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=7892)

2
backend/src/module/ab_decorator/__init__.py
View File

@@ -4,8 +4,6 @@ import logging
import httpx
from .timeout import timeout
logger = logging.getLogger(__name__)
_lock = asyncio.Lock()

23
backend/src/module/ab_decorator/timeout.py
View File

@@ -1,23 +0,0 @@
import signal
def timeout(seconds):
def decorator(func):
def handler(signum, frame):
raise TimeoutError("Function timed out.")
def wrapper(*args, **kwargs):
# 设置信号处理程序，当超时时触发TimeoutError异常
signal.signal(signal.SIGALRM, handler)
signal.alarm(seconds) # 设置alarm定时器
try:
result = func(*args, **kwargs)
finally:
signal.alarm(0) # 取消alarm定时器
return result
return wrapper
return decorator

5
backend/src/module/core/status.py
View File

@@ -1,4 +1,3 @@
import asyncio
import time
from module.checker import Checker
@@ -10,13 +9,9 @@ DOWNLOADER_STATUS_TTL = 60
class ProgramStatus(Checker):
def __init__(self):
super().__init__()
self.stop_event = asyncio.Event()
self.lock = asyncio.Lock()
self._downloader_status = False
self._downloader_last_check: float = 0
self._torrents_status = False
self._tasks_started = False
self.event = asyncio.Event()
@property
def is_running(self):

0
backend/src/module/downloader/client/tr_downloader.py
View File

2
backend/src/module/downloader/exceptions.py
View File

@@ -1,2 +0,0 @@
class ConflictError(Exception):
pass

4
backend/src/module/notification/__init__.py
View File

@@ -1,10 +1,8 @@
from .notification import PostNotification
from .manager import NotificationManager
from .base import NotificationProvider
from .manager import NotificationManager
from .providers import PROVIDER_REGISTRY
__all__ = [
"PostNotification",
"NotificationManager",
"NotificationProvider",
"PROVIDER_REGISTRY",

58
backend/src/module/notification/notification.py
View File

@@ -1,58 +0,0 @@
import asyncio
import logging
from module.conf import settings
from module.database import Database
from module.models import Notification
from .plugin import (
BarkNotification,
ServerChanNotification,
TelegramNotification,
WecomNotification,
)
logger = logging.getLogger(__name__)
def getClient(type: str):
if type.lower() == "telegram":
return TelegramNotification
elif type.lower() == "server-chan":
return ServerChanNotification
elif type.lower() == "bark":
return BarkNotification
elif type.lower() == "wecom":
return WecomNotification
else:
return None
class PostNotification:
def __init__(self):
Notifier = getClient(settings.notification.type)
self.notifier = Notifier(
token=settings.notification.token, chat_id=settings.notification.chat_id
)
@staticmethod
def _get_poster_sync(notify: Notification):
with Database() as db:
poster_path = db.bangumi.match_poster(notify.official_title)
notify.poster_path = poster_path
async def send_msg(self, notify: Notification) -> bool:
await asyncio.to_thread(self._get_poster_sync, notify)
try:
await self.notifier.post_msg(notify)
logger.debug("Send notification: %s", notify.official_title)
except Exception as e:
logger.warning(f"Failed to send notification: {e}")
return False
async def __aenter__(self):
await self.notifier.__aenter__()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.notifier.__aexit__(exc_type, exc_val, exc_tb)

4
backend/src/module/notification/plugin/__init__.py
View File

@@ -1,4 +0,0 @@
from .bark import BarkNotification
from .server_chan import ServerChanNotification
from .telegram import TelegramNotification
from .wecom import WecomNotification

27
backend/src/module/notification/plugin/bark.py
View File

@@ -1,27 +0,0 @@
import logging
from module.models import Notification
from module.network import RequestContent
logger = logging.getLogger(__name__)
class BarkNotification(RequestContent):
def __init__(self, token, **kwargs):
super().__init__()
self.token = token
self.notification_url = "https://api.day.app/push"
@staticmethod
def gen_message(notify: Notification) -> str:
text = f"""
番剧名称：{notify.official_title}\n季度： 第{notify.season}季\n更新集数： 第{notify.episode}集\n{notify.poster_path}\n
"""
return text.strip()
async def post_msg(self, notify: Notification) -> bool:
text = self.gen_message(notify)
data = {"title": notify.official_title, "body": text, "icon": notify.poster_path, "device_key": self.token}
resp = await self.post_data(self.notification_url, data)
logger.debug("Bark notification: %s", resp.status_code)
return resp.status_code == 200

31
backend/src/module/notification/plugin/server_chan.py
View File

@@ -1,31 +0,0 @@
import logging
from module.models import Notification
from module.network import RequestContent
logger = logging.getLogger(__name__)
class ServerChanNotification(RequestContent):
"""Server酱推送"""
def __init__(self, token, **kwargs):
super().__init__()
self.notification_url = f"https://sctapi.ftqq.com/{token}.send"
@staticmethod
def gen_message(notify: Notification) -> str:
text = f"""
番剧名称：{notify.official_title}\n季度： 第{notify.season}季\n更新集数： 第{notify.episode}集\n{notify.poster_path}\n
"""
return text.strip()
async def post_msg(self, notify: Notification) -> bool:
text = self.gen_message(notify)
data = {
"title": notify.official_title,
"desp": text,
}
resp = await self.post_data(self.notification_url, data)
logger.debug("ServerChan notification: %s", resp.status_code)
return resp.status_code == 200

27
backend/src/module/notification/plugin/slack.py
View File

@@ -1,27 +0,0 @@
import logging
from module.models import Notification
from module.network import RequestContent
logger = logging.getLogger(__name__)
class SlackNotification(RequestContent):
def __init__(self, token, **kwargs):
super().__init__()
self.token = token
self.notification_url = "https://api.day.app/push"
@staticmethod
def gen_message(notify: Notification) -> str:
text = f"""
番剧名称：{notify.official_title}\n季度： 第{notify.season}季\n更新集数： 第{notify.episode}集\n{notify.poster_path}\n
"""
return text.strip()
async def post_msg(self, notify: Notification) -> bool:
text = self.gen_message(notify)
data = {"title": notify.official_title, "body": text, "device_key": self.token}
resp = await self.post_data(self.notification_url, data)
logger.debug("Bark notification: %s", resp.status_code)
return resp.status_code == 200

38
backend/src/module/notification/plugin/telegram.py
View File

@@ -1,38 +0,0 @@
import logging
from module.models import Notification
from module.network import RequestContent
from module.utils import load_image
logger = logging.getLogger(__name__)
class TelegramNotification(RequestContent):
def __init__(self, token, chat_id):
super().__init__()
self.photo_url = f"https://api.telegram.org/bot{token}/sendPhoto"
self.message_url = f"https://api.telegram.org/bot{token}/sendMessage"
self.chat_id = chat_id
@staticmethod
def gen_message(notify: Notification) -> str:
text = f"""
番剧名称：{notify.official_title}\n季度： 第{notify.season}季\n更新集数： 第{notify.episode}集
"""
return text.strip()
async def post_msg(self, notify: Notification) -> bool:
text = self.gen_message(notify)
data = {
"chat_id": self.chat_id,
"caption": text,
"text": text,
"disable_notification": True,
}
photo = load_image(notify.poster_path)
if photo:
resp = await self.post_files(self.photo_url, data, files={"photo": photo})
else:
resp = await self.post_data(self.message_url, data)
logger.debug("Telegram notification: %s", resp.status_code)
return resp.status_code == 200

42
backend/src/module/notification/plugin/wecom.py
View File

@@ -1,42 +0,0 @@
import logging
from module.models import Notification
from module.network import RequestContent
logger = logging.getLogger(__name__)
class WecomNotification(RequestContent):
"""企业微信推送 基于图文消息"""
def __init__(self, token, chat_id, **kwargs):
super().__init__()
# Chat_id is used as noti_url in this push tunnel
self.notification_url = f"{chat_id}"
self.token = token
@staticmethod
def gen_message(notify: Notification) -> str:
text = f"""
番剧名称：{notify.official_title}\n季度： 第{notify.season}季\n更新集数： 第{notify.episode}集\n{notify.poster_path}\n
"""
return text.strip()
async def post_msg(self, notify: Notification) -> bool:
##Change message format to match Wecom push better
title = "【番剧更新】" + notify.official_title
msg = self.gen_message(notify)
picurl = notify.poster_path
# Default pic to avoid blank in message. Resolution:1068*455
if picurl == "https://mikanani.me":
picurl = "https://article.biliimg.com/bfs/article/d8bcd0408bf32594fd82f27de7d2c685829d1b2e.png"
data = {
"key": self.token,
"type": "news",
"title": title,
"msg": msg,
"picurl": picurl,
}
resp = await self.post_data(self.notification_url, data)
logger.debug("Wecom notification: %s", resp.status_code)
return resp.status_code == 200

15
backend/src/module/parser/analyser/bgm_parser.py
View File

@@ -1,15 +0,0 @@
from module.network import RequestContent
def search_url(e):
return f"https://api.bgm.tv/search/subject/{e}?responseGroup=large"
async def bgm_parser(title):
url = search_url(title)
async with RequestContent() as req:
contents = await req.get_json(url)
if contents:
return contents[0]
else:
return None

17
backend/src/test/test_rss_engine.py
View File

@@ -1,17 +0,0 @@
import pytest
# Skip the entire module as it requires network access and complex setup
pytestmark = pytest.mark.skip(reason="RSS engine tests require network access and complex async setup")
@pytest.mark.asyncio
async def test_rss_engine():
"""
This test requires:
1. Network access to mikanani.me
2. A properly configured async database
3. The RSS feed to be available
To run this test, you need to set up a proper test environment.
"""
pass

35
backend/src/test_passkey_server.py
View File

@@ -1,35 +0,0 @@
"""
Minimal test server for passkey development.
Uses the real auth and passkey API routes without the downloader check.
Run with: uv run python test_passkey_server.py
"""
import uvicorn
from fastapi import FastAPI
from module.api.auth import router as auth_router
from module.api.passkey import router as passkey_router
from module.database import Database
from module.update.startup import first_run
app = FastAPI(title="AutoBangumi Passkey Test")
# Mount real routers
app.include_router(auth_router, prefix="/api/v1")
app.include_router(passkey_router, prefix="/api/v1")
@app.on_event("startup")
async def startup():
"""Create tables and default user (no downloader check)"""
with Database() as db:
db.create_table()
db.user.add_default_user()
@app.get("/")
def index():
return {"status": "Passkey test server running"}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=7892)