feat: 支持通过ob11自动获取cookies

campux/config/migration.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+import abc
+import typing
+
+from ..core import app
+
+
+preregistered_migrations: list[typing.Type[Migration]] = []
+"""当前阶段暂不支持扩展"""
+
+def migration_class(name: str, number: int):
+    """注册一个迁移
+    """
+    def decorator(cls: typing.Type[Migration]) -> typing.Type[Migration]:
+        cls.name = name
+        cls.number = number
+        preregistered_migrations.append(cls)
+        return cls
+
+    return decorator
+
+
+class Migration(abc.ABC):
+    """一个版本的迁移
+    """
+
+    name: str
+
+    number: int
+
+    ap: app.Application
+
+    def __init__(self, ap: app.Application):
+        self.ap = ap
+
+    @abc.abstractmethod
+    async def need_migrate(self) -> bool:
+        """判断当前环境是否需要运行此迁移
+        """
+        pass
+
+    @abc.abstractmethod
+    async def run(self):
+        """执行迁移
+        """
+        pass

campux/config/migrations/m001_cookies_refresh_config.py

@@ -0,0 +1,14 @@
+from .. import migration
+
+from ...core import app
+
+
+@migration.migration_class("001_cookies_refresh_config", 1)
+class Migration(migration.Migration):
+
+    async def need_migrate(self) -> bool:
+        return 'campux_qzone_cookies_refresh_strategy' not in self.ap.config.data
+
+    async def migrate(self):
+        self.ap.config.data['campux_qzone_cookies_refresh_strategy'] = 'qrcode'
+        await self.ap.config.dump_config()

campux/core/app.py

@@ -14,6 +14,7 @@
 from ..imbot import mgr as imbot_mgr
 from ..common import cache as cache_mgr
 from ..config import manager as config_mgr
+from ..config import migration
 
 
 class Application:
@@ -82,6 +83,12 @@ async def create_app() -> Application:
 
     await config.load_config()
 
+    # 迁移
+    for migration_cls in migration.preregistered_migrations:
+        migration_inst = migration_cls(ap=ap)
+        if await migration_inst.need_migrate():
+            await migration_inst.migrate()
+
     # 读取环境变量进行替换, for config
     config_data = config.data.copy()
 

campux/imbot/nbmod.py

@@ -228,7 +228,12 @@ async def qrcode_callback(content: bytes):
 
         await fdelay()
 
-        await ap.social.platform_api.relogin(qrcode_callback)
+        await ap.social.platform_api.relogin(
+            strategy=ap.config.data['campux_qzone_cookies_refresh_strategy'],
+            qrcode_callback=qrcode_callback,
+            ob11_bot=nonebot.get_bot(),
+            ob11_auto_callback=lambda result: print(result),
+        )
 
         await ap.imbot.send_group_message(
             ap.config.data['campux_review_qq_group_id'],

campux/social/qzone/api.py

@@ -6,6 +6,7 @@
 import traceback
 
 import requests
+from nonebot.adapters import onebot
 
 from ...core import app
 from . import login
@@ -117,10 +118,19 @@ def image_to_base64(self, image: bytes) -> str:
 
         return str(pic_base64)[2:-1]
 
-    async def relogin(self, callback: callable):
+    async def relogin(
+        self,
+        strategy: str='qrcode',
+        qrcode_callback: callable=lambda qrcode: None,
+        ob11_auto_callback: callable=lambda result: None,
+        ob11_bot: onebot.v11.Bot=None,  # 如果strategy为ob11_auto，则需要传入一个onebot v11的bot实例
+    ):
         loginmgr = login.QzoneLogin()
 
-        self.cookies = await loginmgr.login_via_qrcode(callback)
+        if strategy == 'qrcode':
+            self.cookies = await loginmgr.login_via_qrcode(qrcode_callback)
+        elif strategy == 'ob11_auto':
+            self.cookies = await loginmgr.login_via_ob11_bot(ob11_auto_callback, ob11_bot)
 
         if 'p_skey' in self.cookies:
             self.gtk2 = generate_gtk(self.cookies['p_skey'])

campux/social/qzone/login.py

@@ -5,6 +5,7 @@
 import re
 
 import requests
+from nonebot.adapters import onebot
 
 qrcode_url = "https://ssl.ptlogin2.qq.com/ptqrshow?appid=549000912&e=2&l=M&s=3&d=72&v=4&t=0.31232733520361844&daid=5&pt_3rd_aid=0"
 
@@ -89,6 +90,21 @@ async def login_via_qrcode(
 
                     return final_cookie_dict
         raise Exception("{}次尝试失败".format(max_timeout_times))
+
+    async def login_via_ob11_bot(
+        self,
+        ob11_auto_callback: typing.Callable[[dict], typing.Awaitable[None]],
+        ob11_bot: onebot.v11.Bot,
+    ):
+        cookies = await ob11_bot.get_cookies(
+            domain='qzone.qq.com',
+        )
+
+        cookies = cookies.get('cookies', {})
+
+        await ob11_auto_callback(cookies)
+
+        return cookies
 
 
 if __name__ == '__main__':

templates/config.json

@@ -15,5 +15,6 @@
     "campux_review_help_message": "审核命令：\n#通过 <稿件id>\n\n#拒绝 <理由> <稿件id>\n\n例如：\n#通过 10\n#拒绝 测试理由 10",
     "campux_review_qq_group_id": 123456789,
     "campux_domain": "campux",
-    "campux_qq_bot_uin": 123456789
-}
+    "campux_qq_bot_uin": 123456789,
+    "campux_qzone_cookies_refresh_strategy": "qrcode"
+}