✨ use official recommended posts

main.py

@@ -1,8 +1,22 @@
-from uvicorn import run
+import asyncio
+import uvicorn
 
 from src.app import app
 from src.env import PORT
+from src.render.article import refresh_recommend_posts
+
+
+async def main():
+    await refresh_recommend_posts()
+    web_server = uvicorn.Server(config=uvicorn.Config(app, host="0.0.0.0", port=PORT))
+    server_config = web_server.config
+    server_config.setup_event_loop()
+    if not server_config.loaded:
+        server_config.load()
+    web_server.lifespan = server_config.lifespan_class(server_config)
+    await web_server.startup()
+    await web_server.main_loop()
 
 
 if __name__ == "__main__":
-    run(app, host="0.0.0.0", port=PORT)
+    asyncio.run(main())

src/api/hyperion.py

@@ -1,7 +1,7 @@
 from typing import List
 
 from .hyperionrequest import HyperionRequest
-from .models import PostInfo
+from .models import PostInfo, PostRecommend
 from ..typedefs import JSON_DATA
 
 __all__ = ("Hyperion",)
@@ -14,7 +14,6 @@ class Hyperion:
     """
 
     POST_FULL_URL = "https://bbs-api.miyoushe.com/post/wapi/getPostFull"
-    POST_SEM_URL = "https://bbs-api.miyoushe.com/post/wapi/semPosts"
     POST_FULL_IN_COLLECTION_URL = (
         "https://bbs-api.miyoushe.com/post/wapi/getPostFullInCollection"
     )
@@ -70,6 +69,13 @@ class Hyperion:
         )
         return f"?x-oss-process={params}"
 
+    async def get_official_recommended_posts(self, gids: int) -> List[PostRecommend]:
+        params = {"gids": gids}
+        response = await self.client.get(
+            url=self.GET_OFFICIAL_RECOMMENDED_POSTS_URL, params=params
+        )
+        return [PostRecommend(**data) for data in response["list"]]
+
     async def get_post_full_in_collection(
         self, collection_id: int, gids: int = 2, order_type=1
     ) -> JSON_DATA:
@@ -88,13 +94,6 @@ class Hyperion:
         response = await self.client.get(self.POST_FULL_URL, params=params)
         return PostInfo.paste_data(response)
 
-    async def get_same_posts(self, gids: int, post_id: int) -> List[PostInfo]:
-        params = {"gids": gids, "post_id": post_id}
-        response = await self.client.get(self.POST_SEM_URL, params=params)
-        return [
-            PostInfo.paste_data(**{"data": {"post": post}}) for post in response["list"]
-        ]
-
     async def get_new_list(self, gids: int, type_id: int, page_size: int = 20):
         """
         ?gids=2&page_size=20&type=3

src/api/models.py

@@ -7,6 +7,7 @@ __all__ = (
     "PostStat",
     "PostType",
     "PostInfo",
+    "PostRecommend",
 )
 
 
@@ -84,3 +85,10 @@ class PostInfo(BaseModel):
 
     def __getitem__(self, item):
         return self._data[item]
+
+
+class PostRecommend(BaseModel):
+    post_id: int
+    subject: str
+    banner: Optional[str]
+    official_type: Optional[int]

src/log.py

@@ -11,4 +11,5 @@ root_logger.setLevel(logging.ERROR)
 root_logger.addHandler(logging_handler)
 logging.basicConfig(level=logging.INFO)
 logs.setLevel(logging.INFO)
+logging.getLogger("apscheduler").setLevel(logging.INFO)
 logger = logging.getLogger("FixMiYouShe")

src/render/article.py

@@ -1,22 +1,19 @@
 import json
 from datetime import datetime
-from typing import Union, List, Dict
+from typing import Union, List, Dict, Optional
 
 from bs4 import BeautifulSoup, Tag, PageElement
 
 from src import template_env
 from src.api.hyperion import Hyperion
-from src.api.models import PostStat, PostInfo, PostType
-from src.env import DEBUG, DOMAIN
+from src.api.models import PostStat, PostInfo, PostType, PostRecommend
+from src.env import DOMAIN
 from src.error import ArticleNotFoundError
-from src.services.cache import (
-    get_article_cache_file_path,
-    get_article_cache_file,
-    write_article_cache_file,
-)
-from src.services.scheduler import add_delete_file_job
+from src.log import logger
+from src.services.scheduler import scheduler
 
 GAME_ID_MAP = {"bh3": 1, "ys": 2, "bh2": 3, "wd": 4, "dby": 5, "sr": 6, "zzz": 8}
+RECOMMEND_POST_MAP: Dict[str, List[PostRecommend]] = {}
 CHANNEL_MAP = {"ys": "yuanshen", "sr": "HSRCN", "zzz": "ZZZNewsletter"}
 template = template_env.get_template("article.jinja2")
 
@@ -99,9 +96,14 @@ def parse_stat(stat: PostStat):
     )
 
 
-def get_public_data(
-    game_id: str, post_id: int, post_info: PostInfo, post_sam_info: List[PostInfo]
-) -> Dict:
+def get_recommend_post(game_id: str, post_id: Optional[int]) -> List[PostRecommend]:
+    posts = RECOMMEND_POST_MAP.get(game_id, [])
+    if post_id:
+        return [post for post in posts if post.post_id != post_id]
+    return posts
+
+
+def get_public_data(game_id: str, post_id: int, post_info: PostInfo) -> Dict:
     cover = post_info.cover
     if (not post_info.cover) and post_info.image_urls:
         cover = post_info.image_urls[0]
@@ -116,25 +118,21 @@ def get_public_data(
         "cover": cover,
         "post": post_info,
         "author": post_info["post"]["user"],
-        "related_posts": post_sam_info,
+        "related_posts": get_recommend_post(game_id, post_id),
         "DOMAIN": DOMAIN,
     }
 
 
-async def process_article_text(
-    game_id: str, post_id: int, post_info: PostInfo, post_sam_info: List[PostInfo]
-) -> str:
+async def process_article_text(game_id: str, post_id: int, post_info: PostInfo) -> str:
     post_soup = BeautifulSoup(post_info.content, features="lxml")
     return template.render(
         description=get_description(post_soup),
         article=parse_content(post_soup, post_info.subject, post_info.video_urls),
-        **get_public_data(game_id, post_id, post_info, post_sam_info),
+        **get_public_data(game_id, post_id, post_info),
     )
 
 
-async def process_article_image(
-    game_id: str, post_id: int, post_info: PostInfo, post_sam_info: List[PostInfo]
-) -> str:
+async def process_article_image(game_id: str, post_id: int, post_info: PostInfo) -> str:
     json_data = json.loads(post_info.content)
     description = json_data.get("describe", "")
     article = ""
@@ -145,30 +143,38 @@ async def process_article_image(
     return template.render(
         description=description,
         article=article,
-        **get_public_data(game_id, post_id, post_info, post_sam_info),
+        **get_public_data(game_id, post_id, post_info),
     )
 
 
 async def process_article(game_id: str, post_id: int) -> str:
-    path = get_article_cache_file_path(game_id, post_id)
-    if content := await get_article_cache_file(path):
-        return content
     gids = GAME_ID_MAP.get(game_id)
     if not gids:
         raise ArticleNotFoundError(game_id, post_id)
     hyperion = Hyperion()
     try:
         post_info = await hyperion.get_post_info(gids=gids, post_id=post_id)
-        post_sam_info = await hyperion.get_same_posts(gids=gids, post_id=post_id)
     finally:
         await hyperion.close()
     if post_info.view_type in [PostType.TEXT, PostType.VIDEO]:
-        content = await process_article_text(game_id, post_id, post_info, post_sam_info)
+        content = await process_article_text(game_id, post_id, post_info)
     elif post_info.view_type == PostType.IMAGE:
-        content = await process_article_image(
-            game_id, post_id, post_info, post_sam_info
-        )
-    if not DEBUG:
-        await write_article_cache_file(path, content)
-        add_delete_file_job(path)
-    return content
+        content = await process_article_image(game_id, post_id, post_info)
+    return content  # noqa
+
+
+@scheduler.scheduled_job("cron", minute="0", second="10")
+async def refresh_recommend_posts():
+    logger.info("Start to refresh recommend posts")
+    hyperion = Hyperion()
+    try:
+        for key, gids in GAME_ID_MAP.items():
+            try:
+                RECOMMEND_POST_MAP[key] = await hyperion.get_official_recommended_posts(
+                    gids
+                )
+            except Exception as _:
+                logger.exception(f"Failed to get recommend posts gids={gids}")
+    finally:
+        await hyperion.close()
+    logger.info("Finish to refresh recommend posts")

src/services/cache.py

@@ -1,23 +0,0 @@
-from pathlib import Path
-from typing import Optional
-
-import aiofiles
-
-cache_dir = Path("cache")
-cache_dir.mkdir(exist_ok=True)
-
-
-def get_article_cache_file_path(game_id: str, article_id: int) -> Path:
-    return cache_dir / f"article_{game_id}_{article_id}.html"
-
-
-async def get_article_cache_file(path: Path) -> Optional[str]:
-    if not path.exists():
-        return None
-    async with aiofiles.open(path, "r", encoding="utf-8") as f:
-        return await f.read()
-
-
-async def write_article_cache_file(path: Path, content: str) -> None:
-    async with aiofiles.open(path, "w", encoding="utf-8") as f:
-        await f.write(content)

src/services/scheduler.py

@@ -1,28 +1,5 @@
-import datetime
-from pathlib import Path
-
-import pytz
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 
 scheduler = AsyncIOScheduler(timezone="Asia/ShangHai")
 if not scheduler.running:
     scheduler.start()
-
-
-async def delete_file(path: Path):
-    path = Path(path)
-    if path.exists():
-        path.unlink(missing_ok=True)
-
-
-def add_delete_file_job(path: Path, delete_seconds: int = 3600):
-    scheduler.add_job(
-        delete_file,
-        "date",
-        id=f"{hash(path)}|delete_file",
-        name=f"{hash(path)}|delete_file",
-        args=[path],
-        run_date=datetime.datetime.now(pytz.timezone("Asia/Shanghai"))
-        + datetime.timedelta(seconds=delete_seconds),
-        replace_existing=True,
-    )

src/templates/article.jinja2

@@ -71,8 +71,9 @@ Embed MiYouShe posts, videos, polls, and more on Telegram
     {% if related_posts %}
     {% for post in related_posts %}
     <related>
-        <a href="https://{{ DOMAIN }}/{{ game_id }}/article/{{ post.post_id }}"></a>
+        <a href="https://{{ DOMAIN }}/{{ game_id }}/article/{{ post.post_id }}">{{ post.subject }}</a>
     </related>
+    <br/>
     {% endfor %}
     {% endif %}
 </article>