AutoBangumi in Golang - 全自动追番工具
import logging
import os

import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from module.api import v1
from module.conf import VERSION, settings, setup_logger

setup_logger(reset=True)
logger = logging.getLogger(__name__)
uvicorn_logging_config = {
    "version": 1,
    "disable_existing_loggers": False,
    "handlers": logger.handlers,
    "loggers": {
        "uvicorn": {
            "level": logger.level,
        },
        "uvicorn.access": {
            "level": "WARNING",
        },
    },
}


def create_app() -> FastAPI:
    app = FastAPI()

    # mount routers
    app.include_router(v1, prefix="/api")

    return app


app = create_app()


@app.get("/posters/{path:path}", tags=["posters"])
def posters(path: str):
    return FileResponse(f"data/posters/{path}")


if VERSION != "DEV_VERSION":
    app.mount("/assets", StaticFiles(directory="dist/assets"), name="assets")
    app.mount("/images", StaticFiles(directory="dist/images"), name="images")
    # app.mount("/icons", StaticFiles(directory="dist/icons"), name="icons")
    templates = Jinja2Templates(directory="dist")

    @app.get("/{path:path}")
    def html(request: Request, path: str):
        files = os.listdir("dist")
        if path in files:
            return FileResponse(f"dist/{path}")
        else:
            context = {"request": request}
            return templates.TemplateResponse("index.html", context)
else:
    @app.get("/", status_code=302, tags=["html"])
    def index():
        return RedirectResponse("/docs")


if __name__ == "__main__":
    if os.getenv("IPV6"):
        host = "::"
    else:
        host = os.getenv("HOST", "0.0.0.0")
    uvicorn.run(
        app,
        host=host,
        port=settings.program.webui_port,
        log_config=uvicorn_logging_config,
    )

backend/src/main.py
maikirakiwi/Goto_Bangumi
import logging
import threading
import time

from .timeout import timeout

logger = logging.getLogger(__name__)
lock = threading.Lock()


def qb_connect_failed_wait(func):
    def wrapper(*args, **kwargs):
        times = 0
        while times < 5:
            try:
                return func(*args, **kwargs)
            except Exception as e:
                logger.debug(f"URL: {args[0]}")
                logger.warning(e)
                logger.warning("Cannot connect to qBittorrent. Wait 5 min and retry...")
                time.sleep(300)
                times += 1

    return wrapper


def api_failed(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.debug(f"URL: {args[0]}")
            logger.warning("Wrong API response.")
            logger.debug(e)

    return wrapper


def locked(func):
    def wrapper(*args, **kwargs):
        with lock:
            return func(*args, **kwargs)

    return wrapper

backend/src/module/ab_decorator/__init__.py
maikirakiwi/Goto_Bangumi
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

backend/src/module/ab_decorator/timeout.py
maikirakiwi/Goto_Bangumi
from fastapi import APIRouter

from .auth import router as auth_router
from .bangumi import router as bangumi_router
from .config import router as config_router
from .log import router as log_router
from .program import router as program_router
from .rss import router as rss_router
from .search import router as search_router

__all__ = "v1"

# API 1.0
v1 = APIRouter(prefix="/v1")
v1.include_router(auth_router)
v1.include_router(log_router)
v1.include_router(program_router)
v1.include_router(bangumi_router)
v1.include_router(config_router)
v1.include_router(rss_router)
v1.include_router(search_router)

backend/src/module/api/__init__.py
maikirakiwi/Goto_Bangumi
from datetime import timedelta

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.responses import JSONResponse, Response
from fastapi.security import OAuth2PasswordRequestForm

from module.models import APIResponse
from module.models.user import User, UserUpdate
from module.security.api import (
    active_user,
    auth_user,
    get_current_user,
    update_user_info,
)
from module.security.jwt import create_access_token

from .response import u_response

router = APIRouter(prefix="/auth", tags=["auth"])


@router.post("/login", response_model=dict)
async def login(response: Response, form_data=Depends(OAuth2PasswordRequestForm)):
    user = User(username=form_data.username, password=form_data.password)
    resp = auth_user(user)
    if resp.status:
        token = create_access_token(
            data={"sub": user.username}, expires_delta=timedelta(days=1)
        )
        response.set_cookie(key="token", value=token, httponly=True, max_age=86400)
        return {"access_token": token, "token_type": "bearer"}
    return u_response(resp)


@router.get(
    "/refresh_token", response_model=dict, dependencies=[Depends(get_current_user)]
)
async def refresh(response: Response):
    token = create_access_token(
        data={"sub": active_user[0]}, expires_delta=timedelta(days=1)
    )
    response.set_cookie(key="token", value=token, httponly=True, max_age=86400)
    return {"access_token": token, "token_type": "bearer"}


@router.get(
    "/logout", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def logout(response: Response):
    active_user.clear()
    response.delete_cookie(key="token")
    return JSONResponse(
        status_code=200,
        content={"msg_en": "Logout successfully.", "msg_zh": "登出成功。"},
    )


@router.post("/update", response_model=dict, dependencies=[Depends(get_current_user)])
async def update_user(user_data: UserUpdate, response: Response):
    old_user = active_user[0]
    if update_user_info(user_data, old_user):
        token = create_access_token(
            data={"sub": old_user}, expires_delta=timedelta(days=1)
        )
        response.set_cookie(
            key="token",
            value=token,
            httponly=True,
            max_age=86400,
        )
        return {
            "access_token": token,
            "token_type": "bearer",
            "message": "update success",
        }

backend/src/module/api/auth.py
maikirakiwi/Goto_Bangumi
from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse

from module.manager import TorrentManager
from module.models import APIResponse, Bangumi, BangumiUpdate
from module.security.api import UNAUTHORIZED, get_current_user

from .response import u_response

router = APIRouter(prefix="/bangumi", tags=["bangumi"])


def str_to_list(data: Bangumi):
    data.filter = data.filter.split(",")
    data.rss_link = data.rss_link.split(",")
    return data


@router.get(
    "/get/all", response_model=list[Bangumi], dependencies=[Depends(get_current_user)]
)
async def get_all_data():
    with TorrentManager() as manager:
        return manager.bangumi.search_all()


@router.get(
    "/get/{bangumi_id}",
    response_model=Bangumi,
    dependencies=[Depends(get_current_user)],
)
async def get_data(bangumi_id: str):
    with TorrentManager() as manager:
        resp = manager.search_one(bangumi_id)
    return resp


@router.patch(
    "/update/{bangumi_id}",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def update_rule(
    bangumi_id: int,
    data: BangumiUpdate,
):
    with TorrentManager() as manager:
        resp = manager.update_rule(bangumi_id, data)
    return u_response(resp)


@router.delete(
    path="/delete/{bangumi_id}",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def delete_rule(bangumi_id: str, file: bool = False):
    with TorrentManager() as manager:
        resp = manager.delete_rule(bangumi_id, file)
    return u_response(resp)


@router.delete(
    path="/delete/many/",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def delete_many_rule(bangumi_id: list, file: bool = False):
    with TorrentManager() as manager:
        for i in bangumi_id:
            resp = manager.delete_rule(i, file)
    return u_response(resp)


@router.delete(
    path="/disable/{bangumi_id}",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def disable_rule(bangumi_id: str, file: bool = False):
    with TorrentManager() as manager:
        resp = manager.disable_rule(bangumi_id, file)
    return u_response(resp)


@router.delete(
    path="/disable/many/",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def disable_many_rule(bangumi_id: list, file: bool = False):
    with TorrentManager() as manager:
        for i in bangumi_id:
            resp = manager.disable_rule(i, file)
    return u_response(resp)


@router.get(
    path="/enable/{bangumi_id}",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def enable_rule(bangumi_id: str):
    with TorrentManager() as manager:
        resp = manager.enable_rule(bangumi_id)
    return u_response(resp)


@router.get(
    path="/refresh/poster/all",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def refresh_poster():
    with TorrentManager() as manager:
        resp = manager.refresh_poster()
    return u_response(resp)


@router.get(
    "/reset/all", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def reset_all():
    with TorrentManager() as manager:
        manager.bangumi.delete_all()
        return JSONResponse(
            status_code=200,
            content={"msg_en": "Reset all rules successfully.", "msg_zh": "重置所有规则成功。"},
        )

backend/src/module/api/bangumi.py
maikirakiwi/Goto_Bangumi
import logging

from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse

from module.conf import settings
from module.models import APIResponse, Config
from module.security.api import UNAUTHORIZED, get_current_user

router = APIRouter(prefix="/config", tags=["config"])
logger = logging.getLogger(__name__)


@router.get("/get", response_model=Config, dependencies=[Depends(get_current_user)])
async def get_config():
    return settings


@router.patch(
    "/update", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def update_config(config: Config):
    try:
        settings.save(config_dict=config.dict())
        settings.load()
        # update_rss()
        logger.info("Config updated")
        return JSONResponse(
            status_code=200,
            content={"msg_en": "Update config successfully.", "msg_zh": "更新配置成功。"},
        )
    except Exception as e:
        logger.warning(e)
        return JSONResponse(
            status_code=406,
            content={"msg_en": "Update config failed.", "msg_zh": "更新配置失败。"},
        )

backend/src/module/api/config.py
maikirakiwi/Goto_Bangumi
from fastapi import APIRouter, Depends, HTTPException, Response, status
from fastapi.responses import JSONResponse

from module.conf import LOG_PATH
from module.models import APIResponse
from module.security.api import UNAUTHORIZED, get_current_user

router = APIRouter(prefix="/log", tags=["log"])


@router.get("", response_model=str, dependencies=[Depends(get_current_user)])
async def get_log():
    if LOG_PATH.exists():
        with open(LOG_PATH, "rb") as f:
            return Response(f.read(), media_type="text/plain")
    else:
        return Response("Log file not found", status_code=404)


@router.get(
    "/clear", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def clear_log():
    if LOG_PATH.exists():
        LOG_PATH.write_text("")
        return JSONResponse(
            status_code=200,
            content={"msg_en": "Log cleared successfully.", "msg_zh": "日志清除成功。"},
        )
    else:
        return JSONResponse(
            status_code=406,
            content={"msg_en": "Log file not found.", "msg_zh": "日志文件未找到。"},
        )

backend/src/module/api/log.py
maikirakiwi/Goto_Bangumi
import logging
import os
import signal

from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import JSONResponse

from module.conf import VERSION
from module.core import Program
from module.models import APIResponse
from module.security.api import UNAUTHORIZED, get_current_user

from .response import u_response

logger = logging.getLogger(__name__)
program = Program()
router = APIRouter(tags=["program"])


@router.on_event("startup")
async def startup():
    program.startup()


@router.on_event("shutdown")
async def shutdown():
    program.stop()


@router.get(
    "/restart", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def restart():
    try:
        resp = program.restart()
        return u_response(resp)
    except Exception as e:
        logger.debug(e)
        logger.warning("Failed to restart program")
        raise HTTPException(
            status_code=500,
            detail={
                "msg_en": "Failed to restart program.",
                "msg_zh": "重启程序失败。",
            },
        )


@router.get(
    "/start", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def start():
    try:
        resp = program.start()
        return u_response(resp)
    except Exception as e:
        logger.debug(e)
        logger.warning("Failed to start program")
        raise HTTPException(
            status_code=500,
            detail={
                "msg_en": "Failed to start program.",
                "msg_zh": "启动程序失败。",
            },
        )


@router.get(
    "/stop", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def stop():
    return u_response(program.stop())


@router.get("/status", response_model=dict, dependencies=[Depends(get_current_user)])
async def program_status():
    if not program.is_running:
        return {
            "status": False,
            "version": VERSION,
            "first_run": program.first_run,
        }
    else:
        return {
            "status": True,
            "version": VERSION,
            "first_run": program.first_run,
        }


@router.get(
    "/shutdown", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def shutdown_program():
    program.stop()
    logger.info("Shutting down program...")
    os.kill(os.getpid(), signal.SIGINT)
    return JSONResponse(
        status_code=200,
        content={"msg_en": "Shutdown program successfully.", "msg_zh": "关闭程序成功。"},
    )


# Check status
@router.get(
    "/check/downloader",
    tags=["check"],
    response_model=bool,
    dependencies=[Depends(get_current_user)],
)
async def check_downloader_status():
    return program.check_downloader()

backend/src/module/api/program.py
maikirakiwi/Goto_Bangumi
from fastapi.exceptions import HTTPException
from fastapi.responses import JSONResponse

from module.models.response import ResponseModel


def u_response(response_model: ResponseModel):
    return JSONResponse(
        status_code=response_model.status_code,
        content={
            "msg_en": response_model.msg_en,
            "msg_zh": response_model.msg_zh,
        },
    )

backend/src/module/api/response.py
maikirakiwi/Goto_Bangumi
from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse

from module.downloader import DownloadClient
from module.manager import SeasonCollector
from module.models import APIResponse, Bangumi, RSSItem, RSSUpdate, Torrent
from module.rss import RSSAnalyser, RSSEngine
from module.security.api import UNAUTHORIZED, get_current_user

from .response import u_response

router = APIRouter(prefix="/rss", tags=["rss"])


@router.get(
    path="", response_model=list[RSSItem], dependencies=[Depends(get_current_user)]
)
async def get_rss():
    with RSSEngine() as engine:
        return engine.rss.search_all()


@router.post(
    path="/add", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def add_rss(rss: RSSItem):
    with RSSEngine() as engine:
        result = engine.add_rss(rss.url, rss.name, rss.aggregate, rss.parser)
    return u_response(result)


@router.post(
    path="/enable/many",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def enable_many_rss(
    rss_ids: list[int],
):
    with RSSEngine() as engine:
        result = engine.enable_list(rss_ids)
    return u_response(result)


@router.delete(
    path="/delete/{rss_id}",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def delete_rss(rss_id: int):
    with RSSEngine() as engine:
        if engine.rss.delete(rss_id):
            return JSONResponse(
                status_code=200,
                content={"msg_en": "Delete RSS successfully.", "msg_zh": "删除 RSS 成功。"},
            )
        else:
            return JSONResponse(
                status_code=406,
                content={"msg_en": "Delete RSS failed.", "msg_zh": "删除 RSS 失败。"},
            )


@router.post(
    path="/delete/many",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def delete_many_rss(
    rss_ids: list[int],
):
    with RSSEngine() as engine:
        result = engine.delete_list(rss_ids)
    return u_response(result)


@router.patch(
    path="/disable/{rss_id}",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def disable_rss(rss_id: int):
    with RSSEngine() as engine:
        if engine.rss.disable(rss_id):
            return JSONResponse(
                status_code=200,
                content={"msg_en": "Disable RSS successfully.", "msg_zh": "禁用 RSS 成功。"},
            )
        else:
            return JSONResponse(
                status_code=406,
                content={"msg_en": "Disable RSS failed.", "msg_zh": "禁用 RSS 失败。"},
            )


@router.post(
    path="/disable/many",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def disable_many_rss(rss_ids: list[int]):
    with RSSEngine() as engine:
        result = engine.disable_list(rss_ids)
    return u_response(result)


@router.patch(
    path="/update/{rss_id}",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def update_rss(
    rss_id: int, data: RSSUpdate, current_user=Depends(get_current_user)
):
    if not current_user:
        raise UNAUTHORIZED
    with RSSEngine() as engine:
        if engine.rss.update(rss_id, data):
            return JSONResponse(
                status_code=200,
                content={"msg_en": "Update RSS successfully.", "msg_zh": "更新 RSS 成功。"},
            )
        else:
            return JSONResponse(
                status_code=406,
                content={"msg_en": "Update RSS failed.", "msg_zh": "更新 RSS 失败。"},
            )


@router.get(
    path="/refresh/all",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def refresh_all():
    with RSSEngine() as engine, DownloadClient() as client:
        engine.refresh_rss(client)
        return JSONResponse(
            status_code=200,
            content={"msg_en": "Refresh all RSS successfully.", "msg_zh": "刷新 RSS 成功。"},
        )


@router.get(
    path="/refresh/{rss_id}",
    response_model=APIResponse,
    dependencies=[Depends(get_current_user)],
)
async def refresh_rss(rss_id: int):
    with RSSEngine() as engine, DownloadClient() as client:
        engine.refresh_rss(client, rss_id)
        return JSONResponse(
            status_code=200,
            content={"msg_en": "Refresh RSS successfully.", "msg_zh": "刷新 RSS 成功。"},
        )


@router.get(
    path="/torrent/{rss_id}",
    response_model=list[Torrent],
    dependencies=[Depends(get_current_user)],
)
async def get_torrent(
    rss_id: int,
):
    with RSSEngine() as engine:
        return engine.get_rss_torrents(rss_id)


# Old API
analyser = RSSAnalyser()


@router.post(
    "/analysis", response_model=Bangumi, dependencies=[Depends(get_current_user)]
)
async def analysis(rss: RSSItem):
    data = analyser.link_to_data(rss)
    if isinstance(data, Bangumi):
        return data
    else:
        return u_response(data)


@router.post(
    "/collect", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def download_collection(data: Bangumi):
    with SeasonCollector() as collector:
        resp = collector.collect_season(data, data.rss_link)
        return u_response(resp)


@router.post(
    "/subscribe", response_model=APIResponse, dependencies=[Depends(get_current_user)]
)
async def subscribe(data: Bangumi):
    with SeasonCollector() as collector:
        resp = collector.subscribe_season(data)
        return u_response(resp)

backend/src/module/api/rss.py
maikirakiwi/Goto_Bangumi
from fastapi import APIRouter, Depends, Query
from sse_starlette.sse import EventSourceResponse

from module.models import Bangumi
from module.searcher import SEARCH_CONFIG, SearchTorrent
from module.security.api import UNAUTHORIZED, get_current_user

router = APIRouter(prefix="/search", tags=["search"])


@router.get(
    "/bangumi", response_model=list[Bangumi], dependencies=[Depends(get_current_user)]
)
async def search_torrents(site: str = "mikan", keywords: str = Query(None)):
    """
    Server Send Event for per Bangumi item
    """
    if not keywords:
        return []
    keywords = keywords.split(" ")
    with SearchTorrent() as st:
        return EventSourceResponse(
            content=st.analyse_keyword(keywords=keywords, site=site),
        )


@router.get(
    "/provider", response_model=list[str], dependencies=[Depends(get_current_user)]
)
async def search_provider():
    return list(SEARCH_CONFIG.keys())

backend/src/module/api/search.py
maikirakiwi/Goto_Bangumi
import logging
from pathlib import Path

import requests

from module.conf import VERSION, settings
from module.downloader import DownloadClient
from module.models import Config
from module.update import version_check

logger = logging.getLogger(__name__)


class Checker:
    def __init__(self):
        pass

    @staticmethod
    def check_renamer() -> bool:
        if settings.bangumi_manage.enable:
            return True
        else:
            return False

    @staticmethod
    def check_analyser() -> bool:
        if settings.rss_parser.enable:
            return True
        else:
            return False

    @staticmethod
    def check_first_run() -> bool:
        if settings.dict() == Config().dict():
            return True
        else:
            return False

    @staticmethod
    def check_version() -> bool:
        return version_check()

    @staticmethod
    def check_database() -> bool:
        db_path = Path("data/data.db")
        if not db_path.exists():
            return False
        else:
            return True

    @staticmethod
    def check_downloader() -> bool:
        try:
            url = (
                f"http://{settings.downloader.host}"
                if "://" not in settings.downloader.host
                else f"{settings.downloader.host}"
            )
            response = requests.get(url, timeout=2)
            if settings.downloader.type in response.text.lower():
                with DownloadClient() as client:
                    if client.authed:
                        return True
                    else:
                        return False
            else:
                return False
        except requests.exceptions.ReadTimeout:
            logger.error("[Checker] Downloader connect timeout.")
            return False
        except requests.exceptions.ConnectionError:
            logger.error("[Checker] Downloader connect failed.")
            return False
        except Exception as e:
            logger.error(f"[Checker] Downloader connect failed: {e}")
            return False

    @staticmethod
    def check_img_cache() -> bool:
        img_path = Path("data/posters")
        if img_path.exists():
            return True
        else:
            img_path.mkdir()
            return False

backend/src/module/checker/checker.py
maikirakiwi/Goto_Bangumi
from pathlib import Path

from .config import VERSION, settings
from .log import LOG_PATH, setup_logger
from .search_provider import SEARCH_CONFIG

TMDB_API = "32b19d6a05b512190a056fa4e747cbbc"
DATA_PATH = "sqlite:///data/data.db"
LEGACY_DATA_PATH = Path("data/data.json")
VERSION_PATH = Path("config/version.info")

PLATFORM = "Windows" if "\\" in settings.downloader.path else "Unix"

backend/src/module/conf/__init__.py
maikirakiwi/Goto_Bangumi
import json
import logging
import os
from pathlib import Path

from dotenv import load_dotenv

from module.models.config import Config

from .const import ENV_TO_ATTR

logger = logging.getLogger(__name__)
CONFIG_ROOT = Path("config")


try:
    from module.__version__ import VERSION
except ImportError:
    logger.info("Can't find version info, use DEV_VERSION instead")
    VERSION = "DEV_VERSION"

CONFIG_PATH = (
    CONFIG_ROOT / "config_dev.json"
    if VERSION == "DEV_VERSION"
    else CONFIG_ROOT / "config.json"
).resolve()


class Settings(Config):
    def __init__(self):
        super().__init__()
        if CONFIG_PATH.exists():
            self.load()
            self.save()
        else:
            self.init()

    def load(self):
        with open(CONFIG_PATH, "r", encoding="utf-8") as f:
            config = json.load(f)
        config_obj = Config.parse_obj(config)
        self.__dict__.update(config_obj.__dict__)
        logger.info("Config loaded")

    def save(self, config_dict: dict | None = None):
        if not config_dict:
            config_dict = self.dict()
        with open(CONFIG_PATH, "w", encoding="utf-8") as f:
            json.dump(config_dict, f, indent=4, ensure_ascii=False)

    def init(self):
        load_dotenv(".env")
        self.__load_from_env()
        self.save()

    def __load_from_env(self):
        config_dict = self.dict()
        for key, section in ENV_TO_ATTR.items():
            for env, attr in section.items():
                if env in os.environ:
                    if isinstance(attr, list):
                        for _attr in attr:
                            attr_name = _attr[0] if isinstance(_attr, tuple) else _attr
                            config_dict[key][attr_name] = self.__val_from_env(
                                env, _attr
                            )
                    else:
                        attr_name = attr[0] if isinstance(attr, tuple) else attr
                        config_dict[key][attr_name] = self.__val_from_env(env, attr)
        config_obj = Config.parse_obj(config_dict)
        self.__dict__.update(config_obj.__dict__)
        logger.info("Config loaded from env")

    @staticmethod
    def __val_from_env(env: str, attr: tuple):
        if isinstance(attr, tuple):
            conv_func = attr[1]
            return conv_func(os.environ[env])
        else:
            return os.environ[env]

    @property
    def group_rules(self):
        return self.__dict__["group_rules"]


settings = Settings()

backend/src/module/conf/config.py
maikirakiwi/Goto_Bangumi
# -*- encoding: utf-8 -*-
from urllib.parse import parse_qs, urlparse

DEFAULT_SETTINGS = {
    "program": {
        "sleep_time": 7200,
        "times": 20,
        "webui_port": 7892,
        "data_version": "3.1.7",
    },
    "downloader": {
        "type": "qbittorrent",
        "host": "127.0.0.1:8080",
        "username": "admin",
        "password": "adminadmin",
        "path": "/downloads/Bangumi",
        "ssl": False,
    },
    "rss_parser": {
        "enable": True,
        "type": "mikan",
        "custom_url": "mikanani.me",
        "token": "",
        "enable_tmdb": False,
        "filter": ["720", "\\d+-\\d+"],
        "language": "zh",
    },
    "bangumi_manage": {
        "enable": True,
        "eps_complete": False,
        "rename_method": "pn",
        "group_tag": False,
        "remove_bad_torrent": False,
    },
    "log": {
        "debug_enable": False,
    },
    "proxy": {
        "enable": False,
        "type": "http",
        "host": "",
        "port": 1080,
        "username": "",
        "password": "",
    },
    "notification": {"enable": False, "type": "telegram", "token": "", "chat_id": ""},
}


ENV_TO_ATTR = {
    "program": {
        "AB_INTERVAL_TIME": ("sleep_time", lambda e: int(e)),
        "AB_RENAME_FREQ": ("times", lambda e: int(e)),
        "AB_WEBUI_PORT": ("webui_port", lambda e: int(e)),
    },
    "downloader": {
        "AB_DOWNLOADER_HOST": "host",
        "AB_DOWNLOADER_USERNAME": "username",
        "AB_DOWNLOADER_PASSWORD": "password",
        "AB_DOWNLOAD_PATH": "path",
    },
    "rss_parser": {
        "AB_RSS_COLLECTOR": ("enable", lambda e: e.lower() in ("true", "1", "t")),
        "AB_RSS": [
            ("token", lambda e: parse_qs(urlparse(e).query).get("token", [None])[0]),
            ("custom_url", lambda e: urlparse(e).netloc),
        ],
        "AB_NOT_CONTAIN": ("filter", lambda e: e.split("|")),
        "AB_LANGUAGE": "language",
        "AB_ENABLE_TMDB": ("enable_tmdb", lambda e: e.lower() in ("true", "1", "t")),
    },
    "bangumi_manage": {
        "AB_RENAME": ("enable", lambda e: e.lower() in ("true", "1", "t")),
        "AB_METHOD": ("rename_method", lambda e: e.lower()),
        "AB_GROUP_TAG": ("group_tag", lambda e: e.lower() in ("true", "1", "t")),
        "AB_EP_COMPLETE": ("eps_complete", lambda e: e.lower() in ("true", "1", "t")),
        "AB_REMOVE_BAD_BT": (
            "remove_bad_torrent",
            lambda e: e.lower() in ("true", "1", "t"),
        ),
    },
    "log": {
        "AB_DEBUG_MODE": ("debug_enable", lambda e: e.lower() in ("true", "1", "t")),
    },
    "proxy": {
        "AB_HTTP_PROXY": [
            ("enable", lambda e: True),
            ("type", lambda e: "http"),
            ("host", lambda e: e.split(":")[0]),
            ("port", lambda e: int(e.split(":")[1])),
        ],
        "AB_SOCKS": [
            ("enable", lambda e: True),
            ("type", lambda e: "socks"),
            ("host", lambda e: e.split(",")[0]),
            ("port", lambda e: int(e.split(",")[1])),
            ("username", lambda e: e.split(",")[2]),
            ("password", lambda e: e.split(",")[3]),
        ],
    },
}


class BCOLORS:
    @staticmethod
    def _(color: str, *args: str) -> str:
        strings = [str(s) for s in args]
        return f"{color}{', '.join(strings)}{BCOLORS.ENDC}"

    HEADER = "\033[95m"
    OKBLUE = "\033[94m"
    OKCYAN = "\033[96m"
    OKGREEN = "\033[92m"
    WARNING = "\033[93m"
    FAIL = "\033[91m"
    ENDC = "\033[0m"
    BOLD = "\033[1m"
    UNDERLINE = "\033[4m"

backend/src/module/conf/const.py
maikirakiwi/Goto_Bangumi
import logging
from pathlib import Path

from .config import settings

LOG_ROOT = Path("data")
LOG_PATH = LOG_ROOT / "log.txt"


def setup_logger(level: int = logging.INFO, reset: bool = False):
    level = logging.DEBUG if settings.log.debug_enable else level
    LOG_ROOT.mkdir(exist_ok=True)

    if reset and LOG_PATH.exists():
        LOG_PATH.unlink(missing_ok=True)

    logging.addLevelName(logging.DEBUG, "DEBUG:")
    logging.addLevelName(logging.INFO, "INFO:")
    logging.addLevelName(logging.WARNING, "WARNING:")
    LOGGING_FORMAT = "[%(asctime)s] %(levelname)-8s  %(message)s"
    TIME_FORMAT = "%Y-%m-%d %H:%M:%S"
    logging.basicConfig(
        level=level,
        format=LOGGING_FORMAT,
        datefmt=TIME_FORMAT,
        encoding="utf-8",
        handlers=[
            logging.FileHandler(LOG_PATH, encoding="utf-8"),
            logging.StreamHandler(),
        ],
    )

backend/src/module/conf/log.py
maikirakiwi/Goto_Bangumi
import argparse


def parse():
    parser = argparse.ArgumentParser(
        prog="Auto Bangumi",
        description="""
        本项目是基于 Mikan Project、qBittorrent 的全自动追番整理下载工具。
        只需要在 Mikan Project 上订阅番剧，就可以全自动追番。
        并且整理完成的名称和目录可以直接被 Plex、Jellyfin 等媒体库软件识别，
        无需二次刮削。""",
    )

    parser.add_argument("-d", "--debug", action="store_true", help="debug mode")
    return parser.parse_args()

backend/src/module/conf/parse.py
maikirakiwi/Goto_Bangumi
from pathlib import Path

from module.utils import json_config

DEFAULT_PROVIDER = {
    "mikan": "https://mikanani.me/RSS/Search?searchstr=%s",
    "nyaa": "https://nyaa.si/?page=rss&q=%s&c=0_0&f=0",
    "dmhy": "http://dmhy.org/topics/rss/rss.xml?keyword=%s",
}

PROVIDER_PATH = Path("config/search_provider.json")


def load_provider():
    if PROVIDER_PATH.exists():
        return json_config.load(PROVIDER_PATH)
    else:
        json_config.save(PROVIDER_PATH, DEFAULT_PROVIDER)
        return DEFAULT_PROVIDER


SEARCH_CONFIG = load_provider()

backend/src/module/conf/search_provider.py
maikirakiwi/Goto_Bangumi
from .log import LOG_PATH

logging_config = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "default": {
            "format": "[%(asctime)s] %(levelname)-8s  %(message)s",
            "datefmt": "%Y-%m-%d %H:%M:%S",
        }
    },
    "handlers": {
        "file": {
            "class": "logging.FileHandler",
            "filename": LOG_PATH,
            "formatter": "default",
            "encoding": "utf-8",
        },
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "default",
        },
    },
    "loggers": {
        "uvicorn": {
            "handlers": ["console"],
            "level": "INFO",
        },
        "uvicorn.error": {
            "level": "WARNING",
        },
        "uvicorn.access": {
            "handlers": ["console"],
            "level": "DEBUG",
            "propagate": False,
        },
    },
}

backend/src/module/conf/uvicorn_logging.py
maikirakiwi/Goto_Bangumi
import logging

from module.conf import VERSION, settings
from module.models import ResponseModel
from module.update import data_migration, first_run, from_30_to_31, start_up, cache_image

from .sub_thread import RenameThread, RSSThread

logger = logging.getLogger(__name__)

figlet = r"""
                _        ____                                    _
     /\        | |      |  _ \                                  (_)
    /  \  _   _| |_ ___ | |_) | __ _ _ __   __ _ _   _ _ __ ___  _
   / /\ \| | | | __/ _ \|  _ < / _` | '_ \ / _` | | | | '_ ` _ \| |
  / ____ \ |_| | || (_) | |_) | (_| | | | | (_| | |_| | | | | | | |
 /_/    \_\__,_|\__\___/|____/ \__,_|_| |_|\__, |\__,_|_| |_| |_|_|
                                            __/ |
                                           |___/
"""


class Program(RenameThread, RSSThread):
    @staticmethod
    def __start_info():
        for line in figlet.splitlines():
            logger.info(line.strip("\n"))
        logger.info(
            f"Version {VERSION}  Author: EstrellaXD Twitter: https://twitter.com/Estrella_Pan"
        )
        logger.info("GitHub: https://github.com/EstrellaXD/Auto_Bangumi/")
        logger.info("Starting AutoBangumi...")

    def startup(self):
        self.__start_info()
        if not self.database:
            first_run()
            logger.info("[Core] No db file exists, create database file.")
            return {"status": "First run detected."}
        if self.legacy_data:
            logger.info(
                "[Core] Legacy data detected, starting data migration, please wait patiently."
            )
            data_migration()
        elif self.version_update:
            # Update database
            from_30_to_31()
            logger.info("[Core] Database updated.")
        if not self.img_cache:
            logger.info("[Core] No image cache exists, create image cache.")
            cache_image()
        self.start()

    def start(self):
        self.stop_event.clear()
        settings.load()
        if self.downloader_status:
            if self.enable_renamer:
                self.rename_start()
            if self.enable_rss:
                self.rss_start()
            logger.info("Program running.")
            return ResponseModel(
                status=True,
                status_code=200,
                msg_en="Program started.",
                msg_zh="程序启动成功。",
            )
        else:
            self.stop_event.set()
            logger.warning("Program failed to start.")
            return ResponseModel(
                status=False,
                status_code=406,
                msg_en="Program failed to start.",
                msg_zh="程序启动失败。",
            )

    def stop(self):
        if self.is_running:
            self.stop_event.set()
            self.rename_stop()
            self.rss_stop()
            return ResponseModel(
                status=True,
                status_code=200,
                msg_en="Program stopped.",
                msg_zh="程序停止成功。",
            )
        else:
            return ResponseModel(
                status=False,
                status_code=406,
                msg_en="Program is not running.",
                msg_zh="程序未运行。",
            )

    def restart(self):
        self.stop()
        self.start()
        return ResponseModel(
            status=True,
            status_code=200,
            msg_en="Program restarted.",
            msg_zh="程序重启成功。",
        )

    def update_database(self):
        if not self.version_update:
            return {"status": "No update found."}
        else:
            start_up()
            return {"status": "Database updated."}

backend/src/module/core/program.py
maikirakiwi/Goto_Bangumi
import asyncio
import threading

from module.checker import Checker
from module.conf import LEGACY_DATA_PATH


class ProgramStatus(Checker):
    def __init__(self):
        super().__init__()
        self.stop_event = threading.Event()
        self.lock = threading.Lock()
        self._downloader_status = False
        self._torrents_status = False
        self.event = asyncio.Event()

    @property
    def is_running(self):
        if self.stop_event.is_set() or self.check_first_run():
            return False
        else:
            return True

    @property
    def is_stopped(self):
        return self.stop_event.is_set()

    @property
    def downloader_status(self):
        if not self._downloader_status:
            self._downloader_status = self.check_downloader()
        return self._downloader_status

    @property
    def enable_rss(self):
        return self.check_analyser()

    @property
    def enable_renamer(self):
        return self.check_renamer()

    @property
    def first_run(self):
        return self.check_first_run()

    @property
    def legacy_data(self):
        return LEGACY_DATA_PATH.exists()

    @property
    def version_update(self):
        return not self.check_version()

    @property
    def database(self):
        return self.check_database()

    @property
    def img_cache(self):
        return self.check_img_cache()

backend/src/module/core/status.py
maikirakiwi/Goto_Bangumi
import threading
import time

from module.conf import settings
from module.downloader import DownloadClient
from module.manager import Renamer, eps_complete
from module.notification import PostNotification
from module.rss import RSSAnalyser, RSSEngine

from .status import ProgramStatus


class RSSThread(ProgramStatus):
    def __init__(self):
        super().__init__()
        self._rss_thread = threading.Thread(
            target=self.rss_loop,
        )
        self.analyser = RSSAnalyser()

    def rss_loop(self):
        while not self.stop_event.is_set():
            with DownloadClient() as client, RSSEngine() as engine:
                # Analyse RSS
                rss_list = engine.rss.search_aggregate()
                for rss in rss_list:
                    self.analyser.rss_to_data(rss, engine)
                # Run RSS Engine
                engine.refresh_rss(client)
            if settings.bangumi_manage.eps_complete:
                eps_complete()
            self.stop_event.wait(settings.program.rss_time)

    def rss_start(self):
        self.rss_thread.start()

    def rss_stop(self):
        if self._rss_thread.is_alive():
            self._rss_thread.join()

    @property
    def rss_thread(self):
        if not self._rss_thread.is_alive():
            self._rss_thread = threading.Thread(
                target=self.rss_loop,
            )
        return self._rss_thread


class RenameThread(ProgramStatus):
    def __init__(self):
        super().__init__()
        self._rename_thread = threading.Thread(
            target=self.rename_loop,
        )

    def rename_loop(self):
        while not self.stop_event.is_set():
            with Renamer() as renamer:
                renamed_info = renamer.rename()
            if settings.notification.enable:
                with PostNotification() as notifier:
                    for info in renamed_info:
                        notifier.send_msg(info)
                        time.sleep(2)
            self.stop_event.wait(settings.program.rename_time)

    def rename_start(self):
        self.rename_thread.start()

    def rename_stop(self):
        if self._rename_thread.is_alive():
            self._rename_thread.join()

    @property
    def rename_thread(self):
        if not self._rename_thread.is_alive():
            self._rename_thread = threading.Thread(
                target=self.rename_loop,
            )
        return self._rename_thread

backend/src/module/core/sub_thread.py
maikirakiwi/Goto_Bangumi
from .combine import Database
from .engine import engine

backend/src/module/database/__init__.py
maikirakiwi/Goto_Bangumi
import logging
from typing import Optional

from sqlalchemy.sql import func
from sqlmodel import Session, and_, delete, false, or_, select

from module.models import Bangumi, BangumiUpdate

logger = logging.getLogger(__name__)


class BangumiDatabase:
    def __init__(self, session: Session):
        self.session = session

    def add(self, data: Bangumi):
        statement = select(Bangumi).where(Bangumi.title_raw == data.title_raw)
        bangumi = self.session.exec(statement).first()
        if bangumi:
            return False
        self.session.add(data)
        self.session.commit()
        logger.debug(f"[Database] Insert {data.official_title} into database.")
        return True

    def add_all(self, datas: list[Bangumi]):
        self.session.add_all(datas)
        self.session.commit()
        logger.debug(f"[Database] Insert {len(datas)} bangumi into database.")

    def update(self, data: Bangumi | BangumiUpdate, _id: int = None) -> bool:
        if _id and isinstance(data, BangumiUpdate):
            db_data = self.session.get(Bangumi, _id)
        elif isinstance(data, Bangumi):
            db_data = self.session.get(Bangumi, data.id)
        else:
            return False
        if not db_data:
            return False
        bangumi_data = data.dict(exclude_unset=True)
        for key, value in bangumi_data.items():
            setattr(db_data, key, value)
        self.session.add(db_data)
        self.session.commit()
        self.session.refresh(db_data)
        logger.debug(f"[Database] Update {data.official_title}")
        return True

    def update_all(self, datas: list[Bangumi]):
        self.session.add_all(datas)
        self.session.commit()
        logger.debug(f"[Database] Update {len(datas)} bangumi.")

    def update_rss(self, title_raw, rss_set: str):
        # Update rss and added
        statement = select(Bangumi).where(Bangumi.title_raw == title_raw)
        bangumi = self.session.exec(statement).first()
        bangumi.rss_link = rss_set
        bangumi.added = False
        self.session.add(bangumi)
        self.session.commit()
        self.session.refresh(bangumi)
        logger.debug(f"[Database] Update {title_raw} rss_link to {rss_set}.")

    def update_poster(self, title_raw, poster_link: str):
        statement = select(Bangumi).where(Bangumi.title_raw == title_raw)
        bangumi = self.session.exec(statement).first()
        bangumi.poster_link = poster_link
        self.session.add(bangumi)
        self.session.commit()
        self.session.refresh(bangumi)
        logger.debug(f"[Database] Update {title_raw} poster_link to {poster_link}.")

    def delete_one(self, _id: int):
        statement = select(Bangumi).where(Bangumi.id == _id)
        bangumi = self.session.exec(statement).first()
        self.session.delete(bangumi)
        self.session.commit()
        logger.debug(f"[Database] Delete bangumi id: {_id}.")

    def delete_all(self):
        statement = delete(Bangumi)
        self.session.exec(statement)
        self.session.commit()

    def search_all(self) -> list[Bangumi]:
        statement = select(Bangumi)
        return self.session.exec(statement).all()

    def search_id(self, _id: int) -> Optional[Bangumi]:
        statement = select(Bangumi).where(Bangumi.id == _id)
        bangumi = self.session.exec(statement).first()
        if bangumi is None:
            logger.warning(f"[Database] Cannot find bangumi id: {_id}.")
            return None
        else:
            logger.debug(f"[Database] Find bangumi id: {_id}.")
            return self.session.exec(statement).first()

    def match_poster(self, bangumi_name: str) -> str:
        # Use like to match
        statement = select(Bangumi).where(
            func.instr(bangumi_name, Bangumi.official_title) > 0
        )
        data = self.session.exec(statement).first()
        if data:
            return data.poster_link
        else:
            return ""

    def match_list(self, torrent_list: list, rss_link: str) -> list:
        match_datas = self.search_all()
        if not match_datas:
            return torrent_list
        # Match title
        i = 0
        while i < len(torrent_list):
            torrent = torrent_list[i]
            for match_data in match_datas:
                if match_data.title_raw in torrent.name:
                    if rss_link not in match_data.rss_link:
                        match_data.rss_link += f",{rss_link}"
                        self.update_rss(match_data.title_raw, match_data.rss_link)
                    # if not match_data.poster_link:
                    #     self.update_poster(match_data.title_raw, torrent.poster_link)
                    torrent_list.pop(i)
                    break
            else:
                i += 1
        return torrent_list

    def match_torrent(self, torrent_name: str) -> Optional[Bangumi]:
        statement = select(Bangumi).where(
            and_(
                func.instr(torrent_name, Bangumi.title_raw) > 0,
                # use `false()` to avoid E712 checking
                # see: https://docs.astral.sh/ruff/rules/true-false-comparison/
                Bangumi.deleted == false(),
            )
        )
        return self.session.exec(statement).first()

    def not_complete(self) -> list[Bangumi]:
        # Find eps_complete = False
        # use `false()` to avoid E712 checking
        # see: https://docs.astral.sh/ruff/rules/true-false-comparison/
        condition = select(Bangumi).where(Bangumi.eps_collect == false())
        datas = self.session.exec(condition).all()
        return datas

    def not_added(self) -> list[Bangumi]:
        conditions = select(Bangumi).where(
            or_(
                Bangumi.added == 0, Bangumi.rule_name is None, Bangumi.save_path is None
            )
        )
        datas = self.session.exec(conditions).all()
        return datas

    def disable_rule(self, _id: int):
        statement = select(Bangumi).where(Bangumi.id == _id)
        bangumi = self.session.exec(statement).first()
        bangumi.deleted = True
        self.session.add(bangumi)
        self.session.commit()
        self.session.refresh(bangumi)
        logger.debug(f"[Database] Disable rule {bangumi.title_raw}.")

    def search_rss(self, rss_link: str) -> list[Bangumi]:
        statement = select(Bangumi).where(func.instr(rss_link, Bangumi.rss_link) > 0)
        return self.session.exec(statement).all()

backend/src/module/database/bangumi.py
maikirakiwi/Goto_Bangumi
from sqlmodel import Session, SQLModel

from module.models import Bangumi, User

from .bangumi import BangumiDatabase
from .engine import engine as e
from .rss import RSSDatabase
from .torrent import TorrentDatabase
from .user import UserDatabase


class Database(Session):
    def __init__(self, engine=e):
        self.engine = engine
        super().__init__(engine)
        self.rss = RSSDatabase(self)
        self.torrent = TorrentDatabase(self)
        self.bangumi = BangumiDatabase(self)
        self.user = UserDatabase(self)

    def create_table(self):
        SQLModel.metadata.create_all(self.engine)

    def drop_table(self):
        SQLModel.metadata.drop_all(self.engine)

    def migrate(self):
        # Run migration online
        bangumi_data = self.bangumi.search_all()
        user_data = self.exec("SELECT * FROM user").all()
        readd_bangumi = []
        for bangumi in bangumi_data:
            dict_data = bangumi.dict()
            del dict_data["id"]
            readd_bangumi.append(Bangumi(**dict_data))
        self.drop_table()
        self.create_table()
        self.commit()
        bangumi_data = self.bangumi.search_all()
        self.bangumi.add_all(readd_bangumi)
        self.add(User(**user_data[0]))
        self.commit()

backend/src/module/database/combine.py
maikirakiwi/Goto_Bangumi
from sqlmodel import Session, create_engine

from module.conf import DATA_PATH

engine = create_engine(DATA_PATH)

db_session = Session(engine)

backend/src/module/database/engine.py
maikirakiwi/Goto_Bangumi
import logging

from sqlmodel import Session, and_, delete, select

from module.models import RSSItem, RSSUpdate

logger = logging.getLogger(__name__)


class RSSDatabase:
    def __init__(self, session: Session):
        self.session = session

    def add(self, data: RSSItem):
        # Check if exists
        statement = select(RSSItem).where(RSSItem.url == data.url)
        db_data = self.session.exec(statement).first()
        if db_data:
            logger.debug(f"RSS Item {data.url} already exists.")
            return False
        else:
            logger.debug(f"RSS Item {data.url} not exists, adding...")
            self.session.add(data)
            self.session.commit()
            self.session.refresh(data)
            return True

    def add_all(self, data: list[RSSItem]):
        for item in data:
            self.add(item)

    def update(self, _id: int, data: RSSUpdate):
        # Check if exists
        statement = select(RSSItem).where(RSSItem.id == _id)
        db_data = self.session.exec(statement).first()
        if not db_data:
            return False
        # Update
        dict_data = data.dict(exclude_unset=True)
        for key, value in dict_data.items():
            setattr(db_data, key, value)
        self.session.add(db_data)
        self.session.commit()
        self.session.refresh(db_data)
        return True

    def enable(self, _id: int):
        statement = select(RSSItem).where(RSSItem.id == _id)
        db_data = self.session.exec(statement).first()
        if not db_data:
            return False
        db_data.enabled = True
        self.session.add(db_data)
        self.session.commit()
        self.session.refresh(db_data)
        return True

    def disable(self, _id: int):
        statement = select(RSSItem).where(RSSItem.id == _id)
        db_data = self.session.exec(statement).first()
        if not db_data:
            return False
        db_data.enabled = False
        self.session.add(db_data)
        self.session.commit()
        self.session.refresh(db_data)
        return True

    def search_id(self, _id: int) -> RSSItem:
        return self.session.get(RSSItem, _id)

    def search_all(self) -> list[RSSItem]:
        return self.session.exec(select(RSSItem)).all()

    def search_active(self) -> list[RSSItem]:
        return self.session.exec(select(RSSItem).where(RSSItem.enabled)).all()

    def search_aggregate(self) -> list[RSSItem]:
        return self.session.exec(
            select(RSSItem).where(and_(RSSItem.aggregate, RSSItem.enabled))
        ).all()

    def delete(self, _id: int) -> bool:
        condition = delete(RSSItem).where(RSSItem.id == _id)
        try:
            self.session.exec(condition)
            self.session.commit()
            return True
        except Exception as e:
            logger.error(f"Delete RSS Item failed. Because: {e}")
            return False

    def delete_all(self):
        condition = delete(RSSItem)
        self.session.exec(condition)
        self.session.commit()

backend/src/module/database/rss.py
maikirakiwi/Goto_Bangumi
import logging

from sqlmodel import Session, select

from module.models import Torrent

logger = logging.getLogger(__name__)


class TorrentDatabase:
    def __init__(self, session: Session):
        self.session = session

    def add(self, data: Torrent):
        self.session.add(data)
        self.session.commit()
        self.session.refresh(data)
        logger.debug(f"Insert {data.name} in database.")

    def add_all(self, datas: list[Torrent]):
        self.session.add_all(datas)
        self.session.commit()
        logger.debug(f"Insert {len(datas)} torrents in database.")

    def update(self, data: Torrent):
        self.session.add(data)
        self.session.commit()
        self.session.refresh(data)
        logger.debug(f"Update {data.name} in database.")

    def update_all(self, datas: list[Torrent]):
        self.session.add_all(datas)
        self.session.commit()

    def update_one_user(self, data: Torrent):
        self.session.add(data)
        self.session.commit()
        self.session.refresh(data)
        logger.debug(f"Update {data.name} in database.")

    def search(self, _id: int) -> Torrent:
        return self.session.exec(select(Torrent).where(Torrent.id == _id)).first()

    def search_all(self) -> list[Torrent]:
        return self.session.exec(select(Torrent)).all()

    def search_rss(self, rss_id: int) -> list[Torrent]:
        return self.session.exec(select(Torrent).where(Torrent.rss_id == rss_id)).all()

    def check_new(self, torrents_list: list[Torrent]) -> list[Torrent]:
        new_torrents = []
        old_torrents = self.search_all()
        old_urls = [t.url for t in old_torrents]
        for torrent in torrents_list:
            if torrent.url not in old_urls:
                new_torrents.append(torrent)
        return new_torrents

backend/src/module/database/torrent.py
maikirakiwi/Goto_Bangumi
import logging

from fastapi import HTTPException
from sqlmodel import Session, select

from module.models import ResponseModel
from module.models.user import User, UserLogin, UserUpdate
from module.security.jwt import get_password_hash, verify_password

logger = logging.getLogger(__name__)


class UserDatabase:
    def __init__(self, session: Session):
        self.session = session

    def get_user(self, username):
        statement = select(User).where(User.username == username)
        result = self.session.exec(statement).first()
        if not result:
            raise HTTPException(status_code=404, detail="User not found")
        return result

    def auth_user(self, user: User):
        statement = select(User).where(User.username == user.username)
        result = self.session.exec(statement).first()
        if not result:
            return ResponseModel(
                status_code=401, status=False, msg_en="User not found", msg_zh="用户不存在"
            )
        if not verify_password(user.password, result.password):
            return ResponseModel(
                status_code=401,
                status=False,
                msg_en="Incorrect password",
                msg_zh="密码错误",
            )
        return ResponseModel(
            status_code=200, status=True, msg_en="Login successfully", msg_zh="登录成功"
        )

    def update_user(self, username, update_user: UserUpdate):
        # Update username and password
        statement = select(User).where(User.username == username)
        result = self.session.exec(statement).first()
        if not result:
            raise HTTPException(status_code=404, detail="User not found")
        if update_user.username:
            result.username = update_user.username
        if update_user.password:
            result.password = get_password_hash(update_user.password)
        self.session.add(result)
        self.session.commit()
        return result

    def merge_old_user(self):
        # get old data
        statement = """
        SELECT * FROM user
        """
        result = self.session.exec(statement).first()
        if not result:
            return
        # add new data
        user = User(username=result.username, password=result.password)
        # Drop old table
        statement = """
        DROP TABLE user
        """
        self.session.exec(statement)
        # Create new table
        statement = """
        CREATE TABLE user (
            id INTEGER NOT NULL PRIMARY KEY,
            username VARCHAR NOT NULL,
            password VARCHAR NOT NULL
        )
        """
        self.session.exec(statement)
        self.session.add(user)
        self.session.commit()

    def add_default_user(self):
        # Check if user exists
        statement = select(User)
        try:
            result = self.session.exec(statement).all()
        except Exception:
            self.merge_old_user()
            result = self.session.exec(statement).all()
        if len(result) != 0:
            return
        # Add default user
        user = User(username="admin", password=get_password_hash("adminadmin"))
        self.session.add(user)
        self.session.commit()

backend/src/module/database/user.py
maikirakiwi/Goto_Bangumi
from .download_client import DownloadClient

backend/src/module/downloader/__init__.py
maikirakiwi/Goto_Bangumi
import logging
import time

from aria2p import API, Client, ClientException

from module.conf import settings

logger = logging.getLogger(__name__)


class QbDownloader:
    def __init__(self, host, username, password):
        while True:
            try:
                self._client = API(Client(host=host, port=6800, secret=password))
                break
            except ClientException:
                logger.warning(
                    f"Can't login Aria2 Server {host} by {username}, retry in {settings.connect_retry_interval}"
                )
            time.sleep(settings.connect_retry_interval)

    def torrents_add(self, urls, save_path, category):
        return self._client.add_torrent(
            is_paused=settings.dev_debug,
            torrent_file_path=urls,
            save_path=save_path,
            category=category,
        )

backend/src/module/downloader/client/aria2_downloader.py
maikirakiwi/Goto_Bangumi
import logging
import time

from qbittorrentapi import Client, LoginFailed
from qbittorrentapi.exceptions import (
    APIConnectionError,
    Conflict409Error,
    Forbidden403Error,
)

from module.ab_decorator import qb_connect_failed_wait

logger = logging.getLogger(__name__)


class QbDownloader:
    def __init__(self, host: str, username: str, password: str, ssl: bool):
        self._client: Client = Client(
            host=host,
            username=username,
            password=password,
            VERIFY_WEBUI_CERTIFICATE=ssl,
            DISABLE_LOGGING_DEBUG_OUTPUT=True,
            REQUESTS_ARGS={"timeout": (3.1, 10)},
        )
        self.host = host
        self.username = username

    def auth(self, retry=3):
        times = 0
        while times < retry:
            try:
                self._client.auth_log_in()
                return True
            except LoginFailed:
                logger.error(
                    f"Can't login qBittorrent Server {self.host} by {self.username}, retry in {5} seconds."
                )
                time.sleep(5)
                times += 1
            except Forbidden403Error:
                logger.error("Login refused by qBittorrent Server")
                logger.info("Please release the IP in qBittorrent Server")
                break
            except APIConnectionError:
                logger.error("Cannot connect to qBittorrent Server")
                logger.info("Please check the IP and port in WebUI settings")
                time.sleep(10)
                times += 1
            except Exception as e:
                logger.error(f"Unknown error: {e}")
                break
        return False

    def logout(self):
        self._client.auth_log_out()

    def check_host(self):
        try:
            self._client.app_version()
            return True
        except APIConnectionError:
            return False

    def check_rss(self, rss_link: str):
        pass

    @qb_connect_failed_wait
    def prefs_init(self, prefs):
        return self._client.app_set_preferences(prefs=prefs)

    @qb_connect_failed_wait
    def get_app_prefs(self):
        return self._client.app_preferences()

    def add_category(self, category):
        return self._client.torrents_createCategory(name=category)

    @qb_connect_failed_wait
    def torrents_info(self, status_filter, category, tag=None):
        return self._client.torrents_info(
            status_filter=status_filter, category=category, tag=tag
        )

    def add_torrents(self, torrent_urls, torrent_files, save_path, category):
        resp = self._client.torrents_add(
            is_paused=False,
            urls=torrent_urls,
            torrent_files=torrent_files,
            save_path=save_path,
            category=category,
            use_auto_torrent_management=False,
        )
        return resp == "Ok."

    def torrents_delete(self, hash):
        return self._client.torrents_delete(delete_files=True, torrent_hashes=hash)

    def torrents_rename_file(self, torrent_hash, old_path, new_path) -> bool:
        try:
            self._client.torrents_rename_file(
                torrent_hash=torrent_hash, old_path=old_path, new_path=new_path
            )
            return True
        except Conflict409Error:
            logger.debug(f"Conflict409Error: {old_path} >> {new_path}")
            return False

    def rss_add_feed(self, url, item_path):
        try:
            self._client.rss_add_feed(url, item_path)
        except Conflict409Error:
            logger.warning(f"[Downloader] RSS feed {url} already exists")

    def rss_remove_item(self, item_path):
        try:
            self._client.rss_remove_item(item_path)
        except Conflict409Error:
            logger.warning(f"[Downloader] RSS item {item_path} does not exist")

    def rss_get_feeds(self):
        return self._client.rss_items()

    def rss_set_rule(self, rule_name, rule_def):
        self._client.rss_set_rule(rule_name, rule_def)

    def move_torrent(self, hashes, new_location):
        self._client.torrents_set_location(new_location, hashes)

    def get_download_rule(self):
        return self._client.rss_rules()

    def get_torrent_path(self, _hash):
        return self._client.torrents_info(hashes=_hash)[0].save_path

    def set_category(self, _hash, category):
        try:
            self._client.torrents_set_category(category, hashes=_hash)
        except Conflict409Error:
            logger.warning(f"[Downloader] Category {category} does not exist")
            self.add_category(category)
            self._client.torrents_set_category(category, hashes=_hash)

    def check_connection(self):
        return self._client.app_version()

    def remove_rule(self, rule_name):
        self._client.rss_remove_rule(rule_name)

    def add_tag(self, _hash, tag):
        self._client.torrents_add_tags(tags=tag, hashes=_hash)

backend/src/module/downloader/client/qb_downloader.py
maikirakiwi/Goto_Bangumi
import logging

from module.conf import settings
from module.models import Bangumi, Torrent
from module.network import RequestContent

from .path import TorrentPath

logger = logging.getLogger(__name__)


class DownloadClient(TorrentPath):
    def __init__(self):
        super().__init__()
        self.client = self.__getClient()
        self.authed = False

    @staticmethod
    def __getClient():
        # TODO 多下载器支持
        type = settings.downloader.type
        host = settings.downloader.host
        username = settings.downloader.username
        password = settings.downloader.password
        ssl = settings.downloader.ssl
        if type == "qbittorrent":
            from .client.qb_downloader import QbDownloader

            return QbDownloader(host, username, password, ssl)
        else:
            logger.error(f"[Downloader] Unsupported downloader type: {type}")
            raise Exception(f"Unsupported downloader type: {type}")

    def __enter__(self):
        if not self.authed:
            self.auth()
        else:
            logger.error("[Downloader] Already authed.")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.authed:
            self.client.logout()
            self.authed = False

    def auth(self):
        self.authed = self.client.auth()
        if self.authed:
            logger.debug("[Downloader] Authed.")
        else:
            logger.error("[Downloader] Auth failed.")

    def check_host(self):
        return self.client.check_host()

    def init_downloader(self):
        prefs = {
            "rss_auto_downloading_enabled": True,
            "rss_max_articles_per_feed": 500,
            "rss_processing_enabled": True,
            "rss_refresh_interval": 30,
        }
        self.client.prefs_init(prefs=prefs)
        try:
            self.client.add_category("BangumiCollection")
        except Exception:
            logger.debug("[Downloader] Cannot add new category, maybe already exists.")
        if settings.downloader.path == "":
            prefs = self.client.get_app_prefs()
            settings.downloader.path = self._join_path(prefs["save_path"], "Bangumi")

    def set_rule(self, data: Bangumi):
        data.rule_name = self._rule_name(data)
        data.save_path = self._gen_save_path(data)
        rule = {
            "enable": True,
            "mustContain": data.title_raw,
            "mustNotContain": "|".join(data.filter),
            "useRegex": True,
            "episodeFilter": "",
            "smartFilter": False,
            "previouslyMatchedEpisodes": [],
            "affectedFeeds": data.rss_link,
            "ignoreDays": 0,
            "lastMatch": "",
            "addPaused": False,
            "assignedCategory": "Bangumi",
            "savePath": data.save_path,
        }
        self.client.rss_set_rule(rule_name=data.rule_name, rule_def=rule)
        data.added = True
        logger.info(
            f"[Downloader] Add {data.official_title} Season {data.season} to auto download rules."
        )

    def set_rules(self, bangumi_info: list[Bangumi]):
        logger.debug("[Downloader] Start adding rules.")
        for info in bangumi_info:
            self.set_rule(info)
        logger.debug("[Downloader] Finished.")

    def get_torrent_info(self, category="Bangumi", status_filter="completed", tag=None):
        return self.client.torrents_info(
            status_filter=status_filter, category=category, tag=tag
        )

    def rename_torrent_file(self, _hash, old_path, new_path) -> bool:
        logger.info(f"{old_path} >> {new_path}")
        return self.client.torrents_rename_file(
            torrent_hash=_hash, old_path=old_path, new_path=new_path
        )

    def delete_torrent(self, hashes):
        self.client.torrents_delete(hashes)
        logger.info("[Downloader] Remove torrents.")

    def add_torrent(self, torrent: Torrent | list, bangumi: Bangumi) -> bool:
        if not bangumi.save_path:
            bangumi.save_path = self._gen_save_path(bangumi)
        with RequestContent() as req:
            if isinstance(torrent, list):
                if len(torrent) == 0:
                    logger.debug(f"[Downloader] No torrent found: {bangumi.official_title}")
                    return False
                if "magnet" in torrent[0].url:
                    torrent_url = [t.url for t in torrent]
                    torrent_file = None
                else:
                    torrent_file = [req.get_content(t.url) for t in torrent]
                    torrent_url = None
            else:
                if "magnet" in torrent.url:
                    torrent_url = torrent.url
                    torrent_file = None
                else:
                    torrent_file = req.get_content(torrent.url)
                    torrent_url = None
        if self.client.add_torrents(
            torrent_urls=torrent_url,
            torrent_files=torrent_file,
            save_path=bangumi.save_path,
            category="Bangumi",
        ):
            logger.debug(f"[Downloader] Add torrent: {bangumi.official_title}")
            return True
        else:
            logger.debug(f"[Downloader] Torrent added before: {bangumi.official_title}")
            return False

    def move_torrent(self, hashes, location):
        self.client.move_torrent(hashes=hashes, new_location=location)

    # RSS Parts
    def add_rss_feed(self, rss_link, item_path="Mikan_RSS"):
        self.client.rss_add_feed(url=rss_link, item_path=item_path)

    def remove_rss_feed(self, item_path):
        self.client.rss_remove_item(item_path=item_path)

    def get_rss_feed(self):
        return self.client.rss_get_feeds()

    def get_download_rules(self):
        return self.client.get_download_rule()

    def get_torrent_path(self, hashes):
        return self.client.get_torrent_path(hashes)

    def set_category(self, hashes, category):
        self.client.set_category(hashes, category)

    def remove_rule(self, rule_name):
        self.client.remove_rule(rule_name)
        logger.info(f"[Downloader] Delete rule: {rule_name}")

backend/src/module/downloader/download_client.py
maikirakiwi/Goto_Bangumi
import logging
import re
from os import PathLike

from module.conf import PLATFORM, settings
from module.models import Bangumi, BangumiUpdate

logger = logging.getLogger(__name__)

if PLATFORM == "Windows":
    from pathlib import PureWindowsPath as Path
else:
    from pathlib import Path


class TorrentPath:
    def __init__(self):
        pass

    @staticmethod
    def check_files(info):
        media_list = []
        subtitle_list = []
        for f in info.files:
            file_name = f.name
            suffix = Path(file_name).suffix
            if suffix.lower() in [".mp4", ".mkv"]:
                media_list.append(file_name)
            elif suffix.lower() in [".ass", ".srt"]:
                subtitle_list.append(file_name)
        return media_list, subtitle_list

    @staticmethod
    def _path_to_bangumi(save_path: PathLike[str] | str):
        # Split save path and download path
        save_parts = Path(save_path).parts
        download_parts = Path(settings.downloader.path).parts
        # Get bangumi name and season
        bangumi_name = ""
        season = 1
        for part in save_parts:
            if re.match(r"S\d+|[Ss]eason \d+", part):
                season = int(re.findall(r"\d+", part)[0])
            elif part not in download_parts:
                bangumi_name = part
        return bangumi_name, season

    @staticmethod
    def _file_depth(file_path: PathLike[str] | str):
        return len(Path(file_path).parts)

    def is_ep(self, file_path: PathLike[str] | str):
        return self._file_depth(file_path) <= 2

    @staticmethod
    def _gen_save_path(data: Bangumi | BangumiUpdate):
        folder = (
            f"{data.official_title} ({data.year})" if data.year else data.official_title
        )
        save_path = Path(settings.downloader.path) / folder / f"Season {data.season}"
        return str(save_path)

    @staticmethod
    def _rule_name(data: Bangumi):
        rule_name = (
            f"[{data.group_name}] {data.official_title} S{data.season}"
            if settings.bangumi_manage.group_tag
            else f"{data.official_title} S{data.season}"
        )
        return rule_name

    @staticmethod
    def _join_path(*args):
        return str(Path(*args))

backend/src/module/downloader/path.py
maikirakiwi/Goto_Bangumi
from .collector import SeasonCollector, eps_complete
from .renamer import Renamer
from .torrent import TorrentManager

backend/src/module/manager/__init__.py
maikirakiwi/Goto_Bangumi
import logging

from module.downloader import DownloadClient
from module.models import Bangumi, ResponseModel
from module.rss import RSSEngine
from module.searcher import SearchTorrent

logger = logging.getLogger(__name__)


class SeasonCollector(DownloadClient):
    def collect_season(self, bangumi: Bangumi, link: str = None):
        logger.info(
            f"Start collecting {bangumi.official_title} Season {bangumi.season}..."
        )
        with SearchTorrent() as st, RSSEngine() as engine:
            if not link:
                torrents = st.search_season(bangumi)
            else:
                torrents = st.get_torrents(link, bangumi.filter.replace(",", "|"))
            if self.add_torrent(torrents, bangumi):
                logger.info(
                    f"Collections of {bangumi.official_title} Season {bangumi.season} completed."
                )
                for torrent in torrents:
                    torrent.downloaded = True
                bangumi.eps_collect = True
                if engine.bangumi.update(bangumi):
                    engine.bangumi.add(bangumi)
                engine.torrent.add_all(torrents)
                return ResponseModel(
                    status=True,
                    status_code=200,
                    msg_en=f"Collections of {bangumi.official_title} Season {bangumi.season} completed.",
                    msg_zh=f"收集 {bangumi.official_title} 第 {bangumi.season} 季完成。",
                )
            else:
                logger.warning(
                    f"Already collected {bangumi.official_title} Season {bangumi.season}."
                )
                return ResponseModel(
                    status=False,
                    status_code=406,
                    msg_en=f"Collection of {bangumi.official_title} Season {bangumi.season} failed.",
                    msg_zh=f"收集 {bangumi.official_title} 第 {bangumi.season} 季失败, 种子已经添加。",
                )

    @staticmethod
    def subscribe_season(data: Bangumi):
        with RSSEngine() as engine:
            data.added = True
            data.eps_collect = True
            engine.add_rss(
                rss_link=data.rss_link, name=data.official_title, aggregate=False
            )
            engine.bangumi.add(data)
            return engine.download_bangumi(data)


def eps_complete():
    with RSSEngine() as engine:
        datas = engine.bangumi.not_complete()
        if datas:
            logger.info("Start collecting full season...")
            for data in datas:
                if not data.eps_collect:
                    with SeasonCollector() as collector:
                        collector.collect_season(data)
                data.eps_collect = True
            engine.bangumi.update_all(datas)

backend/src/module/manager/collector.py
maikirakiwi/Goto_Bangumi
import logging
import re

from module.conf import settings
from module.downloader import DownloadClient
from module.models import EpisodeFile, Notification, SubtitleFile
from module.parser import TitleParser

logger = logging.getLogger(__name__)


class Renamer(DownloadClient):
    def __init__(self):
        super().__init__()
        self._parser = TitleParser()
        self.check_pool = {}

    @staticmethod
    def print_result(torrent_count, rename_count):
        if rename_count != 0:
            logger.info(
                f"Finished checking {torrent_count} files' name, renamed {rename_count} files."
            )
        logger.debug(f"Checked {torrent_count} files")

    @staticmethod
    def gen_path(
            file_info: EpisodeFile | SubtitleFile, bangumi_name: str, method: str
    ) -> str:
        season = f"0{file_info.season}" if file_info.season < 10 else file_info.season
        episode = (
            f"0{file_info.episode}" if file_info.episode < 10 else file_info.episode
        )
        if method == "none" or method == "subtitle_none":
            return file_info.media_path
        elif method == "pn":
            return f"{file_info.title} S{season}E{episode}{file_info.suffix}"
        elif method == "advance":
            return f"{bangumi_name} S{season}E{episode}{file_info.suffix}"
        elif method == "normal":
            logger.warning("[Renamer] Normal rename method is deprecated.")
            return file_info.media_path
        elif method == "subtitle_pn":
            return f"{file_info.title} S{season}E{episode}.{file_info.language}{file_info.suffix}"
        elif method == "subtitle_advance":
            return f"{bangumi_name} S{season}E{episode}.{file_info.language}{file_info.suffix}"
        else:
            logger.error(f"[Renamer] Unknown rename method: {method}")
            return file_info.media_path

    def rename_file(
            self,
            torrent_name: str,
            media_path: str,
            bangumi_name: str,
            method: str,
            season: int,
            _hash: str,
            **kwargs,
    ):
        ep = self._parser.torrent_parser(
            torrent_name=torrent_name,
            torrent_path=media_path,
            season=season,
        )
        if ep:
            new_path = self.gen_path(ep, bangumi_name, method=method)
            if media_path != new_path:
                if new_path not in self.check_pool.keys():
                    if self.rename_torrent_file(
                        _hash=_hash, old_path=media_path, new_path=new_path
                    ):
                        return Notification(
                            official_title=bangumi_name,
                            season=ep.season,
                            episode=ep.episode,
                        )
        else:
            logger.warning(f"[Renamer] {media_path} parse failed")
            if settings.bangumi_manage.remove_bad_torrent:
                self.delete_torrent(hashes=_hash)
        return None

    def rename_collection(
            self,
            media_list: list[str],
            bangumi_name: str,
            season: int,
            method: str,
            _hash: str,
            **kwargs,
    ):
        for media_path in media_list:
            if self.is_ep(media_path):
                ep = self._parser.torrent_parser(
                    torrent_path=media_path,
                    season=season,
                )
                if ep:
                    new_path = self.gen_path(ep, bangumi_name, method=method)
                    if media_path != new_path:
                        renamed = self.rename_torrent_file(
                            _hash=_hash, old_path=media_path, new_path=new_path
                        )
                        if not renamed:
                            logger.warning(f"[Renamer] {media_path} rename failed")
                            # Delete bad torrent.
                            if settings.bangumi_manage.remove_bad_torrent:
                                self.delete_torrent(_hash)
                                break

    def rename_subtitles(
            self,
            subtitle_list: list[str],
            torrent_name: str,
            bangumi_name: str,
            season: int,
            method: str,
            _hash,
            **kwargs,
    ):
        method = "subtitle_" + method
        for subtitle_path in subtitle_list:
            sub = self._parser.torrent_parser(
                torrent_path=subtitle_path,
                torrent_name=torrent_name,
                season=season,
                file_type="subtitle",
            )
            if sub:
                new_path = self.gen_path(sub, bangumi_name, method=method)
                if subtitle_path != new_path:
                    renamed = self.rename_torrent_file(
                        _hash=_hash, old_path=subtitle_path, new_path=new_path
                    )
                    if not renamed:
                        logger.warning(f"[Renamer] {subtitle_path} rename failed")

    def rename(self) -> list[Notification]:
        # Get torrent info
        logger.debug("[Renamer] Start rename process.")
        rename_method = settings.bangumi_manage.rename_method
        torrents_info = self.get_torrent_info()
        renamed_info: list[Notification] = []
        for info in torrents_info:
            media_list, subtitle_list = self.check_files(info)
            bangumi_name, season = self._path_to_bangumi(info.save_path)
            kwargs = {
                "torrent_name": info.name,
                "bangumi_name": bangumi_name,
                "method": rename_method,
                "season": season,
                "_hash": info.hash,
            }
            # Rename single media file
            if len(media_list) == 1:
                notify_info = self.rename_file(media_path=media_list[0], **kwargs)
                if notify_info:
                    renamed_info.append(notify_info)
                # Rename subtitle file
                if len(subtitle_list) > 0:
                    self.rename_subtitles(subtitle_list=subtitle_list, **kwargs)
            # Rename collection
            elif len(media_list) > 1:
                logger.info("[Renamer] Start rename collection")
                self.rename_collection(media_list=media_list, **kwargs)
                if len(subtitle_list) > 0:
                    self.rename_subtitles(subtitle_list=subtitle_list, **kwargs)
                self.set_category(info.hash, "BangumiCollection")
            else:
                logger.warning(f"[Renamer] {info.name} has no media file")
        logger.debug("[Renamer] Rename process finished.")
        return renamed_info

    def compare_ep_version(self, torrent_name: str, torrent_hash: str):
        if re.search(r"v\d.", torrent_name):
            pass
        else:
            self.delete_torrent(hashes=torrent_hash)


if __name__ == "__main__":
    from module.conf import setup_logger

    settings.log.debug_enable = True
    setup_logger()
    with Renamer() as renamer:
        renamer.rename()

backend/src/module/manager/renamer.py
maikirakiwi/Goto_Bangumi
import logging

from module.database import Database
from module.downloader import DownloadClient
from module.models import Bangumi, BangumiUpdate, ResponseModel
from module.parser import TitleParser

logger = logging.getLogger(__name__)


class TorrentManager(Database):
    @staticmethod
    def __match_torrents_list(data: Bangumi | BangumiUpdate) -> list:
        with DownloadClient() as client:
            torrents = client.get_torrent_info(status_filter=None)
        return [
            torrent.hash for torrent in torrents if torrent.save_path == data.save_path
        ]

    def delete_torrents(self, data: Bangumi, client: DownloadClient):
        hash_list = self.__match_torrents_list(data)
        if hash_list:
            client.delete_torrent(hash_list)
            logger.info(f"Delete rule and torrents for {data.official_title}")
            return ResponseModel(
                status_code=200,
                status=True,
                msg_en=f"Delete rule and torrents for {data.official_title}",
                msg_zh=f"删除 {data.official_title} 规则和种子",
            )
        else:
            return ResponseModel(
                status_code=406,
                status=False,
                msg_en=f"Can't find torrents for {data.official_title}",
                msg_zh=f"无法找到 {data.official_title} 的种子",
            )

    def delete_rule(self, _id: int | str, file: bool = False):
        data = self.bangumi.search_id(int(_id))
        if isinstance(data, Bangumi):
            with DownloadClient() as client:
                # client.remove_rule(data.rule_name)
                # client.remove_rss_feed(data.official_title)
                self.rss.delete(data.official_title)
                self.bangumi.delete_one(int(_id))
                if file:
                    torrent_message = self.delete_torrents(data, client)
                    return torrent_message
                logger.info(f"[Manager] Delete rule for {data.official_title}")
                return ResponseModel(
                    status_code=200,
                    status=True,
                    msg_en=f"Delete rule for {data.official_title}",
                    msg_zh=f"删除 {data.official_title} 规则",
                )
        else:
            return ResponseModel(
                status_code=406,
                status=False,
                msg_en=f"Can't find id {_id}",
                msg_zh=f"无法找到 id {_id}",
            )

    def disable_rule(self, _id: str | int, file: bool = False):
        data = self.bangumi.search_id(int(_id))
        if isinstance(data, Bangumi):
            with DownloadClient() as client:
                # client.remove_rule(data.rule_name)
                data.deleted = True
                self.bangumi.update(data)
                if file:
                    torrent_message = self.delete_torrents(data, client)
                    return torrent_message
                logger.info(f"[Manager] Disable rule for {data.official_title}")
                return ResponseModel(
                    status_code=200,
                    status=True,
                    msg_en=f"Disable rule for {data.official_title}",
                    msg_zh=f"禁用 {data.official_title} 规则",
                )
        else:
            return ResponseModel(
                status_code=406,
                status=False,
                msg_en=f"Can't find id {_id}",
                msg_zh=f"无法找到 id {_id}",
            )

    def enable_rule(self, _id: str | int):
        data = self.bangumi.search_id(int(_id))
        if data:
            data.deleted = False
            self.bangumi.update(data)
            logger.info(f"[Manager] Enable rule for {data.official_title}")
            return ResponseModel(
                status_code=200,
                status=True,
                msg_en=f"Enable rule for {data.official_title}",
                msg_zh=f"启用 {data.official_title} 规则",
            )
        else:
            return ResponseModel(
                status_code=406,
                status=False,
                msg_en=f"Can't find id {_id}",
                msg_zh=f"无法找到 id {_id}",
            )

    def update_rule(self, bangumi_id, data: BangumiUpdate):
        old_data: Bangumi = self.bangumi.search_id(bangumi_id)
        if old_data:
            # Move torrent
            match_list = self.__match_torrents_list(old_data)
            with DownloadClient() as client:
                path = client._gen_save_path(data)
                if match_list:
                    client.move_torrent(match_list, path)
            data.save_path = path
            self.bangumi.update(data, bangumi_id)
            return ResponseModel(
                status_code=200,
                status=True,
                msg_en=f"Update rule for {data.official_title}",
                msg_zh=f"更新 {data.official_title} 规则",
            )
        else:
            logger.error(f"[Manager] Can't find data with {bangumi_id}")
            return ResponseModel(
                status_code=406,
                status=False,
                msg_en=f"Can't find data with {bangumi_id}",
                msg_zh=f"无法找到 id {bangumi_id} 的数据",
            )

    def refresh_poster(self):
        bangumis = self.bangumi.search_all()
        for bangumi in bangumis:
            if not bangumi.poster_link:
                TitleParser().tmdb_poster_parser(bangumi)
        self.bangumi.update_all(bangumis)
        return ResponseModel(
            status_code=200,
            status=True,
            msg_en="Refresh poster link successfully.",
            msg_zh="刷新海报链接成功。",
        )

    def search_all_bangumi(self):
        datas = self.bangumi.search_all()
        if not datas:
            return []
        return [data for data in datas if not data.deleted]

    def search_one(self, _id: int | str):
        data = self.bangumi.search_id(int(_id))
        if not data:
            logger.error(f"[Manager] Can't find data with {_id}")
            return ResponseModel(
                status_code=406,
                status=False,
                msg_en=f"Can't find data with {_id}",
                msg_zh=f"无法找到 id {_id} 的数据",
            )
        else:
            return data


if __name__ == "__main__":
    with TorrentManager() as manager:
        manager.refresh_poster()

backend/src/module/manager/torrent.py
maikirakiwi/Goto_Bangumi
from .bangumi import Bangumi, BangumiUpdate, Episode, Notification
from .config import Config
from .response import APIResponse, ResponseModel
from .rss import RSSItem, RSSUpdate
from .torrent import EpisodeFile, SubtitleFile, Torrent, TorrentUpdate
from .user import User, UserLogin, UserUpdate

backend/src/module/models/__init__.py
maikirakiwi/Goto_Bangumi
from pydantic import BaseModel


class RssLink(BaseModel):
    rss_link: str


class AddRule(BaseModel):
    title: str
    season: int


class ChangeConfig(BaseModel):
    config: dict


class ChangeRule(BaseModel):
    rule: dict

backend/src/module/models/api.py
maikirakiwi/Goto_Bangumi
from dataclasses import dataclass
from typing import Optional

from pydantic import BaseModel
from sqlmodel import Field, SQLModel


class Bangumi(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True)
    official_title: str = Field(
        default="official_title", alias="official_title", title="番剧中文名"
    )
    year: Optional[str] = Field(alias="year", title="番剧年份")
    title_raw: str = Field(default="title_raw", alias="title_raw", title="番剧原名")
    season: int = Field(default=1, alias="season", title="番剧季度")
    season_raw: Optional[str] = Field(alias="season_raw", title="番剧季度原名")
    group_name: Optional[str] = Field(alias="group_name", title="字幕组")
    dpi: Optional[str] = Field(alias="dpi", title="分辨率")
    source: Optional[str] = Field(alias="source", title="来源")
    subtitle: Optional[str] = Field(alias="subtitle", title="字幕")
    eps_collect: bool = Field(default=False, alias="eps_collect", title="是否已收集")
    offset: int = Field(default=0, alias="offset", title="番剧偏移量")
    filter: str = Field(default="720,\\d+-\\d+", alias="filter", title="番剧过滤器")
    rss_link: str = Field(default="", alias="rss_link", title="番剧RSS链接")
    poster_link: Optional[str] = Field(alias="poster_link", title="番剧海报链接")
    added: bool = Field(default=False, alias="added", title="是否已添加")
    rule_name: Optional[str] = Field(alias="rule_name", title="番剧规则名")
    save_path: Optional[str] = Field(alias="save_path", title="番剧保存路径")
    deleted: bool = Field(False, alias="deleted", title="是否已删除")


class BangumiUpdate(SQLModel):
    official_title: str = Field(
        default="official_title", alias="official_title", title="番剧中文名"
    )
    year: Optional[str] = Field(alias="year", title="番剧年份")
    title_raw: str = Field(default="title_raw", alias="title_raw", title="番剧原名")
    season: int = Field(default=1, alias="season", title="番剧季度")
    season_raw: Optional[str] = Field(alias="season_raw", title="番剧季度原名")
    group_name: Optional[str] = Field(alias="group_name", title="字幕组")
    dpi: Optional[str] = Field(alias="dpi", title="分辨率")
    source: Optional[str] = Field(alias="source", title="来源")
    subtitle: Optional[str] = Field(alias="subtitle", title="字幕")
    eps_collect: bool = Field(default=False, alias="eps_collect", title="是否已收集")
    offset: int = Field(default=0, alias="offset", title="番剧偏移量")
    filter: str = Field(default="720,\\d+-\\d+", alias="filter", title="番剧过滤器")
    rss_link: str = Field(default="", alias="rss_link", title="番剧RSS链接")
    poster_link: Optional[str] = Field(alias="poster_link", title="番剧海报链接")
    added: bool = Field(default=False, alias="added", title="是否已添加")
    rule_name: Optional[str] = Field(alias="rule_name", title="番剧规则名")
    save_path: Optional[str] = Field(alias="save_path", title="番剧保存路径")
    deleted: bool = Field(False, alias="deleted", title="是否已删除")


class Notification(BaseModel):
    official_title: str = Field(..., alias="official_title", title="番剧名")
    season: int = Field(..., alias="season", title="番剧季度")
    episode: int = Field(..., alias="episode", title="番剧集数")
    poster_path: Optional[str] = Field(None, alias="poster_path", title="番剧海报路径")


@dataclass
class Episode:
    title_en: Optional[str]
    title_zh: Optional[str]
    title_jp: Optional[str]
    season: int
    season_raw: str
    episode: int
    sub: str
    group: str
    resolution: str
    source: str


@dataclass
class SeasonInfo(dict):
    official_title: str
    title_raw: str
    season: int
    season_raw: str
    group: str
    filter: list | None
    offset: int | None
    dpi: str
    source: str
    subtitle: str
    added: bool
    eps_collect: bool

backend/src/module/models/bangumi.py
maikirakiwi/Goto_Bangumi
from os.path import expandvars

from pydantic import BaseModel, Field


class Program(BaseModel):
    rss_time: int = Field(900, description="Sleep time")
    rename_time: int = Field(60, description="Rename times in one loop")
    webui_port: int = Field(7892, description="WebUI port")


class Downloader(BaseModel):
    type: str = Field("qbittorrent", description="Downloader type")
    host_: str = Field("172.17.0.1:8080", alias="host", description="Downloader host")
    username_: str = Field("admin", alias="username", description="Downloader username")
    password_: str = Field(
        "adminadmin", alias="password", description="Downloader password"
    )
    path: str = Field("/downloads/Bangumi", description="Downloader path")
    ssl: bool = Field(False, description="Downloader ssl")

    @property
    def host(self):
        return expandvars(self.host_)

    @property
    def username(self):
        return expandvars(self.username_)

    @property
    def password(self):
        return expandvars(self.password_)


class RSSParser(BaseModel):
    enable: bool = Field(True, description="Enable RSS parser")
    filter: list[str] = Field(["720", r"\d+-\d"], description="Filter")
    language: str = "zh"


class BangumiManage(BaseModel):
    enable: bool = Field(True, description="Enable bangumi manage")
    eps_complete: bool = Field(False, description="Enable eps complete")
    rename_method: str = Field("pn", description="Rename method")
    group_tag: bool = Field(False, description="Enable group tag")
    remove_bad_torrent: bool = Field(False, description="Remove bad torrent")


class Log(BaseModel):
    debug_enable: bool = Field(False, description="Enable debug")


class Proxy(BaseModel):
    enable: bool = Field(False, description="Enable proxy")
    type: str = Field("http", description="Proxy type")
    host: str = Field("", description="Proxy host")
    port: int = Field(0, description="Proxy port")
    username_: str = Field("", alias="username", description="Proxy username")
    password_: str = Field("", alias="password", description="Proxy password")

    @property
    def username(self):
        return expandvars(self.username_)

    @property
    def password(self):
        return expandvars(self.password_)


class Notification(BaseModel):
    enable: bool = Field(False, description="Enable notification")
    type: str = Field("telegram", description="Notification type")
    token_: str = Field("", alias="token", description="Notification token")
    chat_id_: str = Field("", alias="chat_id", description="Notification chat id")

    @property
    def token(self):
        return expandvars(self.token_)

    @property
    def chat_id(self):
        return expandvars(self.chat_id_)


class ExperimentalOpenAI(BaseModel):
    enable: bool = Field(False, description="Enable experimental OpenAI")
    api_key: str = Field("", description="OpenAI api key")
    api_base: str = Field(
        "https://api.openai.com/v1", description="OpenAI api base url"
    )
    model: str = Field("gpt-3.5-turbo", description="OpenAI model")


class Config(BaseModel):
    program: Program = Program()
    downloader: Downloader = Downloader()
    rss_parser: RSSParser = RSSParser()
    bangumi_manage: BangumiManage = BangumiManage()
    log: Log = Log()
    proxy: Proxy = Proxy()
    notification: Notification = Notification()
    experimental_openai: ExperimentalOpenAI = ExperimentalOpenAI()

    def dict(self, *args, by_alias=True, **kwargs):
        return super().dict(*args, by_alias=by_alias, **kwargs)

backend/src/module/models/config.py
maikirakiwi/Goto_Bangumi
from pydantic import BaseModel, Field


class ResponseModel(BaseModel):
    status: bool = Field(..., example=True)
    status_code: int = Field(..., example=200)
    msg_en: str
    msg_zh: str


class APIResponse(BaseModel):
    status: bool = Field(..., example=True)
    msg_en: str = Field(..., example="Success")
    msg_zh: str = Field(..., example="成功")
backend/src/module/models/response.py
maikirakiwi/Goto_Bangumi
from typing import Optional

from sqlmodel import Field, SQLModel


class RSSItem(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True, alias="id")
    name: Optional[str] = Field(None, alias="name")
    url: str = Field("https://mikanani.me", alias="url")
    aggregate: bool = Field(False, alias="aggregate")
    parser: str = Field("mikan", alias="parser")
    enabled: bool = Field(True, alias="enabled")


class RSSUpdate(SQLModel):
    name: Optional[str] = Field(None, alias="name")
    url: Optional[str] = Field("https://mikanani.me", alias="url")
    aggregate: Optional[bool] = Field(True, alias="aggregate")
    parser: Optional[str] = Field("mikan", alias="parser")
    enabled: Optional[bool] = Field(True, alias="enabled")

backend/src/module/models/rss.py
maikirakiwi/Goto_Bangumi
from typing import Optional

from pydantic import BaseModel
from sqlmodel import Field, SQLModel


class Torrent(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True, alias="id")
    bangumi_id: Optional[int] = Field(None, alias="refer_id", foreign_key="bangumi.id")
    rss_id: Optional[int] = Field(None, alias="rss_id", foreign_key="rssitem.id")
    name: str = Field("", alias="name")
    url: str = Field("https://example.com/torrent", alias="url")
    homepage: Optional[str] = Field(None, alias="homepage")
    downloaded: bool = Field(False, alias="downloaded")


class TorrentUpdate(SQLModel):
    downloaded: bool = Field(False, alias="downloaded")


class EpisodeFile(BaseModel):
    media_path: str = Field(...)
    group: str | None = Field(None)
    title: str = Field(...)
    season: int = Field(...)
    episode: int = Field(None)
    suffix: str = Field(..., regex=r"\.(mkv|mp4|MKV|MP4)$")


class SubtitleFile(BaseModel):
    media_path: str = Field(...)
    group: str | None = Field(None)
    title: str = Field(...)
    season: int = Field(...)
    episode: int = Field(None)
    language: str = Field(..., regex=r"(zh|zh-tw)")
    suffix: str = Field(..., regex=r"\.(ass|srt|ASS|SRT)$")

backend/src/module/models/torrent.py
maikirakiwi/Goto_Bangumi
from typing import Optional

from pydantic import BaseModel
from sqlmodel import Field, SQLModel


class User(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True)
    username: str = Field(
        "admin", min_length=4, max_length=20, regex=r"^[a-zA-Z0-9_]+$"
    )
    password: str = Field("adminadmin", min_length=8)


class UserUpdate(SQLModel):
    username: Optional[str] = Field(
        None, min_length=4, max_length=20, regex=r"^[a-zA-Z0-9_]+$"
    )
    password: Optional[str] = Field(None, min_length=8)


class UserLogin(SQLModel):
    username: str
    password: str = Field(..., min_length=8)


class Token(BaseModel):
    token: str
    token_type: str


class TokenData(BaseModel):
    username: str | None = None

backend/src/module/models/user.py
maikirakiwi/Goto_Bangumi
from .request_contents import RequestContent

backend/src/module/network/__init__.py
maikirakiwi/Goto_Bangumi
import logging
import re
import xml.etree.ElementTree

from module.conf import settings
from module.models import Torrent

from .request_url import RequestURL
from .site import rss_parser

logger = logging.getLogger(__name__)


class RequestContent(RequestURL):
    def get_torrents(
        self,
        _url: str,
        _filter: str = "|".join(settings.rss_parser.filter),
        limit: int = None,
        retry: int = 3,
    ) -> list[Torrent]:
        soup = self.get_xml(_url, retry)
        if soup:
            torrent_titles, torrent_urls, torrent_homepage = rss_parser(soup)
            torrents: list[Torrent] = []
            for _title, torrent_url, homepage in zip(
                torrent_titles, torrent_urls, torrent_homepage
            ):
                if re.search(_filter, _title) is None:
                    torrents.append(
                        Torrent(name=_title, url=torrent_url, homepage=homepage)
                    )
                if isinstance(limit, int):
                    if len(torrents) >= limit:
                        break
            return torrents
        else:
            logger.warning(f"[Network] Failed to get torrents: {_url}")
            return []

    def get_xml(self, _url, retry: int = 3) -> xml.etree.ElementTree.Element:
        req = self.get_url(_url, retry)
        if req:
            return xml.etree.ElementTree.fromstring(req.text)

    # API JSON
    def get_json(self, _url) -> dict:
        req = self.get_url(_url)
        if req:
            return req.json()

    def post_json(self, _url, data: dict) -> dict:
        return self.post_url(_url, data).json()

    def post_data(self, _url, data: dict) -> dict:
        return self.post_url(_url, data)

    def post_files(self, _url, data: dict, files: dict) -> dict:
        return self.post_form(_url, data, files)

    def get_html(self, _url):
        return self.get_url(_url).text

    def get_content(self, _url):
        req = self.get_url(_url)
        if req:
            return req.content

    def check_connection(self, _url):
        return self.check_url(_url)

    def get_rss_title(self, _url):
        soup = self.get_xml(_url)
        if soup:
            return soup.find("./channel/title").text

backend/src/module/network/request_contents.py
maikirakiwi/Goto_Bangumi
import logging
import socket
import time

import requests
import socks

from module.conf import settings

logger = logging.getLogger(__name__)


class RequestURL:
    def __init__(self):
        self.header = {"user-agent": "Mozilla/5.0", "Accept": "application/xml"}
        self._socks5_proxy = False

    def get_url(self, url, retry=3):
        try_time = 0
        while True:
            try:
                req = self.session.get(url=url, headers=self.header, timeout=5)
                req.raise_for_status()
                return req
            except requests.RequestException:
                logger.warning(
                    f"[Network] Cannot connect to {url}. Wait for 5 seconds."
                )
                try_time += 1
                if try_time >= retry:
                    break
                time.sleep(5)
            except Exception as e:
                logger.debug(e)
                break
        logger.error(f"[Network] Failed connecting to {url}")
        logger.warning("[Network] Please check DNS/Connection settings")
        return None

    def post_url(self, url: str, data: dict, retry=3):
        try_time = 0
        while True:
            try:
                req = self.session.post(
                    url=url, headers=self.header, data=data, timeout=5
                )
                req.raise_for_status()
                return req
            except requests.RequestException:
                logger.warning(
                    f"[Network] Cannot connect to {url}. Wait for 5 seconds."
                )
                try_time += 1
                if try_time >= retry:
                    break
                time.sleep(5)
            except Exception as e:
                logger.debug(e)
                break
        logger.error(f"[Network] Failed connecting to {url}")
        logger.warning("[Network] Please check DNS/Connection settings")
        return None

    def check_url(self, url: str):
        if "://" not in url:
            url = f"http://{url}"
        try:
            req = requests.head(url=url, headers=self.header, timeout=5)
            req.raise_for_status()
            return True
        except requests.RequestException:
            logger.debug(f"[Network] Cannot connect to {url}.")
            return False

    def post_form(self, url: str, data: dict, files):
        try:
            req = self.session.post(
                url=url, headers=self.header, data=data, files=files, timeout=5
            )
            req.raise_for_status()
            return req
        except requests.RequestException:
            logger.warning(f"[Network] Cannot connect to {url}.")
            return None

    def __enter__(self):
        self.session = requests.Session()
        if settings.proxy.enable:
            if "http" in settings.proxy.type:
                url = f"{settings.proxy.type}://{settings.proxy.host}:{settings.proxy.port}"
                self.session.proxies = {
                    "https": url,
                    "http": url,
                }
            elif settings.proxy.type == "socks5":
                self._socks5_proxy = True
                socks.set_default_proxy(
                    socks.SOCKS5,
                    addr=settings.proxy.host,
                    port=settings.proxy.port,
                    rdns=True,
                    username=settings.proxy.username,
                    password=settings.proxy.password,
                )
                socket.socket = socks.socksocket
            else:
                logger.error(f"[Network] Unsupported proxy type: {settings.proxy.type}")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self._socks5_proxy:
            socks.set_default_proxy()
            socket.socket = socks.socksocket
            self._socks5_proxy = False
        self.session.close()

backend/src/module/network/request_url.py
maikirakiwi/Goto_Bangumi
from .mikan import rss_parser

backend/src/module/network/site/__init__.py
maikirakiwi/Goto_Bangumi
def rss_parser(soup):
    torrent_titles = []
    torrent_urls = []
    torrent_homepage = []
    for item in soup.findall("./channel/item"):
        torrent_titles.append(item.find("title").text)
        torrent_urls.append(item.find("enclosure").attrib["url"])
        torrent_homepage.append(item.find("link").text)
    return torrent_titles, torrent_urls, torrent_homepage


def mikan_title(soup):
    return soup.find("title").text

backend/src/module/network/site/mikan.py
maikirakiwi/Goto_Bangumi
from .notification import PostNotification

backend/src/module/notification/__init__.py
maikirakiwi/Goto_Bangumi
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(notify: Notification):
        with Database() as db:
            poster_path = db.bangumi.match_poster(notify.official_title)
        notify.poster_path = poster_path

    def send_msg(self, notify: Notification) -> bool:
        self._get_poster(notify)
        try:
            self.notifier.post_msg(notify)
            logger.debug(f"Send notification: {notify.official_title}")
        except Exception as e:
            logger.warning(f"Failed to send notification: {e}")
            return False

    def __enter__(self):
        self.notifier.__enter__()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.notifier.__exit__(exc_type, exc_val, exc_tb)

backend/src/module/notification/notification.py
maikirakiwi/Goto_Bangumi
from .bark import BarkNotification
from .server_chan import ServerChanNotification
from .telegram import TelegramNotification
from .wecom import WecomNotification

backend/src/module/notification/plugin/__init__.py
maikirakiwi/Goto_Bangumi
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()

    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 = self.post_data(self.notification_url, data)
        logger.debug(f"Bark notification: {resp.status_code}")
        return resp.status_code == 200

backend/src/module/notification/plugin/bark.py
maikirakiwi/Goto_Bangumi
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()

    def post_msg(self, notify: Notification) -> bool:
        text = self.gen_message(notify)
        data = {
            "title": notify.official_title,
            "desp": text,
        }
        resp = self.post_data(self.notification_url, data)
        logger.debug(f"ServerChan notification: {resp.status_code}")
        return resp.status_code == 200

backend/src/module/notification/plugin/server_chan.py
maikirakiwi/Goto_Bangumi
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()

    def post_msg(self, notify: Notification) -> bool:
        text = self.gen_message(notify)
        data = {"title": notify.official_title, "body": text, "device_key": self.token}
        resp = self.post_data(self.notification_url, data)
        logger.debug(f"Bark notification: {resp.status_code}")
        return resp.status_code == 200

backend/src/module/notification/plugin/slack.py
maikirakiwi/Goto_Bangumi
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()

    def post_msg(self, notify: Notification) -> bool:
        text = self.gen_message(notify)
        data = {
            "chat_id": self.chat_id,
            "caption": text,
            "disable_notification": True,
        }
        photo = load_image(notify.poster_path)
        if photo:
            resp = self.post_files(
                self.photo_url, data, files={"photo": photo}
            )
        else:
            resp = self.post_data(self.message_url, data)
        logger.debug(f"Telegram notification: {resp.status_code}")
        return resp.status_code == 200

backend/src/module/notification/plugin/telegram.py
maikirakiwi/Goto_Bangumi
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()

    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 = self.post_data(self.notification_url, data)
        logger.debug(f"Wecom notification: {resp.status_code}")
        return resp.status_code == 200

backend/src/module/notification/plugin/wecom.py
maikirakiwi/Goto_Bangumi
from .title_parser import TitleParser

backend/src/module/parser/__init__.py
maikirakiwi/Goto_Bangumi
from .mikan_parser import mikan_parser
from .openai import OpenAIParser
from .raw_parser import raw_parser
from .tmdb_parser import tmdb_parser
from .torrent_parser import torrent_parser

backend/src/module/parser/analyser/__init__.py
maikirakiwi/Goto_Bangumi
from module.network import RequestContent


def search_url(e):
    return f"https://api.bgm.tv/search/subject/{e}?responseGroup=large"


def bgm_parser(title):
    url = search_url(title)
    with RequestContent() as req:
        contents = req.get_json(url)
        if contents:
            return contents[0]
        else:
            return None

backend/src/module/parser/analyser/bgm_parser.py
maikirakiwi/Goto_Bangumi
from bs4 import BeautifulSoup
from urllib3.util import parse_url

from module.network import RequestContent
from module.utils import save_image


def mikan_parser(homepage: str):
    root_path = parse_url(homepage).host
    with RequestContent() as req:
        content = req.get_html(homepage)
        soup = BeautifulSoup(content, "html.parser")
        poster_div = soup.find("div", {"class": "bangumi-poster"}).get("style")
        official_title = soup.select_one(
            'p.bangumi-title a[href^="/Home/Bangumi/"]'
        ).text
        if poster_div:
            poster_path = poster_div.split("url('")[1].split("')")[0]
            img = req.get_content(f"https://{root_path}{poster_path}")
            suffix = poster_path.split(".")[-1]
            poster_link = save_image(img, suffix)
            return poster_link, official_title
        return "", ""

backend/src/module/parser/analyser/mikan_parser.py
maikirakiwi/Goto_Bangumi
import asyncio
import json
import logging

import openai

logger = logging.getLogger(__name__)

DEFAULT_PROMPT = """\
You will now play the role of a super assistant. 
Your task is to extract structured data from unstructured text content and output it in JSON format. 
If you are unable to extract any information, please keep all fields and leave the field empty or default value like `''`, `None`.
But Do not fabricate data!

the python structured data type is:

```python
@dataclass
class Episode:
    title_en: Optional[str]
    title_zh: Optional[str]
    title_jp: Optional[str]
    season: int
    season_raw: str
    episode: int
    sub: str
    group: str
    resolution: str
    source: str
```

Example:

```
input: "【喵萌奶茶屋】★04月新番★[夏日重现/Summer Time Rendering][11][1080p][繁日双语][招募翻译]"
output: '{"group": "喵萌奶茶屋", "title_en": "Summer Time Rendering", "resolution": "1080p", "episode": 11, "season": 1, "title_zh": "夏日重现", "sub": "", "title_jp": "", "season_raw": "", "source": ""}'

input: "【幻樱字幕组】【4月新番】【古见同学有交流障碍症 第二季 Komi-san wa, Komyushou Desu. S02】【22】【GB_MP4】【1920X1080】"
output: '{"group": "幻樱字幕组", "title_en": "Komi-san wa, Komyushou Desu.", "resolution": "1920X1080", "episode": 22, "season": 2, "title_zh": "古见同学有交流障碍症", "sub": "", "title_jp": "", "season_raw": "", "source": ""}'

input: "[Lilith-Raws] 关于我在无意间被隔壁的天使变成废柴这件事 / Otonari no Tenshi-sama - 09 [Baha][WEB-DL][1080p][AVC AAC][CHT][MP4]"
output: '{"group": "Lilith-Raws", "title_en": "Otonari no Tenshi-sama", "resolution": "1080p", "episode": 9, "season": 1, "source": "WEB-DL", "title_zh": "关于我在无意间被隔壁的天使变成废柴这件事", "sub": "CHT", "title_jp": ""}'
```
"""


class OpenAIParser:
    def __init__(
        self,
        api_key: str,
        api_base: str = "https://api.openai.com/v1",
        model: str = "gpt-3.5-turbo",
        **kwargs,
    ) -> None:
        """OpenAIParser is a class to parse text with openai

        Args:
            api_key (str): the OpenAI api key
            api_base (str):
                the OpenAI api base url, you can use custom url here. \
                Defaults to "https://api.openai.com/v1".
            model (str):
                the ChatGPT model parameter, you can get more details from \
                https://platform.openai.com/docs/api-reference/chat/create. \
                Defaults to "gpt-3.5-turbo".
            kwargs (dict):
                the OpenAI ChatGPT parameters, you can get more details from \
                https://platform.openai.com/docs/api-reference/chat/create.

        Raises:
            ValueError: if api_key is not provided.
        """
        if not api_key:
            raise ValueError("API key is required.")

        self._api_key = api_key
        self.api_base = api_base
        self.model = model
        self.openai_kwargs = kwargs

    def parse(
        self, text: str, prompt: str | None = None, asdict: bool = True
    ) -> dict | str:
        """parse text with openai

        Args:
            text (str): the text to be parsed
            prompt (str | None, optional):
                the custom prompt. Built-in prompt will be used if no prompt is provided. \
                Defaults to None.
            asdict (bool, optional):
                whether to return the result as dict or not. \
                Defaults to True.

        Returns:
            dict | str: the parsed result.
        """
        if not prompt:
            prompt = DEFAULT_PROMPT

        async def complete() -> str:
            resp = await openai.ChatCompletion.acreate(
                api_key=self._api_key,
                api_base=self.api_base,
                model=self.model,
                messages=[
                    dict(role="system", content=prompt),
                    dict(role="user", content=text),
                ],
                # set temperature to 0 to make results be more stable and reproducible.
                temperature=0,
                **self.openai_kwargs,
            )

            result = resp["choices"][0]["message"]["content"]
            return result

        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(complete())

        if asdict:
            try:
                result = json.loads(result)
            except json.JSONDecodeError:
                logger.warning(f"Cannot parse result {result} as python dict.")

        logger.debug(f"the parsed result is: {result}")

        return result

backend/src/module/parser/analyser/openai.py
maikirakiwi/Goto_Bangumi
import logging
import re

from module.models import Episode

logger = logging.getLogger(__name__)

EPISODE_RE = re.compile(r"\d+")
TITLE_RE = re.compile(
    r"(.*|\[.*])( -? \d+|\[\d+]|\[\d+.?[vV]\d]|第\d+[话話集]|\[第?\d+[话話集]]|\[\d+.?END]|[Ee][Pp]?\d+)(.*)"
)
RESOLUTION_RE = re.compile(r"1080|720|2160|4K")
SOURCE_RE = re.compile(r"B-Global|[Bb]aha|[Bb]ilibili|AT-X|Web")
SUB_RE = re.compile(r"[简繁日字幕]|CH|BIG5|GB")

PREFIX_RE = re.compile(r"[^\w\s\u4e00-\u9fff\u3040-\u309f\u30a0-\u30ff-]")

CHINESE_NUMBER_MAP = {
    "一": 1,
    "二": 2,
    "三": 3,
    "四": 4,
    "五": 5,
    "六": 6,
    "七": 7,
    "八": 8,
    "九": 9,
    "十": 10,
}


def get_group(name: str) -> str:
    return re.split(r"[\[\]]", name)[1]


def pre_process(raw_name: str) -> str:
    return raw_name.replace("【", "[").replace("】", "]")


def prefix_process(raw: str, group: str) -> str:
    raw = re.sub(f".{group}.", "", raw)
    raw_process = PREFIX_RE.sub("/", raw)
    arg_group = raw_process.split("/")
    while "" in arg_group:
        arg_group.remove("")
    if len(arg_group) == 1:
        arg_group = arg_group[0].split(" ")
    for arg in arg_group:
        if re.search(r"新番|月?番", arg) and len(arg) <= 5:
            raw = re.sub(f".{arg}.", "", raw)
        elif re.search(r"港澳台地区", arg):
            raw = re.sub(f".{arg}.", "", raw)
    return raw


def season_process(season_info: str):
    name_season = season_info
    # if re.search(r"新番|月?番", season_info):
    #     name_season = re.sub(".*新番.", "", season_info)
    #     # 去除「新番」信息
    # name_season = re.sub(r"^[^]】]*[]】]", "", name_season).strip()
    season_rule = r"S\d{1,2}|Season \d{1,2}|[第].[季期]"
    name_season = re.sub(r"[\[\]]", " ", name_season)
    seasons = re.findall(season_rule, name_season)
    if not seasons:
        return name_season, "", 1
    name = re.sub(season_rule, "", name_season)
    for season in seasons:
        season_raw = season
        if re.search(r"Season|S", season) is not None:
            season = int(re.sub(r"Season|S", "", season))
            break
        elif re.search(r"[第 ].*[季期(部分)]|部分", season) is not None:
            season_pro = re.sub(r"[第季期 ]", "", season)
            try:
                season = int(season_pro)
            except ValueError:
                season = CHINESE_NUMBER_MAP[season_pro]
                break
    return name, season_raw, season


def name_process(name: str):
    name_en, name_zh, name_jp = None, None, None
    name = name.strip()
    name = re.sub(r"[(（]仅限港澳台地区[）)]", "", name)
    split = re.split(r"/|\s{2}|-\s{2}", name)
    while "" in split:
        split.remove("")
    if len(split) == 1:
        if re.search("_{1}", name) is not None:
            split = re.split("_", name)
        elif re.search(" - {1}", name) is not None:
            split = re.split("-", name)
    if len(split) == 1:
        split_space = split[0].split(" ")
        for idx in [0, -1]:
            if re.search(r"^[\u4e00-\u9fa5]{2,}", split_space[idx]) is not None:
                chs = split_space[idx]
                split_space.remove(chs)
                split = [chs, " ".join(split_space)]
                break
    for item in split:
        if re.search(r"[\u0800-\u4e00]{2,}", item) and not name_jp:
            name_jp = item.strip()
        elif re.search(r"[\u4e00-\u9fa5]{2,}", item) and not name_zh:
            name_zh = item.strip()
        elif re.search(r"[a-zA-Z]{3,}", item) and not name_en:
            name_en = item.strip()
    return name_en, name_zh, name_jp


def find_tags(other):
    elements = re.sub(r"[\[\]()（）]", " ", other).split(" ")
    # find CHT
    sub, resolution, source = None, None, None
    for element in filter(lambda x: x != "", elements):
        if SUB_RE.search(element):
            sub = element
        elif RESOLUTION_RE.search(element):
            resolution = element
        elif SOURCE_RE.search(element):
            source = element
    return clean_sub(sub), resolution, source


def clean_sub(sub: str | None) -> str | None:
    if sub is None:
        return sub
    return re.sub(r"_MP4|_MKV", "", sub)


def process(raw_title: str):
    raw_title = raw_title.strip()
    content_title = pre_process(raw_title)
    # 预处理标题
    group = get_group(content_title)
    # 翻译组的名字
    match_obj = TITLE_RE.match(content_title)
    # 处理标题
    season_info, episode_info, other = list(
        map(lambda x: x.strip(), match_obj.groups())
    )
    process_raw = prefix_process(season_info, group)
    # 处理 前缀
    raw_name, season_raw, season = season_process(process_raw)
    # 处理 第n季
    name_en, name_zh, name_jp = "", "", ""
    try:
        name_en, name_zh, name_jp = name_process(raw_name)
        # 处理 名字
    except ValueError:
        pass
    # 处理 集数
    raw_episode = EPISODE_RE.search(episode_info)
    episode = 0
    if raw_episode is not None:
        episode = int(raw_episode.group())
    sub, dpi, source = find_tags(other)  # 剩余信息处理
    return (
        name_en,
        name_zh,
        name_jp,
        season,
        season_raw,
        episode,
        sub,
        dpi,
        source,
        group,
    )


def raw_parser(raw: str) -> Episode | None:
    ret = process(raw)
    if ret is None:
        logger.error(f"Parser cannot analyse {raw}")
        return None
    name_en, name_zh, name_jp, season, sr, episode, sub, dpi, source, group = ret
    return Episode(
        name_en, name_zh, name_jp, season, sr, episode, sub, group, dpi, source
    )


if __name__ == "__main__":
    title = "[动漫国字幕组&LoliHouse] THE MARGINAL SERVICE - 08 [WebRip 1080p HEVC-10bit AAC][简繁内封字幕]"
    print(raw_parser(title))

backend/src/module/parser/analyser/raw_parser.py
maikirakiwi/Goto_Bangumi
import re
import time
from dataclasses import dataclass

from module.conf import TMDB_API
from module.network import RequestContent
from module.utils import save_image

TMDB_URL = "https://api.themoviedb.org"


@dataclass
class TMDBInfo:
    id: int
    title: str
    original_title: str
    season: list[dict]
    last_season: int
    year: str
    poster_link: str = None


LANGUAGE = {"zh": "zh-CN", "jp": "ja-JP", "en": "en-US"}


def search_url(e):
    return f"{TMDB_URL}/3/search/tv?api_key={TMDB_API}&page=1&query={e}&include_adult=false"


def info_url(e, key):
    return f"{TMDB_URL}/3/tv/{e}?api_key={TMDB_API}&language={LANGUAGE[key]}"


def is_animation(tv_id, language) -> bool:
    url_info = info_url(tv_id, language)
    with RequestContent() as req:
        type_id = req.get_json(url_info)["genres"]
        for type in type_id:
            if type.get("id") == 16:
                return True
    return False


def get_season(seasons: list) -> tuple[int, str]:
    ss = [s for s in seasons if s["air_date"] is not None and "特别" not in s["season"]]
    ss = sorted(ss, key=lambda e: e.get("air_date"), reverse=True)
    for season in ss:
        if re.search(r"第 \d 季", season.get("season")) is not None:
            date = season.get("air_date").split("-")
            [year, _, _] = date
            now_year = time.localtime().tm_year
            if int(year) <= now_year:
                return int(re.findall(r"\d", season.get("season"))[0]), season.get(
                    "poster_path"
                )
    return len(ss), ss[-1].get("poster_path")


def tmdb_parser(title, language, test: bool = False) -> TMDBInfo | None:
    with RequestContent() as req:
        url = search_url(title)
        contents = req.get_json(url).get("results")
        if contents.__len__() == 0:
            url = search_url(title.replace(" ", ""))
            contents = req.get_json(url).get("results")
        # 判断动画
        if contents:
            for content in contents:
                id = content["id"]
                if is_animation(id, language):
                    break
            url_info = info_url(id, language)
            info_content = req.get_json(url_info)
            season = [
                {
                    "season": s.get("name"),
                    "air_date": s.get("air_date"),
                    "poster_path": s.get("poster_path"),
                }
                for s in info_content.get("seasons")
            ]
            last_season, poster_path = get_season(season)
            if poster_path is None:
                poster_path = info_content.get("poster_path")
            original_title = info_content.get("original_name")
            official_title = info_content.get("name")
            year_number = info_content.get("first_air_date").split("-")[0]
            if poster_path:
                if not test:
                    img = req.get_content(f"https://image.tmdb.org/t/p/w780{poster_path}")
                    poster_link = save_image(img, "jpg")
                else:
                    poster_link = "https://image.tmdb.org/t/p/w780" + poster_path
            else:
                poster_link = None
            return TMDBInfo(
                id,
                official_title,
                original_title,
                season,
                last_season,
                str(year_number),
                poster_link,
            )
        else:
            return None


if __name__ == "__main__":
    print(tmdb_parser("魔法禁书目录", "zh"))

backend/src/module/parser/analyser/tmdb_parser.py
maikirakiwi/Goto_Bangumi
import logging
import re
from pathlib import Path

from module.models import EpisodeFile, SubtitleFile

logger = logging.getLogger(__name__)

PLATFORM = "Unix"

RULES = [
    r"(.*) - (\d{1,4}(?!\d|p)|\d{1,4}\.\d{1,2}(?!\d|p))(?:v\d{1,2})?(?: )?(?:END)?(.*)",
    r"(.*)[\[\ E](\d{1,4}|\d{1,4}\.\d{1,2})(?:v\d{1,2})?(?: )?(?:END)?[\]\ ](.*)",
    r"(.*)\[(?:第)?(\d*\.*\d*)[话集話](?:END)?\](.*)",
    r"(.*)第?(\d*\.*\d*)[话話集](?:END)?(.*)",
    r"(.*)(?:S\d{2})?EP?(\d+)(.*)",
]

SUBTITLE_LANG = {
    "zh-tw": ["tc", "cht", "繁", "zh-tw"],
    "zh": ["sc", "chs", "简", "zh"],
}


def get_path_basename(torrent_path: str) -> str:
    """
    Returns the basename of a path string.

    :param torrent_path: A string representing a path to a file.
    :type torrent_path: str
    :return: A string representing the basename of the given path.
    :rtype: str
    """
    return Path(torrent_path).name


def get_group(group_and_title) -> tuple[str | None, str]:
    n = re.split(r"[\[\]()【】（）]", group_and_title)
    while "" in n:
        n.remove("")
    if len(n) > 1:
        if re.match(r"\d+", n[1]):
            return None, group_and_title
        return n[0], n[1]
    else:
        return None, n[0]


def get_season_and_title(season_and_title) -> tuple[str, int]:
    title = re.sub(r"([Ss]|Season )\d{1,3}", "", season_and_title).strip()
    try:
        season = re.search(r"([Ss]|Season )(\d{1,3})", season_and_title, re.I).group(2)
    except AttributeError:
        season = 1
    return title, int(season)


def get_subtitle_lang(subtitle_name: str) -> str:
    for key, value in SUBTITLE_LANG.items():
        for v in value:
            if v in subtitle_name.lower():
                return key


def torrent_parser(
    torrent_path: str,
    torrent_name: str | None = None,
    season: int | None = None,
    file_type: str = "media",
) -> EpisodeFile | SubtitleFile:
    media_path = get_path_basename(torrent_path)
    for rule in RULES:
        if torrent_name:
            match_obj = re.match(rule, torrent_name, re.I)
        else:
            match_obj = re.match(rule, media_path, re.I)
        if match_obj:
            group, title = get_group(match_obj.group(1))
            if not season:
                title, season = get_season_and_title(title)
            else:
                title, _ = get_season_and_title(title)
            episode = int(match_obj.group(2))
            suffix = Path(torrent_path).suffix
            if file_type == "media":
                return EpisodeFile(
                    media_path=torrent_path,
                    group=group,
                    title=title,
                    season=season,
                    episode=episode,
                    suffix=suffix,
                )
            elif file_type == "subtitle":
                language = get_subtitle_lang(media_path)
                return SubtitleFile(
                    media_path=torrent_path,
                    group=group,
                    title=title,
                    season=season,
                    language=language,
                    episode=episode,
                    suffix=suffix,
                )

backend/src/module/parser/analyser/torrent_parser.py
maikirakiwi/Goto_Bangumi
import logging

from module.conf import settings
from module.models import Bangumi
from module.models.bangumi import Episode
from module.parser.analyser import (
    OpenAIParser,
    mikan_parser,
    raw_parser,
    tmdb_parser,
    torrent_parser,
)

logger = logging.getLogger(__name__)


class TitleParser:
    def __init__(self):
        pass

    @staticmethod
    def torrent_parser(
        torrent_path: str,
        torrent_name: str | None = None,
        season: int | None = None,
        file_type: str = "media",
    ):
        try:
            return torrent_parser(torrent_path, torrent_name, season, file_type)
        except Exception as e:
            logger.warning(f"Cannot parse {torrent_path} with error {e}")

    @staticmethod
    def tmdb_parser(title: str, season: int, language: str):
        tmdb_info = tmdb_parser(title, language)
        if tmdb_info:
            logger.debug(f"TMDB Matched, official title is {tmdb_info.title}")
            tmdb_season = tmdb_info.last_season if tmdb_info.last_season else season
            return tmdb_info.title, tmdb_season, tmdb_info.year, tmdb_info.poster_link
        else:
            logger.warning(f"Cannot match {title} in TMDB. Use raw title instead.")
            logger.warning("Please change bangumi info manually.")
            return title, season, None, None

    @staticmethod
    def tmdb_poster_parser(bangumi: Bangumi):
        tmdb_info = tmdb_parser(bangumi.official_title, settings.rss_parser.language)
        if tmdb_info:
            logger.debug(f"TMDB Matched, official title is {tmdb_info.title}")
            bangumi.poster_link = tmdb_info.poster_link
        else:
            logger.warning(
                f"Cannot match {bangumi.official_title} in TMDB. Use raw title instead."
            )
            logger.warning("Please change bangumi info manually.")

    @staticmethod
    def raw_parser(raw: str) -> Bangumi | None:
        language = settings.rss_parser.language
        try:
            # use OpenAI ChatGPT to parse raw title and get structured data
            if settings.experimental_openai.enable:
                kwargs = settings.experimental_openai.dict(exclude={"enable"})
                gpt = OpenAIParser(**kwargs)
                episode_dict = gpt.parse(raw, asdict=True)
                episode = Episode(**episode_dict)
            else:
                episode = raw_parser(raw)

            titles = {
                "zh": episode.title_zh,
                "en": episode.title_en,
                "jp": episode.title_jp,
            }
            title_raw = episode.title_en if episode.title_en else episode.title_zh
            if titles[language]:
                official_title = titles[language]
            elif titles["zh"]:
                official_title = titles["zh"]
            elif titles["en"]:
                official_title = titles["en"]
            elif titles["jp"]:
                official_title = titles["jp"]
            else:
                official_title = title_raw
            _season = episode.season
            logger.debug(f"RAW:{raw} >> {title_raw}")
            return Bangumi(
                official_title=official_title,
                title_raw=title_raw,
                season=_season,
                season_raw=episode.season_raw,
                group_name=episode.group,
                dpi=episode.resolution,
                source=episode.source,
                subtitle=episode.sub,
                eps_collect=False if episode.episode > 1 else True,
                offset=0,
                filter=",".join(settings.rss_parser.filter),
            )
        except Exception as e:
            logger.debug(e)
            logger.warning(f"Cannot parse {raw}.")
            return None

    @staticmethod
    def mikan_parser(homepage: str) -> tuple[str, str]:
        return mikan_parser(homepage)

backend/src/module/parser/title_parser.py
maikirakiwi/Goto_Bangumi
from .analyser import RSSAnalyser
from .engine import RSSEngine

backend/src/module/rss/__init__.py
maikirakiwi/Goto_Bangumi
import logging
import re

from module.conf import settings
from module.models import Bangumi, ResponseModel, RSSItem, Torrent
from module.network import RequestContent
from module.parser import TitleParser

from .engine import RSSEngine

logger = logging.getLogger(__name__)


class RSSAnalyser(TitleParser):
    def official_title_parser(self, bangumi: Bangumi, rss: RSSItem, torrent: Torrent):
        if rss.parser == "mikan":
            try:
                bangumi.poster_link, bangumi.official_title = self.mikan_parser(
                    torrent.homepage
                )
            except AttributeError:
                logger.warning("[Parser] Mikan torrent has no homepage info.")
                pass
        elif rss.parser == "tmdb":
            tmdb_title, season, year, poster_link = self.tmdb_parser(
                bangumi.official_title, bangumi.season, settings.rss_parser.language
            )
            bangumi.official_title = tmdb_title
            bangumi.year = year
            bangumi.season = season
            bangumi.poster_link = poster_link
        else:
            pass
        bangumi.official_title = re.sub(r"[/:.\\]", " ", bangumi.official_title)

    @staticmethod
    def get_rss_torrents(rss_link: str, full_parse: bool = True) -> list[Torrent]:
        with RequestContent() as req:
            if full_parse:
                rss_torrents = req.get_torrents(rss_link)
            else:
                rss_torrents = req.get_torrents(rss_link, "\\d+-\\d+")
        return rss_torrents

    def torrents_to_data(
        self, torrents: list[Torrent], rss: RSSItem, full_parse: bool = True
    ) -> list:
        new_data = []
        for torrent in torrents:
            bangumi = self.raw_parser(raw=torrent.name)
            if bangumi and bangumi.title_raw not in [i.title_raw for i in new_data]:
                self.official_title_parser(bangumi=bangumi, rss=rss, torrent=torrent)
                if not full_parse:
                    return [bangumi]
                new_data.append(bangumi)
                logger.info(f"[RSS] New bangumi founded: {bangumi.official_title}")
        return new_data

    def torrent_to_data(self, torrent: Torrent, rss: RSSItem) -> Bangumi:
        bangumi = self.raw_parser(raw=torrent.name)
        if bangumi:
            self.official_title_parser(bangumi=bangumi, rss=rss, torrent=torrent)
            bangumi.rss_link = rss.url
            return bangumi

    def rss_to_data(
        self, rss: RSSItem, engine: RSSEngine, full_parse: bool = True
    ) -> list[Bangumi]:
        rss_torrents = self.get_rss_torrents(rss.url, full_parse)
        torrents_to_add = engine.bangumi.match_list(rss_torrents, rss.url)
        if not torrents_to_add:
            logger.debug("[RSS] No new title has been found.")
            return []
        # New List
        new_data = self.torrents_to_data(torrents_to_add, rss, full_parse)
        if new_data:
            # Add to database
            engine.bangumi.add_all(new_data)
            return new_data
        else:
            return []

    def link_to_data(self, rss: RSSItem) -> Bangumi | ResponseModel:
        torrents = self.get_rss_torrents(rss.url, False)
        try:
            for torrent in torrents:
                data = self.torrent_to_data(torrent, rss)
                if data:
                    return data
        except Exception as e:
            logger.debug(e)
            return ResponseModel(
                status=False,
                status_code=406,
                msg_en="No new title has been found.",
                msg_zh="没有找到新的番剧。",
            )

backend/src/module/rss/analyser.py
maikirakiwi/Goto_Bangumi
import logging
import re
from typing import Optional

from module.database import Database, engine
from module.downloader import DownloadClient
from module.models import Bangumi, ResponseModel, RSSItem, Torrent
from module.network import RequestContent

logger = logging.getLogger(__name__)


class RSSEngine(Database):
    def __init__(self, _engine=engine):
        super().__init__(_engine)
        self._to_refresh = False

    @staticmethod
    def _get_torrents(rss: RSSItem) -> list[Torrent]:
        with RequestContent() as req:
            torrents = req.get_torrents(rss.url)
            # Add RSS ID
            for torrent in torrents:
                torrent.rss_id = rss.id
        return torrents

    def get_rss_torrents(self, rss_id: int) -> list[Torrent]:
        rss = self.rss.search_id(rss_id)
        if rss:
            return self.torrent.search_rss(rss_id)
        else:
            return []

    def add_rss(
        self,
        rss_link: str,
        name: str | None = None,
        aggregate: bool = True,
        parser: str = "mikan",
    ):
        if not name:
            with RequestContent() as req:
                name = req.get_rss_title(rss_link)
                if not name:
                    return ResponseModel(
                        status=False,
                        status_code=406,
                        msg_en="Failed to get RSS title.",
                        msg_zh="无法获取 RSS 标题。",
                    )
        rss_data = RSSItem(name=name, url=rss_link, aggregate=aggregate, parser=parser)
        if self.rss.add(rss_data):
            return ResponseModel(
                status=True,
                status_code=200,
                msg_en="RSS added successfully.",
                msg_zh="RSS 添加成功。",
            )
        else:
            return ResponseModel(
                status=False,
                status_code=406,
                msg_en="RSS added failed.",
                msg_zh="RSS 添加失败。",
            )

    def disable_list(self, rss_id_list: list[int]):
        for rss_id in rss_id_list:
            self.rss.disable(rss_id)
        return ResponseModel(
            status=True,
            status_code=200,
            msg_en="Disable RSS successfully.",
            msg_zh="禁用 RSS 成功。",
        )

    def enable_list(self, rss_id_list: list[int]):
        for rss_id in rss_id_list:
            self.rss.enable(rss_id)
        return ResponseModel(
            status=True,
            status_code=200,
            msg_en="Enable RSS successfully.",
            msg_zh="启用 RSS 成功。",
        )

    def delete_list(self, rss_id_list: list[int]):
        for rss_id in rss_id_list:
            self.rss.delete(rss_id)
        return ResponseModel(
            status=True,
            status_code=200,
            msg_en="Delete RSS successfully.",
            msg_zh="删除 RSS 成功。",
        )

    def pull_rss(self, rss_item: RSSItem) -> list[Torrent]:
        torrents = self._get_torrents(rss_item)
        new_torrents = self.torrent.check_new(torrents)
        return new_torrents

    def match_torrent(self, torrent: Torrent) -> Optional[Bangumi]:
        matched: Bangumi = self.bangumi.match_torrent(torrent.name)
        if matched:
            if matched.filter == "":
                return matched
            _filter = matched.filter.replace(",", "|")
            if not re.search(_filter, torrent.name, re.IGNORECASE):
                torrent.bangumi_id = matched.id
                return matched
        return None

    def refresh_rss(self, client: DownloadClient, rss_id: Optional[int] = None):
        # Get All RSS Items
        if not rss_id:
            rss_items: list[RSSItem] = self.rss.search_active()
        else:
            rss_item = self.rss.search_id(rss_id)
            rss_items = [rss_item] if rss_item else []
        # From RSS Items, get all torrents
        logger.debug(f"[Engine] Get {len(rss_items)} RSS items")
        for rss_item in rss_items:
            new_torrents = self.pull_rss(rss_item)
            # Get all enabled bangumi data
            for torrent in new_torrents:
                matched_data = self.match_torrent(torrent)
                if matched_data:
                    if client.add_torrent(torrent, matched_data):
                        logger.debug(f"[Engine] Add torrent {torrent.name} to client")
                    torrent.downloaded = True
            # Add all torrents to database
            self.torrent.add_all(new_torrents)

    def download_bangumi(self, bangumi: Bangumi):
        with RequestContent() as req:
            torrents = req.get_torrents(
                bangumi.rss_link, bangumi.filter.replace(",", "|")
            )
            if torrents:
                with DownloadClient() as client:
                    client.add_torrent(torrents, bangumi)
                    self.torrent.add_all(torrents)
                    return ResponseModel(
                        status=True,
                        status_code=200,
                        msg_en=f"[Engine] Download {bangumi.official_title} successfully.",
                        msg_zh=f"下载 {bangumi.official_title} 成功。",
                    )
            else:
                return ResponseModel(
                    status=False,
                    status_code=406,
                    msg_en=f"[Engine] Download {bangumi.official_title} failed.",
                    msg_zh=f"[Engine] 下载 {bangumi.official_title} 失败。",
                )

backend/src/module/rss/engine.py
maikirakiwi/Goto_Bangumi
from .provider import SEARCH_CONFIG
from .searcher import SearchTorrent

backend/src/module/searcher/__init__.py
maikirakiwi/Goto_Bangumi
import re

from module.conf import SEARCH_CONFIG
from module.models import RSSItem


def search_url(site: str, keywords: list[str]) -> RSSItem:
    keyword = "+".join(keywords)
    search_str = re.sub(r"[\W_ ]", "+", keyword)
    if site in SEARCH_CONFIG.keys():
        url = re.sub(r"%s", search_str, SEARCH_CONFIG[site])
        parser = "mikan" if site == "mikan" else "tmdb"
        rss_item = RSSItem(
            url=url,
            aggregate=False,
            parser=parser,
        )
        return rss_item
    else:
        raise ValueError(f"Site {site} is not supported")

backend/src/module/searcher/provider.py
maikirakiwi/Goto_Bangumi
import json
from typing import TypeAlias

from module.models import Bangumi, RSSItem, Torrent
from module.network import RequestContent
from module.rss import RSSAnalyser

from .provider import search_url

SEARCH_KEY = [
    "group_name",
    "title_raw",
    "season_raw",
    "subtitle",
    "source",
    "dpi",
]

BangumiJSON: TypeAlias = str


class SearchTorrent(RequestContent, RSSAnalyser):
    def search_torrents(self, rss_item: RSSItem) -> list[Torrent]:
        torrents = self.get_torrents(rss_item.url)
        return torrents

    def analyse_keyword(
        self, keywords: list[str], site: str = "mikan", limit: int = 5
    ) -> BangumiJSON:
        rss_item = search_url(site, keywords)
        torrents = self.search_torrents(rss_item)
        # yield for EventSourceResponse (Server Send)
        exist_list = []
        for torrent in torrents:
            if len(exist_list) >= limit:
                break
            bangumi = self.torrent_to_data(torrent=torrent, rss=rss_item)
            if bangumi:
                special_link = self.special_url(bangumi, site).url
                if special_link not in exist_list:
                    bangumi.rss_link = special_link
                    exist_list.append(special_link)
                    yield json.dumps(bangumi.dict(), separators=(",", ":"))

    @staticmethod
    def special_url(data: Bangumi, site: str) -> RSSItem:
        keywords = [getattr(data, key) for key in SEARCH_KEY if getattr(data, key)]
        url = search_url(site, keywords)
        return url

    def search_season(self, data: Bangumi, site: str = "mikan") -> list[Torrent]:
        rss_item = self.special_url(data, site)
        torrents = self.search_torrents(rss_item)
        return [torrent for torrent in torrents if data.title_raw in torrent.name]

backend/src/module/searcher/searcher.py
maikirakiwi/Goto_Bangumi
from fastapi import Cookie, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

from module.database import Database
from module.models.user import User, UserUpdate

from .jwt import verify_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

active_user = []


async def get_current_user(token: str = Cookie(None)):
    if not token:
        raise UNAUTHORIZED
    payload = verify_token(token)
    if not payload:
        raise UNAUTHORIZED
    username = payload.get("sub")
    if not username:
        raise UNAUTHORIZED
    if username not in active_user:
        raise UNAUTHORIZED
    return username


async def get_token_data(token: str = Depends(oauth2_scheme)):
    payload = verify_token(token)
    if not payload:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid token"
        )
    return payload


def update_user_info(user_data: UserUpdate, current_user):
    try:
        with Database() as db:
            db.user.update_user(current_user, user_data)
        return True
    except Exception as e:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))


def auth_user(user: User):
    with Database() as db:
        resp = db.user.auth_user(user)
        if resp.status:
            active_user.append(user.username)
        return resp


UNAUTHORIZED = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized"
)

backend/src/module/security/api.py
maikirakiwi/Goto_Bangumi
from datetime import datetime, timedelta

from jose import JWTError, jwt
from passlib.context import CryptContext


def generate_key():
    import secrets

    return secrets.token_urlsafe(32)


app_pwd_key = generate_key()
app_pwd_algorithm = "HS256"

# Hashing 密码
app_pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


# 创建 JWT Token
def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=1440)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, app_pwd_key, algorithm=app_pwd_algorithm)
    return encoded_jwt


# 解码 Token
def decode_token(token: str):
    try:
        payload = jwt.decode(token, app_pwd_key, algorithms=[app_pwd_algorithm])
        username = payload.get("sub")
        if username is None:
            return None
        return payload
    except JWTError:
        return None


def verify_token(token: str):
    token_data = decode_token(token)
    if token_data is None:
        return None
    expires = token_data.get("exp")
    if datetime.utcnow() >= datetime.fromtimestamp(expires):
        raise JWTError("Token expired")
    return token_data


# 密码加密&验证
def verify_password(plain_password, hashed_password):
    return app_pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return app_pwd_context.hash(password)

backend/src/module/security/jwt.py
maikirakiwi/Goto_Bangumi
from .cross_version import from_30_to_31, cache_image
from .data_migration import data_migration
from .startup import first_run, start_up
from .version_check import version_check

backend/src/module/update/__init__.py
maikirakiwi/Goto_Bangumi
import re

from urllib3.util import parse_url

from module.rss import RSSEngine
from module.utils import save_image
from module.network import RequestContent


def from_30_to_31():
    with RSSEngine() as db:
        db.migrate()
        # Update poster link
        bangumis = db.bangumi.search_all()
        rss_pool = []
        for bangumi in bangumis:
            if bangumi.poster_link:
                rss_link = bangumi.rss_link.split(",")[-1]
                if rss_link not in rss_pool and not re.search(
                    r"\d+.\d+.\d+.\d+", rss_link
                ):
                    rss_pool.append(rss_link)
                root_path = parse_url(rss_link).host
                if "://" not in bangumi.poster_link:
                    bangumi.poster_link = f"https://{root_path}{bangumi.poster_link}"
        db.bangumi.update_all(bangumis)
        for rss in rss_pool:
            if "mybangumi" in rss.lower():
                aggregate = True
            else:
                aggregate = False
            db.add_rss(rss_link=rss, aggregate=aggregate)


def cache_image():
    with RSSEngine() as db, RequestContent() as req:
        bangumis = db.bangumi.search_all()
        for bangumi in bangumis:
            if bangumi.poster_link:
                # Hash local path
                img = req.get_content(bangumi.poster_link)
                suffix = bangumi.poster_link.split(".")[-1]
                img_path = save_image(img, suffix)
                bangumi.poster_link = img_path
        db.bangumi.update_all(bangumis)


backend/src/module/update/cross_version.py
maikirakiwi/Goto_Bangumi
from module.conf import LEGACY_DATA_PATH
from module.models import Bangumi
from module.rss import RSSEngine
from module.utils import json_config


def data_migration():
    if not LEGACY_DATA_PATH.exists():
        return False
    old_data = json_config.load(LEGACY_DATA_PATH)
    infos = old_data["bangumi_info"]
    rss_link = old_data["rss_link"]
    new_data = []
    for info in infos:
        new_data.append(Bangumi(**info, rss_link=rss_link))
    with RSSEngine() as engine:
        engine.bangumi.add_all(new_data)
        engine.add_rss(rss_link)
    LEGACY_DATA_PATH.unlink(missing_ok=True)


def database_migration():
    with RSSEngine() as engine:
        engine.migrate()

backend/src/module/update/data_migration.py
maikirakiwi/Goto_Bangumi
from module.rss import RSSEngine


def update_main_rss(rss_link: str):
    with RSSEngine() as engine:
        engine.add_rss(rss_link, "main", True)

backend/src/module/update/rss.py
maikirakiwi/Goto_Bangumi
import logging

from module.rss import RSSEngine

logger = logging.getLogger(__name__)


def start_up():
    with RSSEngine() as engine:
        engine.create_table()
        engine.user.add_default_user()


def first_run():
    with RSSEngine() as engine:
        engine.create_table()
        engine.user.add_default_user()

backend/src/module/update/startup.py
maikirakiwi/Goto_Bangumi
import semver

from module.conf import VERSION, VERSION_PATH


def version_check() -> bool:
    if VERSION == "DEV_VERSION":
        return True
    if VERSION == "local":
        return True
    if not VERSION_PATH.exists():
        with open(VERSION_PATH, "w") as f:
            f.write(VERSION + "\n")
        return False
    else:
        with open(VERSION_PATH, "r+") as f:
            # Read last version
            versions = f.readlines()
            last_version = versions[-1]
            last_ver = semver.VersionInfo.parse(last_version)
            now_ver = semver.VersionInfo.parse(VERSION)
            if now_ver.minor == last_ver.minor:
                return True
            else:
                if now_ver.minor > last_ver.minor:
                    f.write(VERSION + "\n")
                    return False
                else:
                    return True

backend/src/module/update/version_check.py
maikirakiwi/Goto_Bangumi
from .cache_image import save_image, load_image
backend/src/module/utils/__init__.py
maikirakiwi/Goto_Bangumi
import hashlib


def save_image(img, suffix):
    img_hash = hashlib.md5(img).hexdigest()[0:8]
    image_path = f"data/posters/{img_hash}.{suffix}"
    with open(image_path, "wb") as f:
        f.write(img)
    return f"posters/{img_hash}.{suffix}"


def load_image(img_path):
    if img_path:
        with open(f"data/{img_path}", "rb") as f:
            return f.read()
    else:
        return None

backend/src/module/utils/cache_image.py
maikirakiwi/Goto_Bangumi
import json

import requests


def load(filename):
    with open(filename, "r", encoding="utf-8") as f:
        return json.load(f)


def save(filename, obj):
    with open(filename, "w", encoding="utf-8") as f:
        json.dump(obj, f, indent=4, separators=(",", ": "), ensure_ascii=False)
    pass


def get(url):
    req = requests.get(url)
    return req.json()

backend/src/module/utils/json_config.py
maikirakiwi/Goto_Bangumi
from module.database.combine import Database
from module.models import Bangumi, RSSItem, Torrent
from sqlmodel import SQLModel, create_engine
from sqlmodel.pool import StaticPool

# sqlite mock engine
engine = create_engine(
    "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
)


def test_bangumi_database():
    test_data = Bangumi(
        official_title="无职转生，到了异世界就拿出真本事",
        year="2021",
        title_raw="Mushoku Tensei",
        season=1,
        season_raw="",
        group_name="Lilith-Raws",
        dpi="1080p",
        source="Baha",
        subtitle="CHT",
        eps_collect=False,
        offset=0,
        filter="720p,\\d+-\\d+",
        rss_link="test",
        poster_link="/test/test.jpg",
        added=False,
        rule_name=None,
        save_path="downloads/无职转生，到了异世界就拿出真本事/Season 1",
        deleted=False,
    )
    with Database(engine) as db:
        db.create_table()
        # insert
        db.bangumi.add(test_data)
        assert db.bangumi.search_id(1) == test_data

        # update
        test_data.official_title = "无职转生，到了异世界就拿出真本事II"
        db.bangumi.update(test_data)
        assert db.bangumi.search_id(1) == test_data

        # search poster
        assert db.bangumi.match_poster("无职转生，到了异世界就拿出真本事II (2021)") == "/test/test.jpg"

        # match torrent
        result = db.bangumi.match_torrent(
            "[Lilith-Raws] 无职转生，到了异世界就拿出真本事 / Mushoku Tensei - 11 [Baha][WEB-DL][1080p][AVC AAC][CHT][MP4]"
        )
        assert result.official_title == "无职转生，到了异世界就拿出真本事II"

        # delete
        db.bangumi.delete_one(1)
        assert db.bangumi.search_id(1) is None


def test_torrent_database():
    test_data = Torrent(
        name="[Sub Group]test S02 01 [720p].mkv",
        url="https://test.com/test.mkv",
    )
    with Database(engine) as db:
        # insert
        db.torrent.add(test_data)
        assert db.torrent.search(1) == test_data

        # update
        test_data.downloaded = True
        db.torrent.update(test_data)
        assert db.torrent.search(1) == test_data


def test_rss_database():
    rss_url = "https://test.com/test.xml"

    with Database(engine) as db:
        db.rss.add(RSSItem(url=rss_url))

backend/src/test/test_database.py
maikirakiwi/Goto_Bangumi
import json
import os
from unittest import mock

from dotenv import load_dotenv
from module.parser.analyser.openai import OpenAIParser

load_dotenv()


class TestOpenAIParser:
    @classmethod
    def setup_class(cls):
        api_key = os.getenv("OPENAI_API_KEY") or "testing!"
        cls.parser = OpenAIParser(api_key=api_key)

    def test_parse(self):
        text = "[梦蓝字幕组]New Doraemon 哆啦A梦新番[747][2023.02.25][AVC][1080P][GB_JP][MP4]"
        expected = {
            "group": "梦蓝字幕组",
            "title_en": "New Doraemon",
            "resolution": "1080P",
            "episode": 747,
            "season": 1,
            "title_zh": "哆啦A梦新番",
            "sub": "GB_JP",
            "title_jp": "",
            "season_raw": "2023.02.25",
            "source": "AVC",
        }

        with mock.patch("module.parser.analyser.OpenAIParser.parse") as mocker:
            mocker.return_value = json.dumps(expected)

            result = self.parser.parse(text=text, asdict=False)
            assert json.loads(result) == expected

backend/src/test/test_openai.py
maikirakiwi/Goto_Bangumi
from module.conf import PLATFORM


def test_path_to_bangumi():
    # Test for unix-like path
    from module.downloader.path import TorrentPath
    path = "Downloads/Bangumi/Kono Subarashii Sekai ni Shukufuku wo!/Season 2/"
    bangumi_name, season = TorrentPath()._path_to_bangumi(path)
    assert bangumi_name == "Kono Subarashii Sekai ni Shukufuku wo!"
    assert season == 2



backend/src/test/test_path_parser.py
maikirakiwi/Goto_Bangumi
from module.parser.analyser import raw_parser


def test_raw_parser():
    content = "【幻樱字幕组】【4月新番】【古见同学有交流障碍症 第二季 Komi-san wa, Komyushou Desu. S02】【22】【GB_MP4】【1920X1080】"
    info = raw_parser(content)
    assert info.title_en == "Komi-san wa, Komyushou Desu."
    assert info.resolution == "1920X1080"
    assert info.episode == 22
    assert info.season == 2

    content = "[百冬练习组&LoliHouse] BanG Dream! 少女乐团派对！☆PICO FEVER！ / Garupa Pico: Fever! - 26 [WebRip 1080p HEVC-10bit AAC][简繁内封字幕][END] [101.69 MB]"
    info = raw_parser(content)
    assert info.group == "百冬练习组&LoliHouse"
    assert info.title_zh == "BanG Dream! 少女乐团派对！☆PICO FEVER！"
    assert info.resolution == "1080p"
    assert info.episode == 26
    assert info.season == 1

    content = "【喵萌奶茶屋】★04月新番★[夏日重现/Summer Time Rendering][11][1080p][繁日双语][招募翻译]"
    info = raw_parser(content)
    assert info.group == "喵萌奶茶屋"
    assert info.title_en == "Summer Time Rendering"
    assert info.resolution == "1080p"
    assert info.episode == 11
    assert info.season == 1

    content = "[Lilith-Raws] 关于我在无意间被隔壁的天使变成废柴这件事 / Otonari no Tenshi-sama - 09 [Baha][WEB-DL][1080p][AVC AAC][CHT][MP4]"
    info = raw_parser(content)
    assert info.group == "Lilith-Raws"
    assert info.title_zh == "关于我在无意间被隔壁的天使变成废柴这件事"
    assert info.title_en == "Otonari no Tenshi-sama"
    assert info.resolution == "1080p"
    assert info.episode == 9
    assert info.season == 1

    content = "[梦蓝字幕组]New Doraemon 哆啦A梦新番[747][2023.02.25][AVC][1080P][GB_JP][MP4]"
    info = raw_parser(content)
    assert info.group == "梦蓝字幕组"
    assert info.title_zh == "哆啦A梦新番"
    assert info.title_en == "New Doraemon"
    assert info.resolution == "1080P"
    assert info.episode == 747
    assert info.season == 1

    content = "[织梦字幕组][尼尔：机械纪元 NieR Automata Ver1.1a][02集][1080P][AVC][简日双语]"
    info = raw_parser(content)
    assert info.group == "织梦字幕组"
    assert info.title_zh == "尼尔：机械纪元"
    assert info.title_en == "NieR Automata Ver1.1a"
    assert info.resolution == "1080P"
    assert info.episode == 2
    assert info.season == 1

    content = "[MagicStar] 假面骑士Geats / 仮面ライダーギーツ EP33 [WEBDL] [1080p] [TTFC]【生】"
    info = raw_parser(content)
    assert info.group == "MagicStar"
    assert info.title_zh == "假面骑士Geats"
    assert info.title_jp == "仮面ライダーギーツ"
    assert info.resolution == "1080p"
    assert info.episode == 33
    assert info.season == 1

    content = "【极影字幕社】★4月新番 天国大魔境 Tengoku Daimakyou 第05话 GB 720P MP4（字幕社招人内详）"
    info = raw_parser(content)
    assert info.group == "极影字幕社"
    assert info.title_zh == "天国大魔境"
    assert info.title_en == "Tengoku Daimakyou"
    assert info.resolution == "720P"
    assert info.episode == 5
    assert info.season == 1

    content = "【喵萌奶茶屋】★07月新番★[银砂糖师与黑妖精 ~ Sugar Apple Fairy Tale ~][13][1080p][简日双语][招募翻译]"
    info = raw_parser(content)
    assert info.group == "喵萌奶茶屋"
    assert info.title_zh == "银砂糖师与黑妖精"
    assert info.title_en == "~ Sugar Apple Fairy Tale ~"
    assert info.resolution == "1080p"
    assert info.episode == 13
    assert info.season == 1

    content = "[ANi]  16bit 的感动 ANOTHER LAYER - 01 [1080P][Baha][WEB-DL][AAC AVC][CHT][MP4]"
    info = raw_parser(content)
    assert info.group == "ANi"
    assert info.title_zh == "16bit 的感动 ANOTHER LAYER"
    assert info.resolution == "1080P"
    assert info.episode == 1
    assert info.season == 1




backend/src/test/test_raw_parser.py
maikirakiwi/Goto_Bangumi
from module.rss.engine import RSSEngine

from .test_database import engine as e


def test_rss_engine():
    with RSSEngine(e) as engine:
        rss_link = "https://mikanani.me/RSS/Bangumi?bangumiId=2353&subgroupid=552"

        engine.add_rss(rss_link, aggregate=False)

        result = engine.rss.search_active()
        assert result[1].name == "Mikan Project - 无职转生～到了异世界就拿出真本事～"

        new_torrents = engine.pull_rss(result[1])
        torrent = new_torrents[0]
        assert torrent.name == "[Lilith-Raws] 无职转生，到了异世界就拿出真本事 / Mushoku Tensei - 11 [Baha][WEB-DL][1080p][AVC AAC][CHT][MP4]"


backend/src/test/test_rss_engine.py
maikirakiwi/Goto_Bangumi
import json
import os

import pytest
from module.conf import settings
from module.parser.title_parser import TitleParser


class TestTitleParser:
    def test_parse_without_openai(self):
        text = "[梦蓝字幕组]New Doraemon 哆啦A梦新番[747][2023.02.25][AVC][1080P][GB_JP][MP4]"
        result = TitleParser.raw_parser(text)
        assert result.group_name == "梦蓝字幕组"
        assert result.title_raw == "New Doraemon"
        assert result.dpi == "1080P"
        assert result.season == 1
        assert result.subtitle == "GB_JP"

    @pytest.mark.skipif(
        not settings.experimental_openai.enable,
        reason="OpenAI is not enabled in settings",
    )
    def test_parse_with_openai(self):
        text = "[梦蓝字幕组]New Doraemon 哆啦A梦新番[747][2023.02.25][AVC][1080P][GB_JP][MP4]"
        result = TitleParser.raw_parser(text)
        assert result.group_name == "梦蓝字幕组"
        assert result.title_raw == "New Doraemon"
        assert result.dpi == "1080P"
        assert result.season == 1
        assert result.subtitle == "GB_JP"

backend/src/test/test_title_parser.py
maikirakiwi/Goto_Bangumi
from module.parser.analyser.tmdb_parser import tmdb_parser


def test_tmdb_parser():
    bangumi_title = "海盗战记"
    bangumi_year = "2019"
    bangumi_season = 2

    tmdb_info = tmdb_parser(bangumi_title, "zh", test=True)

    assert tmdb_info.title == "冰海战记"
    assert tmdb_info.year == bangumi_year
    assert tmdb_info.last_season == bangumi_season

backend/src/test/test_tmdb.py
maikirakiwi/Goto_Bangumi
import sys

import pytest
from module.parser.analyser import torrent_parser
from module.parser.analyser.torrent_parser import get_path_basename


def test_torrent_parser():
    file_name = "[Lilith-Raws] Boku no Kokoro no Yabai Yatsu - 01 [Baha][WEB-DL][1080p][AVC AAC][CHT][MP4].mp4"
    bf = torrent_parser(file_name)
    assert bf.title == "Boku no Kokoro no Yabai Yatsu"
    assert bf.group == "Lilith-Raws"
    assert bf.episode == 1
    assert bf.season == 1

    file_name = "[Sakurato] Tonikaku Kawaii S2 [01][AVC-8bit 1080p AAC][CHS].mp4"
    bf = torrent_parser(file_name)
    assert bf.title == "Tonikaku Kawaii"
    assert bf.group == "Sakurato"
    assert bf.episode == 1
    assert bf.season == 2

    file_name = "[SweetSub&LoliHouse] Heavenly Delusion - 01 [WebRip 1080p HEVC-10bit AAC ASSx2].mkv"
    bf = torrent_parser(file_name)
    assert bf.title == "Heavenly Delusion"
    assert bf.group == "SweetSub&LoliHouse"
    assert bf.episode == 1
    assert bf.season == 1

    file_name = "[SBSUB][CONAN][1082][V2][1080P][AVC_AAC][CHS_JP](C1E4E331).mp4"
    bf = torrent_parser(file_name)
    assert bf.title == "CONAN"
    assert bf.group == "SBSUB"
    assert bf.episode == 1082
    assert bf.season == 1

    file_name = "海盗战记 (2019) S01E01.mp4"
    bf = torrent_parser(file_name)
    assert bf.title == "海盗战记 (2019)"
    assert bf.episode == 1
    assert bf.season == 1

    file_name = "海盗战记/海盗战记 S01E01.mp4"
    bf = torrent_parser(file_name)
    assert bf.title == "海盗战记"
    assert bf.episode == 1
    assert bf.season == 1

    file_name = "海盗战记 S01E01.zh-tw.ass"
    sf = torrent_parser(file_name, file_type="subtitle")
    assert sf.title == "海盗战记"
    assert sf.episode == 1
    assert sf.season == 1
    assert sf.language == "zh-tw"

    file_name = "海盗战记 S01E01.SC.ass"
    sf = torrent_parser(file_name, file_type="subtitle")
    assert sf.title == "海盗战记"
    assert sf.season == 1
    assert sf.episode == 1
    assert sf.language == "zh"

    file_name = "水星的魔女(2022) S00E19.mp4"
    bf = torrent_parser(file_name, season=0)
    assert bf.title == "水星的魔女(2022)"
    assert bf.season == 0
    assert bf.episode == 19

    file_name = "【失眠搬运组】放学后失眠的你-Kimi wa Houkago Insomnia - 06 [bilibili - 1080p AVC1 CHS-JP].mp4"
    bf = torrent_parser(file_name, season=1)
    assert bf.title == "放学后失眠的你-Kimi wa Houkago Insomnia"
    assert bf.season == 1
    assert bf.episode == 6


class TestGetPathBasename:
    def test_regular_path(self):
        assert get_path_basename("/path/to/file.txt") == "file.txt"

    def test_empty_path(self):
        assert get_path_basename("") == ""

    def test_path_with_trailing_slash(self):
        assert get_path_basename("/path/to/folder/") == "folder"

    @pytest.mark.skipif(not sys.platform.startswith("win"), reason="Windows specific")
    def test_windows_path(self):
        assert get_path_basename("C:\\path\\to\\file.txt") == "file.txt"

backend/src/test/test_torrent_parser.py
maikirakiwi/Goto_Bangumi