✨ 添加 rss 对 Mikan 的支持

3 files changed, 295 insertions, 0 deletions

diff --git a/ATRI/plugins/rss/rss_mikanan/__init__.py b/ATRI/plugins/rss/rss_mikanan/__init__.py
index e69de29..4de04b5 100644
--- a/ATRI/plugins/rss/rss_mikanan/__init__.py
+++ b/ATRI/plugins/rss/rss_mikanan/__init__.py
@@ -0,0 +1,166 @@
+import pytz
+import asyncio
+from tabulate import tabulate
+from datetime import timedelta, datetime
+
+from apscheduler.triggers.base import BaseTrigger
+from apscheduler.triggers.combining import AndTrigger
+from apscheduler.triggers.interval import IntervalTrigger
+
+from nonebot import get_bot
+from nonebot.matcher import Matcher
+from nonebot.params import CommandArg, ArgPlainText
+from nonebot.permission import Permission
+from nonebot.adapters.onebot.v11 import Message, GroupMessageEvent
+
+from ATRI.log import logger as log
+from ATRI.plugins.rss.rss_rsshub.data_source import RssHubSubscriptor
+from ATRI.utils import timestamp2datetime
+from ATRI.utils.apscheduler import scheduler
+from ATRI.database import RssMikananiSubcription
+
+from .data_source import RssMikananSubscriptor
+
+
+add_sub = RssMikananSubscriptor().cmd_as_group("add", "为本群添加 Mikan 订阅")
+
+
+@add_sub.handle()
+async def _(matcher: Matcher, args: Message = CommandArg()):
+    msg = args.extract_plain_text()
+    if msg:
+        matcher.set_arg("rm_add_url", args)
+
+
+@add_sub.got("rm_add_url", prompt="Mikan 链接呢? 速速")
+async def _(event: GroupMessageEvent, _url: str = ArgPlainText("rm_add_url")):
+    group_id = event.group_id
+    sub = RssMikananSubscriptor()
+
+    result = await sub.add_sub(_url, group_id)
+    await add_sub.finish(result)
+
+
+del_sub = RssMikananSubscriptor().cmd_as_group("del", "删除本群 Mikan 订阅")
+
+
+@del_sub.handle()
+async def _(event: GroupMessageEvent):
+    group_id = event.group_id
+    sub = RssMikananSubscriptor()
+
+    query_result = await sub.get_sub_list({"group_id": group_id})
+    if not query_result:
+        await del_sub.finish("本群还没有任何订阅呢...")
+
+    subs = list()
+    for i in query_result:
+        subs.append([i._id, i.title])
+
+    output = "本群的 Mikan 订阅列表如下~\n" + tabulate(
+        subs, headers=["ID", "Title"], tablefmt="plain"
+    )
+    await del_sub.send(output)
+
+
+@del_sub.got("rm_del_sub_id", prompt="要取消的ID呢? 速速\n(键入 q 以取消)")
+async def _(event: GroupMessageEvent, _id: str = ArgPlainText("rm_del_sub_id")):
+    if _id == "q":
+        await del_sub.finish("已取消操作~")
+
+    group_id = event.group_id
+    sub = RssMikananSubscriptor()
+
+    result = await sub.del_sub(_id, group_id)
+    await del_sub.finish(result)
+
+
+get_sub_list = RssMikananSubscriptor().cmd_as_group("list", "获取本群 Mikan 订阅列表", permission=Permission())
+
+
+@get_sub_list.handle()
+async def _(event: GroupMessageEvent):
+    group_id = event.group_id
+    sub = RssMikananSubscriptor()
+
+    query_result = await sub.get_sub_list({"group_id": group_id})
+    if not query_result:
+        await get_sub_list.finish("本群还没有任何订阅呢...")
+    
+    subs = list()
+    for i in query_result:
+        subs.append([i.update_time, i.title])
+    
+    output = "本群的 Mikan 订阅列表如下~\n" + tabulate(
+        subs, headers=["最后更新时间", "标题"], tablefmt="plain"
+    )
+    await get_sub_list.finish(output)
+
+
+tq = asyncio.Queue()
+
+
+class RssMikanDynamicChecker(BaseTrigger):
+    def get_next_fire_time(self, previous_fire_time, now):
+        conf = RssHubSubscriptor().load_service("rss.mikan")
+        if conf.get("enabled"):
+            return now
+
+
[email protected]_job(
+    AndTrigger([IntervalTrigger(seconds=60), RssMikanDynamicChecker()]),
+    name="Mikan 订阅检查",
+    max_instances=3,
+    misfire_grace_time=60,
+)
+async def _():
+    sub = RssMikananSubscriptor()
+    try:
+        all_dy = await sub.get_all_subs()
+    except Exception:
+        log.debug("Mikan 订阅列表为空 跳过")
+        return
+    
+    if tq.empty():
+        for i in all_dy:
+            await tq.put(i)
+    else:
+        data: RssMikananiSubcription = tq.get_nowait()
+        log.info(f"准备查询 Mikan: {data.title} 的动态, 队列剩余 {tq.qsize()}")
+
+        raw_ts = data.update_time.replace(
+            tzinfo=pytz.timezone("Asia/Shanghai")
+        ) + timedelta(hours=8)
+        ts = raw_ts.timestamp()
+
+        info = await sub.get_mikan_info(data.rss_link)
+        if not info:
+            log.warning(f"无法获取 Mikan: {data.rss_link} 的动态")
+            return
+        
+        time_patt = "%Y-%m-%dT%H:%M:%S.%f"
+
+        if len(info) == 1:
+            pub_date = info["item"]["torrent"]["pubDate"]
+            link = info["item"]["torrent"]["link"]
+        else:
+            item = info["item"][0]
+
+            pub_date = item["torrent"]["pubDate"]
+            link = item["torrent"]["link"]
+
+        m_t = datetime.strptime(pub_date, time_patt).timestamp()
+
+        if ts < m_t:
+            title = data.title
+
+            repo = f"""本群订阅的 Mikan 更新啦!
+            {title}
+            {link}
+            """
+
+            bot = get_bot()
+            await bot.send_group_msg(group_id=data.group_id, message=repo)
+            await sub.update_sub(
+                data._id, data.group_id, {"update_time": timestamp2datetime(m_t)}
+            )
diff --git a/ATRI/plugins/rss/rss_mikanan/data_source.py b/ATRI/plugins/rss/rss_mikanan/data_source.py
index e69de29..eadf97b 100644
--- a/ATRI/plugins/rss/rss_mikanan/data_source.py
+++ b/ATRI/plugins/rss/rss_mikanan/data_source.py
@@ -0,0 +1,105 @@
+import xmltodict
+
+from nonebot.permission import SUPERUSER
+from nonebot.adapters.onebot.v11 import GROUP_OWNER, GROUP_ADMIN
+
+
+from ATRI.service import Service
+from ATRI.rule import is_in_service
+from ATRI.exceptions import RssError
+from ATRI.utils import request, gen_random_str
+
+
+from .db import DB
+
+
+class RssMikananSubscriptor(Service):
+    def __init__(self):
+        Service.__init__(
+            self,
+            "rss.mikan",
+            "Rss的mikan支持",
+            rule=is_in_service("rss.mikan"),
+            permission=SUPERUSER | GROUP_OWNER | GROUP_ADMIN,
+            main_cmd="/rss.mikan",
+        )
+
+    async def __add_sub(self, _id: str, group_id: int):
+        try:
+            async with DB() as db:
+                await db.add_sub(_id, group_id)
+        except Exception:
+            raise RssError("rss.mikan: 添加订阅失败")
+
+    async def update_sub(self, _id: str, group_id: int, update_map: dict):
+        try:
+            async with DB() as db:
+                await db.update_sub(_id, group_id, update_map)
+        except Exception:
+            raise RssError("rss.mikan: 更新订阅失败")
+
+    async def __del_sub(self, _id: str, group_id: int):
+        try:
+            async with DB() as db:
+                await db.del_sub({"_id": _id, "group_id": group_id})
+        except Exception:
+            raise RssError("rss.mikan: 删除订阅失败")
+
+    async def get_sub_list(self, query_map: dict) -> list:
+        try:
+            async with DB() as db:
+                return await db.get_sub_list(query_map)
+        except Exception:
+            raise RssError("rss.mikan: 获取订阅列表失败")
+
+    async def get_all_subs(self) -> list:
+        try:
+            async with DB() as db:
+                return await db.get_all_subs()
+        except Exception:
+            raise RssError("rss.mikan: 获取所有订阅失败")
+
+    async def add_sub(self, url: str, group_id: int) -> str:
+        data = await self.get_mikan_info(url)
+        if not data:
+            return "该链接不含mikan内容"
+        
+        rss_link = data["link"]
+
+        query_result = await self.get_sub_list(
+            {"rss_link": rss_link, "group_id": group_id}
+        )
+        if query_result:
+            _id = query_result[0]._id
+            return f"该链接已经订阅过啦! ID: {_id}"
+
+        _id = gen_random_str(6)
+        title = data["title"]
+        disc = data["description"]
+
+        await self.__add_sub(_id, group_id)
+        await self.update_sub(
+            _id, group_id, {"title": title, "rss_link": rss_link, "discription": disc}
+        )
+        return f"订阅成功! ID: {_id}"
+
+    async def del_sub(self, _id: str, group_id: int) -> str:
+        query_result = await self.get_sub_list({"_id": _id, "group_id": group_id})
+        if not query_result:
+            return "没有找到订阅..."
+        
+        await self.__del_sub(_id, group_id)
+        return f"成功取消ID为 {_id} 的订阅"
+    
+    async def get_mikan_info(self, url: str) -> dict:
+        if "mikanani.me" not in url:
+            return dict()
+
+        try:
+            resp = await request.get(url)
+        except Exception:
+            RssError("rss.mikan: 请求链接失败")
+
+        xml_data = resp.read()
+        data = xmltodict.parse(xml_data)
+        return data["rss"]["@version"]["channel"]
diff --git a/ATRI/plugins/rss/rss_mikanan/db.py b/ATRI/plugins/rss/rss_mikanan/db.py
index e69de29..326f4fd 100644
--- a/ATRI/plugins/rss/rss_mikanan/db.py
+++ b/ATRI/plugins/rss/rss_mikanan/db.py
@@ -0,0 +1,24 @@
+from ATRI.database import RssMikananiSubcription
+
+
+class DB:
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        pass
+
+    async def add_sub(self, _id: str, group_id: int):
+        await RssMikananiSubcription.create(_id=_id, group_id=group_id)
+
+    async def update_sub(self, _id: str, group_id: int, update_map: dict):
+        await RssMikananiSubcription.filter(_id=_id, group_id=group_id).update(**update_map)
+    
+    async def del_sub(self, query_map: dict):
+        await RssMikananiSubcription.filter(**query_map).delete()
+    
+    async def get_sub_list(self, query_map: dict) -> list:
+        return await RssMikananiSubcription.filter(**query_map)
+    
+    async def get_all_subs(self) -> list:
+        return await RssMikananiSubcription.all()