support hoyolab

This commit is contained in:
xtaodada 2023-08-24 15:35:45 +08:00
parent 4be1f7c455
commit e75c7f7943
Signed by: xtaodada
GPG Key ID: 4CBB3F4FA8C85659
15 changed files with 489 additions and 114 deletions

View File

@ -1,3 +1,5 @@
DEBUG=False DEBUG=False
DOMAIN=127.0.0.1 DOMAIN=127.0.0.1
PORT=8080 PORT=8080
MIYOUSHE=true
HOYOLAB=true

10
main.py
View File

@ -2,12 +2,18 @@ import asyncio
import uvicorn import uvicorn
from src.app import app from src.app import app
from src.env import PORT from src.env import PORT, MIYOUSHE, HOYOLAB
from src.render.article import refresh_recommend_posts
async def main(): async def main():
if MIYOUSHE:
from src.render.article import refresh_recommend_posts
await refresh_recommend_posts() await refresh_recommend_posts()
if HOYOLAB:
from src.render.article_hoyolab import refresh_hoyo_recommend_posts
await refresh_hoyo_recommend_posts()
web_server = uvicorn.Server(config=uvicorn.Config(app, host="0.0.0.0", port=PORT)) web_server = uvicorn.Server(config=uvicorn.Config(app, host="0.0.0.0", port=PORT))
server_config = web_server.config server_config = web_server.config
server_config.setup_event_loop() server_config.setup_event_loop()

83
src/api/hoyolab.py Normal file
View File

@ -0,0 +1,83 @@
from typing import List
from .hyperionrequest import HyperionRequest
from .models import PostInfo, PostRecommend, HoYoPostMultiLang
__all__ = ("Hoyolab",)
class Hoyolab:
POST_FULL_URL = "https://bbs-api-os.hoyolab.com/community/post/wapi/getPostFull"
NEW_LIST_URL = "https://bbs-api-os.hoyolab.com/community/post/wapi/getNewsList"
LANG = "zh-cn"
USER_AGENT = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/90.0.4430.72 Safari/537.36"
)
def __init__(self, *args, **kwargs):
self.client = HyperionRequest(headers=self.get_headers(), *args, **kwargs)
def get_headers(self, lang: str = LANG):
return {
"User-Agent": self.USER_AGENT,
"Referer": "https://www.hoyolab.com/",
"X-Rpc-Language": lang,
}
@staticmethod
def get_images_params(
resize: int = 600,
quality: int = 80,
auto_orient: int = 0,
interlace: int = 1,
images_format: str = "jpg",
) -> str:
"""
image/resize,s_600/quality,q_80/auto-orient,0/interlace,1/format,jpg
:param resize: 图片大小
:param quality: 图片质量
:param auto_orient: 自适应
:param interlace: 未知
:param images_format: 图片格式
:return:
"""
params = (
f"image/resize,s_{resize}/quality,q_{quality}/auto-orient,"
f"{auto_orient}/interlace,{interlace}/format,{images_format}"
)
return f"?x-oss-process={params}"
async def get_news_recommend(
self, gids: int, page_size: int = 3, type_: int = 1
) -> List[PostRecommend]:
params = {"gids": gids, "page_size": page_size, "type": type_}
response = await self.client.get(url=self.NEW_LIST_URL, params=params)
return [
PostRecommend(
post_id=data["post"]["post_id"],
subject=data["post"]["subject"],
multi_language_info=HoYoPostMultiLang(
**data["post"]["multi_language_info"]
),
)
for data in response["list"]
]
async def get_post_info(
self, post_id: int, read: int = 1, scene: int = 1, lang: str = LANG
) -> PostInfo:
params = {"post_id": post_id, "read": read, "scene": scene}
response = await self.client.get(
self.POST_FULL_URL, params=params, headers=self.get_headers(lang=lang)
)
return PostInfo.paste_data(response, hoyolab=True)
async def close(self):
await self.client.shutdown()
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.close()

View File

@ -2,7 +2,6 @@ from typing import List
from .hyperionrequest import HyperionRequest from .hyperionrequest import HyperionRequest
from .models import PostInfo, PostRecommend from .models import PostInfo, PostRecommend
from ..typedefs import JSON_DATA
__all__ = ("Hyperion",) __all__ = ("Hyperion",)
@ -14,10 +13,6 @@ class Hyperion:
""" """
POST_FULL_URL = "https://bbs-api.miyoushe.com/post/wapi/getPostFull" POST_FULL_URL = "https://bbs-api.miyoushe.com/post/wapi/getPostFull"
POST_FULL_IN_COLLECTION_URL = (
"https://bbs-api.miyoushe.com/post/wapi/getPostFullInCollection"
)
GET_NEW_LIST_URL = "https://bbs-api.miyoushe.com/post/wapi/getNewsList"
GET_OFFICIAL_RECOMMENDED_POSTS_URL = ( GET_OFFICIAL_RECOMMENDED_POSTS_URL = (
"https://bbs-api.miyoushe.com/post/wapi/getOfficialRecommendedPosts" "https://bbs-api.miyoushe.com/post/wapi/getOfficialRecommendedPosts"
) )
@ -33,19 +28,6 @@ class Hyperion:
def get_headers(self, referer: str = "https://www.miyoushe.com/ys/"): def get_headers(self, referer: str = "https://www.miyoushe.com/ys/"):
return {"User-Agent": self.USER_AGENT, "Referer": referer} return {"User-Agent": self.USER_AGENT, "Referer": referer}
@staticmethod
def get_list_url_params(
forum_id: int, is_good: bool = False, is_hot: bool = False, page_size: int = 20
) -> dict:
return {
"forum_id": forum_id,
"gids": 2,
"is_good": is_good,
"is_hot": is_hot,
"page_size": page_size,
"sort_type": 1,
}
@staticmethod @staticmethod
def get_images_params( def get_images_params(
resize: int = 600, resize: int = 600,
@ -76,32 +58,16 @@ class Hyperion:
) )
return [PostRecommend(**data) for data in response["list"]] 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:
params = {
"collection_id": collection_id,
"gids": gids,
"order_type": order_type,
}
response = await self.client.get(
url=self.POST_FULL_IN_COLLECTION_URL, params=params
)
return response
async def get_post_info(self, gids: int, post_id: int, read: int = 1) -> PostInfo: async def get_post_info(self, gids: int, post_id: int, read: int = 1) -> PostInfo:
params = {"gids": gids, "post_id": post_id, "read": read} params = {"gids": gids, "post_id": post_id, "read": read}
response = await self.client.get(self.POST_FULL_URL, params=params) response = await self.client.get(self.POST_FULL_URL, params=params)
return PostInfo.paste_data(response) return PostInfo.paste_data(response)
async def get_new_list(self, gids: int, type_id: int, page_size: int = 20):
"""
?gids=2&page_size=20&type=3
:return:
"""
params = {"gids": gids, "page_size": page_size, "type": type_id}
response = await self.client.get(url=self.GET_NEW_LIST_URL, params=params)
return response
async def close(self): async def close(self):
await self.client.shutdown() await self.client.shutdown()
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.close()

94
src/api/i18n.py Normal file
View File

@ -0,0 +1,94 @@
from enum import Enum
class I18nLang(str, Enum):
ZH_CN = "zh-cn"
ZH_TW = "zh-tw"
DE_DE = "de-de"
EN_US = "en-us"
ES_ES = "es-es"
FR_FR = "fr-fr"
ID_ID = "id-id"
IT_IT = "it-it"
JA_JP = "ja-jp"
KO_KR = "ko-kr"
PT_PT = "pt-pt"
RU_RU = "ru-ru"
TH_TH = "th-th"
TR_TR = "tr-tr"
VI_VN = "vi-vn"
i18n_map = {
I18nLang.ZH_CN: {
"view": "查看原文",
"author": "作者信息",
},
I18nLang.ZH_TW: {
"view": "查看原文",
"author": "作者信息",
},
I18nLang.DE_DE: {
"view": "Originaltext anzeigen",
"author": "Informationen zum Autor",
},
I18nLang.EN_US: {
"view": "View Original",
"author": "Author Information",
},
I18nLang.ES_ES: {
"view": "Ver original",
"author": "Información del autor",
},
I18nLang.FR_FR: {
"view": "Voir l'original",
"author": "Informations sur l'auteur",
},
I18nLang.ID_ID: {
"view": "Lihat aslinya",
"author": "Informasi penulis",
},
I18nLang.IT_IT: {
"view": "Visualizza originale",
"author": "Informazioni sull'autore",
},
I18nLang.JA_JP: {
"view": "元の記事を見る",
"author": "作者情報",
},
I18nLang.KO_KR: {
"view": "원본 보기",
"author": "작성자 정보",
},
I18nLang.PT_PT: {
"view": "Ver original",
"author": "Informações do autor",
},
I18nLang.RU_RU: {
"view": "Посмотреть оригинал",
"author": "Информация об авторе",
},
I18nLang.TH_TH: {
"view": "ดูต้นฉบับ",
"author": "ข้อมูลผู้เขียน",
},
I18nLang.TR_TR: {
"view": "Orijinali Görüntüle",
"author": "Yazar Bilgisi",
},
I18nLang.VI_VN: {
"view": "Xem bản gốc",
"author": "Thông tin tác giả",
},
}
class I18n:
def __init__(self, lang: str = "zh-cn"):
self.lang = I18nLang(lang)
def get_property(self, name: str):
return i18n_map.get(self.lang, {}).get(name, "")
def __getitem__(self, item):
return self.get_property(item)

View File

@ -1,11 +1,19 @@
from datetime import datetime
from enum import Enum from enum import Enum
from typing import Any, List, Optional from typing import Any, List, Optional
from pydantic import BaseModel, PrivateAttr from pydantic import BaseModel, PrivateAttr, Field, AliasChoices
GAME_ID_MAP = {"bh3": 1, "ys": 2, "bh2": 3, "wd": 4, "dby": 5, "sr": 6, "zzz": 8}
GAME_STR_MAP = {1: "bh3", 2: "ys", 3: "bh2", 4: "wd", 5: "dby", 6: "sr", 8: "zzz"}
CHANNEL_MAP = {"ys": "yuanshen", "sr": "HSRCN", "zzz": "ZZZNewsletter"}
__all__ = ( __all__ = (
"GAME_ID_MAP",
"GAME_STR_MAP",
"CHANNEL_MAP",
"PostStat", "PostStat",
"PostType", "PostType",
"HoYoPostMultiLang",
"PostInfo", "PostInfo",
"PostRecommend", "PostRecommend",
) )
@ -13,7 +21,9 @@ __all__ = (
class PostStat(BaseModel): class PostStat(BaseModel):
reply_num: int = 0 reply_num: int = 0
forward_num: int = 0 forward_num: int = Field(
default=0, validation_alias=AliasChoices("forward_num", "share_num")
)
like_num: int = 0 like_num: int = 0
view_num: int = 0 view_num: int = 0
bookmark_num: int = 0 bookmark_num: int = 0
@ -22,6 +32,14 @@ class PostStat(BaseModel):
class PostTopic(BaseModel): class PostTopic(BaseModel):
id: int id: int
name: str name: str
game_id_: int
hoyolab: bool
@property
def url(self) -> str:
if not self.hoyolab:
return f"https://www.miyoushe.com/{self.game_id_}/topicDetail/{self.id}"
return f"https://www.hoyolab.com/topicDetail/{self.id}"
class PostType(int, Enum): class PostType(int, Enum):
@ -32,26 +50,70 @@ class PostType(int, Enum):
VIDEO = 5 VIDEO = 5
class HoYoPostVideo(BaseModel):
id: str
cover: Optional[str]
url: str
@property
def is_youtube(self) -> bool:
return "www.youtube.com" in self.url
class HoYoPostMultiLang(BaseModel):
lang_subject: dict
class PostInfo(BaseModel): class PostInfo(BaseModel):
_data: dict = PrivateAttr() _data: dict = PrivateAttr()
hoyolab: bool
post_id: int post_id: int
user_uid: int user_uid: int
subject: str subject: str
image_urls: List[str] image_urls: List[str]
created_at: int created_at: datetime
video_urls: List[str] video_urls: List[str]
content: str content: str
cover: Optional[str] cover: Optional[str]
game_id: int
topics: List[PostTopic] topics: List[PostTopic]
view_type: PostType view_type: PostType
stat: PostStat stat: PostStat
video: Optional[HoYoPostVideo] = None
def __init__(self, _data: dict, **data: Any): def __init__(self, _data: dict, **data: Any):
super().__init__(**data) super().__init__(**data)
self._data = _data self._data = _data
@property
def game_id_str(self) -> str:
return GAME_STR_MAP.get(self.game_id, "")
@property
def url_start(self) -> str:
if not self.hoyolab:
return f"{self.game_id_str}/article"
return "article"
@property
def url_path(self) -> str:
return f"{self.url_start}/{self.post_id}"
@property
def url(self) -> str:
if not self.hoyolab:
return f"https://www.miyoushe.com/{self.url_path}"
return f"https://www.hoyolab.com/{self.url_path}"
@property
def author_url(self) -> str:
author = self._data["post"]["user"]
if not self.hoyolab:
return f"https://www.miyoushe.com/{self.game_id_str}/accountCenter/postList?id={author['uid']}"
return f"https://www.hoyolab.com/accountCenter/postList?id={author['uid']}"
@classmethod @classmethod
def paste_data(cls, data: dict) -> "PostInfo": def paste_data(cls, data: dict, hoyolab: bool = False) -> "PostInfo":
_data_post = data["post"] _data_post = data["post"]
post = _data_post["post"] post = _data_post["post"]
post_id = post["post_id"] post_id = post["post_id"]
@ -62,18 +124,33 @@ class PostInfo(BaseModel):
for image in image_list for image in image_list
if abs(image["width"] - image["height"]) < 1300 if abs(image["width"] - image["height"]) < 1300
] ]
vod_list = _data_post["vod_list"] vod_list = _data_post.get("vod_list", [])
video_urls = [vod["resolutions"][-1]["url"] for vod in vod_list] video_urls = [vod["resolutions"][-1]["url"] for vod in vod_list]
created_at = post["created_at"] created_at = post["created_at"]
user = _data_post["user"] # 用户数据 user = _data_post["user"] # 用户数据
user_uid = user["uid"] # 用户ID user_uid = user["uid"] # 用户ID
content = post["content"] content = post["content"]
cover = post["cover"] cover = post["cover"]
topics = [PostTopic(**topic) for topic in _data_post["topics"]] cover_list = _data_post.get("cover_list", [])
if (not cover) and cover_list:
cover = cover_list[0]["url"]
if (not cover) and image_urls:
cover = image_urls[0]
game_id = post["game_id"]
topics = [
PostTopic(game_id_=game_id, hoyolab=hoyolab, **topic)
for topic in _data_post["topics"]
]
view_type = PostType(post["view_type"]) view_type = PostType(post["view_type"])
stat = PostStat(**_data_post["stat"]) stat = PostStat(**_data_post["stat"])
video = (
None
if _data_post.get("video") is None
else HoYoPostVideo(**_data_post["video"])
)
return PostInfo( return PostInfo(
_data=data, _data=data,
hoyolab=hoyolab,
post_id=post_id, post_id=post_id,
user_uid=user_uid, user_uid=user_uid,
subject=subject, subject=subject,
@ -82,9 +159,11 @@ class PostInfo(BaseModel):
created_at=created_at, created_at=created_at,
content=content, content=content,
cover=cover, cover=cover,
game_id=game_id,
topics=topics, topics=topics,
view_type=view_type, view_type=view_type,
stat=stat, stat=stat,
video=video,
) )
def __getitem__(self, item): def __getitem__(self, item):
@ -94,5 +173,6 @@ class PostInfo(BaseModel):
class PostRecommend(BaseModel): class PostRecommend(BaseModel):
post_id: int post_id: int
subject: str subject: str
banner: Optional[str] banner: Optional[str] = None
official_type: Optional[int] official_type: Optional[int] = None
multi_language_info: Optional[HoYoPostMultiLang] = None

View File

@ -7,3 +7,5 @@ load_dotenv()
DEBUG = os.getenv("DEBUG", "True").lower() == "true" DEBUG = os.getenv("DEBUG", "True").lower() == "true"
DOMAIN = os.getenv("DOMAIN", "127.0.0.1") DOMAIN = os.getenv("DOMAIN", "127.0.0.1")
PORT = int(os.getenv("PORT", 8080)) PORT = int(os.getenv("PORT", 8080))
MIYOUSHE = os.getenv("MIYOUSHE", "True").lower() == "true"
HOYOLAB = os.getenv("HOYOLAB", "True").lower() == "true"

View File

@ -1,20 +1,25 @@
import json import json
from datetime import datetime from typing import Union, List, Dict, Callable
from typing import Union, List, Dict, Optional
from bs4 import BeautifulSoup, Tag, PageElement from bs4 import BeautifulSoup, Tag, PageElement
from src import template_env from src import template_env
from src.api.hyperion import Hyperion from src.api.hyperion import Hyperion
from src.api.models import PostStat, PostInfo, PostType, PostRecommend from src.api.i18n import I18n
from src.env import DOMAIN from src.api.models import (
PostStat,
PostInfo,
PostType,
PostRecommend,
CHANNEL_MAP,
GAME_ID_MAP,
)
from src.env import DOMAIN, MIYOUSHE
from src.error import ArticleNotFoundError from src.error import ArticleNotFoundError
from src.log import logger from src.log import logger
from src.services.scheduler import scheduler 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]] = {} RECOMMEND_POST_MAP: Dict[str, List[PostRecommend]] = {}
CHANNEL_MAP = {"ys": "yuanshen", "sr": "HSRCN", "zzz": "ZZZNewsletter"}
template = template_env.get_template("article.jinja2") template = template_env.get_template("article.jinja2")
@ -64,6 +69,11 @@ def parse_tag(tag: Union[Tag, PageElement], post_info: PostInfo) -> str:
if text := parse_tag(tag_, post_info): if text := parse_tag(tag_, post_info):
post_text.append(text) post_text.append(text)
return "<p>" + "\n".join(post_text) + "</p>" return "<p>" + "\n".join(post_text) + "</p>"
elif tag.name == "iframe":
src = tag.get("src")
if src and "https://www.youtube.com" in src:
return str(tag)
return ""
elif tag.name == "div": elif tag.name == "div":
post_text = [] post_text = []
for tag_ in tag.children: for tag_ in tag.children:
@ -96,43 +106,48 @@ def parse_stat(stat: PostStat):
) )
def get_recommend_post(game_id: str, post_id: Optional[int]) -> List[PostRecommend]: def get_recommend_post(post_info: PostInfo, _: I18n) -> List[PostRecommend]:
posts = RECOMMEND_POST_MAP.get(game_id, []) posts = RECOMMEND_POST_MAP.get(post_info.game_id_str, [])
if post_id: if post_info.post_id:
return [post for post in posts if post.post_id != post_id] return [post for post in posts if post.post_id != post_info.post_id]
return posts return posts
def get_public_data(game_id: str, post_id: int, post_info: PostInfo) -> Dict: def get_public_data(
cover = post_info.cover post_info: PostInfo,
if (not post_info.cover) and post_info.image_urls: related_posts: Callable[[PostInfo, I18n], List[PostRecommend]],
cover = post_info.image_urls[0] i18n: I18n,
) -> Dict:
return { return {
"url": f"https://www.miyoushe.com/{game_id}/article/{post_id}", "published_time": post_info.created_at.strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
"published_time": datetime.fromtimestamp(post_info.created_at).strftime( "channel": CHANNEL_MAP.get(post_info.game_id, "HSRCN"),
"%Y-%m-%dT%H:%M:%S.%fZ"
),
"channel": CHANNEL_MAP.get(game_id, "HSRCN"),
"stat": parse_stat(post_info.stat), "stat": parse_stat(post_info.stat),
"game_id": game_id,
"cover": cover,
"post": post_info, "post": post_info,
"author": post_info["post"]["user"], "author": post_info["post"]["user"],
"related_posts": get_recommend_post(game_id, post_id), "related_posts": related_posts(post_info, i18n),
"DOMAIN": DOMAIN, "DOMAIN": DOMAIN,
"i18n": i18n,
} }
async def process_article_text(game_id: str, post_id: int, post_info: PostInfo) -> str: async def process_article_text(
post_info: PostInfo,
related_posts: Callable[[PostInfo, I18n], List[PostRecommend]],
i18n: I18n,
) -> str:
post_soup = BeautifulSoup(post_info.content, features="lxml") post_soup = BeautifulSoup(post_info.content, features="lxml")
return template.render( return template.render(
description=get_description(post_soup), description=get_description(post_soup),
article=parse_content(post_soup, post_info), article=parse_content(post_soup, post_info),
**get_public_data(game_id, post_id, post_info), **get_public_data(post_info, related_posts, i18n),
) )
async def process_article_image(game_id: str, post_id: int, post_info: PostInfo) -> str: async def process_article_image(
post_info: PostInfo,
related_posts: Callable[[PostInfo, I18n], List[PostRecommend]],
i18n: I18n,
) -> str:
json_data = json.loads(post_info.content) json_data = json.loads(post_info.content)
description = json_data.get("describe", "") description = json_data.get("describe", "")
article = "" article = ""
@ -144,38 +159,34 @@ async def process_article_image(game_id: str, post_id: int, post_info: PostInfo)
return template.render( return template.render(
description=description, description=description,
article=article, article=article,
**get_public_data(game_id, post_id, post_info), **get_public_data(post_info, related_posts, i18n),
) )
async def process_article(game_id: str, post_id: int) -> str: async def process_article(game_id: str, post_id: int, i18n: I18n = I18n()) -> str:
gids = GAME_ID_MAP.get(game_id) gids = GAME_ID_MAP.get(game_id)
if not gids: if not gids:
raise ArticleNotFoundError(game_id, post_id) raise ArticleNotFoundError(game_id, post_id)
hyperion = Hyperion() async with Hyperion() as hyperion:
try:
post_info = await hyperion.get_post_info(gids=gids, post_id=post_id) post_info = await hyperion.get_post_info(gids=gids, post_id=post_id)
finally:
await hyperion.close()
if post_info.view_type in [PostType.TEXT, PostType.VIDEO]: if post_info.view_type in [PostType.TEXT, PostType.VIDEO]:
content = await process_article_text(game_id, post_id, post_info) content = await process_article_text(post_info, get_recommend_post, i18n)
elif post_info.view_type == PostType.IMAGE: elif post_info.view_type == PostType.IMAGE:
content = await process_article_image(game_id, post_id, post_info) content = await process_article_image(post_info, get_recommend_post, i18n)
return content # noqa return content # noqa
if MIYOUSHE:
@scheduler.scheduled_job("cron", minute="0", second="10") @scheduler.scheduled_job("cron", minute="0", second="10")
async def refresh_recommend_posts(): async def refresh_recommend_posts():
logger.info("Start to refresh recommend posts") logger.info("Start to refresh recommend posts")
hyperion = Hyperion() async with Hyperion() as hyperion:
try:
for key, gids in GAME_ID_MAP.items(): for key, gids in GAME_ID_MAP.items():
try: try:
RECOMMEND_POST_MAP[key] = await hyperion.get_official_recommended_posts( RECOMMEND_POST_MAP[
gids key
) ] = await hyperion.get_official_recommended_posts(gids)
except Exception as _: except Exception as _:
logger.exception(f"Failed to get recommend posts gids={gids}") logger.exception(f"Failed to get recommend posts gids={gids}")
finally:
await hyperion.close()
logger.info("Finish to refresh recommend posts") logger.info("Finish to refresh recommend posts")

View File

@ -0,0 +1,88 @@
import json
from typing import Dict, List, Callable
from src.api.hoyolab import Hoyolab
from src.api.i18n import I18n
from src.api.models import PostRecommend, PostType, PostInfo
from src.env import HOYOLAB
from src.log import logger
from src.render.article import (
process_article_text,
process_article_image,
template,
get_public_data,
)
from src.services.scheduler import scheduler
GAME_ID_MAP = {"bh3": 1, "ys": 2, "wd": 4, "dby": 5, "sr": 6, "zzz": 8}
RECOMMEND_POST_MAP: Dict[int, List[PostRecommend]] = {}
def get_recommend_post(post_info: PostInfo, i18n: I18n) -> List[PostRecommend]:
posts = RECOMMEND_POST_MAP.get(post_info.game_id, [])
return [
PostRecommend(
post_id=post.post_id,
subject=post.multi_language_info.lang_subject.get(
i18n.lang.value, post.subject
)
if post.multi_language_info
else post.subject,
)
for post in posts
if post.post_id != post_info.post_id
]
async def process_article_video(
post_info: PostInfo,
related_posts: Callable[[PostInfo, I18n], List[PostRecommend]],
i18n: I18n,
) -> str:
json_data = json.loads(post_info.content)
description = json_data.get("describe", "")
article = ""
if post_info.video and post_info.video.is_youtube:
article += f'<iframe allow="accelerometer; autoplay; encrypted-media; gyroscope; picture-in-picture" allowfullscreen="true" border="0" frameborder="0" framespacing="0" scrolling="no" src="{post_info.video.url}"></iframe>\n'
if description:
article += f"<p>{description}</p>\n"
return template.render(
description=description,
article=article,
**get_public_data(post_info, related_posts, i18n),
)
async def process_article(post_id: int, lang: str) -> str:
try:
i18n = I18n(lang)
except ValueError:
i18n = I18n()
async with Hoyolab() as hoyolab:
post_info = await hoyolab.get_post_info(post_id=post_id, lang=i18n.lang.value)
if post_info.view_type == PostType.TEXT:
content = await process_article_text(post_info, get_recommend_post, i18n)
elif post_info.view_type == PostType.IMAGE:
content = await process_article_image(post_info, get_recommend_post, i18n)
elif post_info.view_type == PostType.VIDEO:
content = await process_article_video(post_info, get_recommend_post, i18n)
return content # noqa
if HOYOLAB:
@scheduler.scheduled_job("cron", minute="0", second="10")
async def refresh_hoyo_recommend_posts():
logger.info("Start to refresh hoyolab recommend posts")
async with Hoyolab() as hoyolab:
for gids in GAME_ID_MAP.values():
temp = []
for k in (1, 2, 3):
try:
temp.extend(await hoyolab.get_news_recommend(gids, type_=k))
except Exception as _:
logger.exception(
f"Failed to get recommend posts gids={gids} type={k}"
)
RECOMMEND_POST_MAP[gids] = temp
logger.info("Finish to refresh hoyolab recommend posts")

View File

@ -1,8 +1,18 @@
from src.env import MIYOUSHE, HOYOLAB
def get_routes(): def get_routes():
from .article import parse_article
from .error import validation_exception_handler from .error import validation_exception_handler
return [ routes = [
parse_article,
validation_exception_handler, validation_exception_handler,
] ]
if MIYOUSHE:
from .article import parse_article
routes.append(parse_article)
if HOYOLAB:
from .article_hoyolab import parse_hoyo_article
routes.append(parse_hoyo_article)

View File

@ -0,0 +1,24 @@
from starlette.requests import Request
from starlette.responses import HTMLResponse
from .base import get_redirect_response
from ..app import app
from ..error import ArticleError, ResponseException
from ..log import logger
from ..render.article_hoyolab import process_article
@app.get("/article/{post_id}")
@app.get("/article/{post_id}/{lang}")
async def parse_hoyo_article(post_id: int, request: Request, lang: str = "zh-cn"):
try:
return HTMLResponse(await process_article(post_id, lang))
except ResponseException as e:
logger.warning(e.message)
return get_redirect_response(request)
except ArticleError as e:
logger.warning(e.msg)
return get_redirect_response(request)
except Exception as _:
logger.exception(f"Failed to get article {post_id} lang {lang}")
return get_redirect_response(request)

View File

@ -3,15 +3,19 @@ from typing import TYPE_CHECKING
from starlette.middleware.base import BaseHTTPMiddleware from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import RedirectResponse from starlette.responses import RedirectResponse
from src.env import MIYOUSHE
if TYPE_CHECKING: if TYPE_CHECKING:
from starlette.middleware.base import RequestResponseEndpoint from starlette.middleware.base import RequestResponseEndpoint
from starlette.requests import Request from starlette.requests import Request
from starlette.responses import Response from starlette.responses import Response
BASE_URL = "https://www.miyoushe.com" if MIYOUSHE else "https://www.hoyolab.com"
def get_redirect_response(request: "Request") -> RedirectResponse: def get_redirect_response(request: "Request") -> RedirectResponse:
path = request.url.path path = request.url.path
return RedirectResponse(url=f"https://www.miyoushe.com{path}", status_code=302) return RedirectResponse(url=f"{BASE_URL}{path}", status_code=302)
class UserAgentMiddleware(BaseHTTPMiddleware): class UserAgentMiddleware(BaseHTTPMiddleware):

View File

@ -13,16 +13,16 @@ Embed MiYouShe posts, videos, polls, and more on Telegram
<head> <head>
<meta charset="UTF-8"> <meta charset="UTF-8">
<title></title> <title></title>
<link rel="canonical" href="{{ url }}"/> <link rel="canonical" href="{{ post.url }}"/>
<meta property="theme-color" content="#00a8fc"/> <meta property="theme-color" content="#00a8fc"/>
<meta property="twitter:site" content="{{ author.nickname }}"/> <meta property="twitter:site" content="{{ author.nickname }}"/>
<meta property="twitter:creator" content="{{ author.nickname }}"/> <meta property="twitter:creator" content="{{ author.nickname }}"/>
<meta property="twitter:title" content="{{ post.subject }} ({{ author.nickname }})"/> <meta property="twitter:title" content="{{ post.subject }} ({{ author.nickname }})"/>
<meta property="twitter:image" content="{{ cover }}"/> <meta property="twitter:image" content="{{ post.cover }}"/>
<meta property="twitter:card" content="summary_large_image"/> <meta property="twitter:card" content="summary_large_image"/>
<meta property="og:url" content="{{ url }}"/> <meta property="og:url" content="{{ post.url }}"/>
<meta property="og:image" content="{{ cover }}"/> <meta property="og:image" content="{{ post.cover }}"/>
<meta property="og:title" content="{{ post.subject }} ({{ author.nickname }})"/> <meta property="og:title" content="{{ post.subject }} ({{ author.nickname }})"/>
<meta property="og:description" content="{{ description }}"/> <meta property="og:description" content="{{ description }}"/>
<meta property="og:site_name" content="{{ post.subject }} - {{ author.nickname }} - 米游社"/> <meta property="og:site_name" content="{{ post.subject }} - {{ author.nickname }} - 米游社"/>
@ -39,7 +39,7 @@ Embed MiYouShe posts, videos, polls, and more on Telegram
</section> </section>
<section class="section--first"> <section class="section--first">
If you can see this, your browser is doing something weird with your user agent. If you can see this, your browser is doing something weird with your user agent.
<a href="{{ url }}">View original post</a> <a href="{{ post.url }}">View original post</a>
</section> </section>
<article> <article>
<!-- article content --> <!-- article content -->
@ -50,28 +50,33 @@ Embed MiYouShe posts, videos, polls, and more on Telegram
{% if post.topics %} {% if post.topics %}
<p> <p>
{% for topic in post.topics %} {% for topic in post.topics %}
<a href="https://www.miyoushe.com/{{ game_id }}/topicDetail/{{ topic.id }}">#{{ topic.name }} </a> <a href="{{ topic.url }}">#{{ topic.name }} </a>
{% endfor %} {% endfor %}
</p> </p>
{% endif %} {% endif %}
<p><a href="{{ url }}">查看原文</a></p> <p><a href="{{ post.url }}">{{ i18n.view }}</a></p>
<!-- author --> <!-- author -->
<details> <details>
<summary>作者信息</summary> <summary>{{ i18n.author }}</summary>
{% if author.avatar_url %} {% if author.avatar_url %}
<img src="{{ author.avatar_url }}" alt="profile picture"/> <img src="{{ author.avatar_url }}" alt="profile picture"/>
{% endif %} {% endif %}
<h2>{{ author.nickname }}</h2> <h2>{{ author.nickname }}</h2>
<p> <p>
<a href="https://www.miyoushe.com/{{ game_id }}/accountCenter/postList?id={{ author.uid }}">@{{ author.nickname }}</a> <a href="{{ post.author_url }}">@{{ author.nickname }}</a>
lv.{{ author.level_exp.level }} lv.{{ author.level_exp.level }}
</p> </p>
</details> </details>
<!-- related posts --> <!-- related posts -->
{% if related_posts %} {% if related_posts %}
{% for post in related_posts %} {% for post_ in related_posts %}
<related> <related>
<a href="https://{{ DOMAIN }}/{{ game_id }}/article/{{ post.post_id }}">{{ post.subject }}</a> {% if i18n.lang.value == 'zh-cn' %}
{% set ends = '' %}
{% else %}
{% set ends = '/' + i18n.lang.value %}
{% endif %}
<a href="https://{{ DOMAIN }}/{{ post.url_start }}/{{ post_.post_id }}{{ ends }}">{{ post_.subject }}</a>
</related> </related>
<br/> <br/>
{% endfor %} {% endfor %}

0
src/utils/__init__.py Normal file
View File

0
src/utils/article.py Normal file
View File