add support for wechat

src/plugins/nonebot_hk_reporter/platform/__init__.py

@@ -1,6 +1,7 @@
 from .bilibili import Bilibili
 from .rss import Rss
 from .weibo import Weibo
+from .wechat import Wechat
 from .utils import check_sub_target
 from .platform import PlatformNoTarget
 from .utils import platform_manager

src/plugins/nonebot_hk_reporter/platform/utils.py

@@ -5,6 +5,7 @@ from typing import Type
 from .weibo import Weibo
 from .bilibili import Bilibili
 from .rss import Rss
+from .wechat import Wechat
 from .platform import PlatformProto
 from ..config import Config
 from ..post import Post
@@ -16,7 +17,8 @@ async def check_sub_target(target_type, target):
 platform_manager: dict[str, PlatformProto] = {
         'bilibili': Bilibili(),
         'weibo': Weibo(),
-        'rss': Rss()
+        'rss': Rss(),
+        'wechat': Wechat(),
     }
 
 async def fetch_and_send(target_type: str):

src/plugins/nonebot_hk_reporter/platform/wechat.py

@@ -0,0 +1,73 @@
+from datetime import datetime
+import hashlib
+import json
+import re
+from typing import Any, Optional
+
+from bs4 import BeautifulSoup as bs
+import httpx
+
+from ..post import Post
+from ..types import *
+from .platform import Platform
+
+
+class Wechat(Platform):
+
+    categories = {}
+    enable_tag = False
+    platform_name = 'wechat'
+
+    @classmethod
+    def _get_query_url(cls, target: Target):
+        return 'https://weixin.sogou.com/weixin?type=1&s_from=input&query={}&ie=utf8&_sug_=n&_sug_type_='.format(target)
+
+    @classmethod
+    async def _get_target_soup(cls, target: Target) -> Optional[bs]:
+        target_url = cls._get_query_url(target)
+        async with httpx.AsyncClient() as client:
+            res = await client.get(target_url)
+        soup = bs(res.text, 'html.parser')
+        blocks = soup.find(class_='news-list2').find_all('li',recursive=False)
+        for block in blocks:
+            if block.find(string=[target]):
+                return block
+
+    @classmethod
+    async def get_account_name(cls, target: Target) -> Optional[str]:
+        if not (block := await cls._get_target_soup(target)):
+            return None
+        return block.find('p', class_='tit').find('a').text
+
+    async def get_sub_list(self, target: Target) -> list[RawPost]:
+        block = await self._get_target_soup(target)
+        if (last_post_dt := block.find('dt', string='最近文章：')):
+            post = {
+                    'title': last_post_dt.find_parent().find('a').text,
+                    'target': target,
+                    'page_url': self._get_query_url(target)
+                    }
+            return [post]
+        else:
+            return []
+
+    def get_id(self, post: RawPost) -> Any:
+        return post['title']
+
+    def get_date(self, post: RawPost):
+        return None
+
+    def get_tags(self, post: RawPost):
+        return None
+
+    def get_category(self, post: RawPost):
+        return None
+
+    async def parse(self, raw_post: RawPost) -> Post:
+        # TODO get content of post
+        return Post(target_type='wechat',
+                text='{}\n详细内容请自行查看公众号'.format(raw_post['title']),
+                pics=[],
+                url=''
+                )
+

src/plugins/nonebot_hk_reporter/utils.py

@@ -1,10 +1,11 @@
 import os
 import asyncio
-from typing import Optional
+from typing import Awaitable, Callable, Optional
 import nonebot
 from nonebot import logger
 import base64
 from pyppeteer import launch
+from pyppeteer.page import Page
 from pyppeteer.chromium_downloader import check_chromium, download_chromium
 from html import escape
 from hashlib import sha256
@@ -30,14 +31,18 @@ class Render(metaclass=Singleton):
     def __init__(self):
         self.lock = asyncio.Lock()
 
-    async def render(self, url: str, viewport: Optional[dict] = None, target: Optional[str] = None) -> str:
+    async def render(self, url: str, viewport: Optional[dict] = None, target: Optional[str] = None,
+            operation: Optional[Callable[[Page], Awaitable[None]]] = None) -> str:
         async with self.lock:
             if plugin_config.hk_reporter_use_local:
                 browser = await launch(executablePath='/usr/bin/chromium', args=['--no-sandbox'])
             else:
                 browser = await launch(args=['--no-sandbox'])
             page = await browser.newPage()
-            await page.goto(url)
+            if operation:
+                await operation(page)
+            else:
+                await page.goto(url)
             if viewport:
                 await page.setViewport(viewport)
             if target: