diff --git a/src/nonebot_plugin_parser/parsers/douyin/__init__.py b/src/nonebot_plugin_parser/parsers/douyin/__init__.py index 9f2a65d3..cabebebb 100644 --- a/src/nonebot_plugin_parser/parsers/douyin/__init__.py +++ b/src/nonebot_plugin_parser/parsers/douyin/__init__.py @@ -1,8 +1,8 @@ import re +import json from typing import ClassVar from httpx import AsyncClient -from nonebot import logger from ..base import ( COMMON_TIMEOUT, @@ -12,11 +12,21 @@ ParseException, handle, ) +from .aweme import decoder class DouyinParser(BaseParser): platform: ClassVar[Platform] = Platform(name=PlatformEnum.DOUYIN, display_name="抖音") + def __init__(self): + super().__init__() + self.headers.update( + { + "Origin": "https://open.douyin.com", + "Referer": "https://open.douyin.com/", + } + ) + # https://v.douyin.com/_2ljF4AmKL8 @handle("v.douyin", r"v\.douyin\.com/[a-zA-Z0-9_\-]+") @handle("jx.douyin", r"jx\.douyin\.com/[a-zA-Z0-9_\-]+") @@ -26,112 +36,83 @@ async def _parse_short_link(self, searched: re.Match[str]): # https://www.douyin.com/video/7521023890996514083 # https://www.douyin.com/note/7469411074119322899 - @handle("douyin", r"douyin\.com/(?Pvideo|note)/(?P\d+)") - @handle("iesdouyin", r"iesdouyin\.com/share/(?Pslides|video|note)/(?P\d+)") - @handle("m.douyin", r"m\.douyin\.com/share/(?Pslides|video|note)/(?P\d+)") + @handle("douyin", r"douyin\.com/[a-z]+/(?P\d+)") + @handle("iesdouyin", r"iesdouyin\.com/share/[a-z]+/(?P\d+)") + @handle("m.douyin", r"m\.douyin\.com/share/[a-z]+/(?P\d+)") # https://jingxuan.douyin.com/m/video/7574300896016862490?app=yumme&utm_source=copy_link - @handle("jingxuan.douyin", r"jingxuan\.douyin.com/m/(?Pslides|video|note)/(?P\d+)") + @handle("jingxuan.douyin", r"jingxuan\.douyin.com/m/[a-z]+/(?P\d+)") async def _parse_douyin(self, searched: re.Match[str]): - ty, vid = searched.group("ty"), searched.group("vid") - if ty == "slides": - return await self.parse_slides(vid) - - for url in (self._build_m_douyin_url(ty, vid), self._build_iesdouyin_url(ty, vid)): - try: - return await self.parse_video(url) - except ParseException as e: - logger.warning(f"failed to parse {url}, error: {e}") - continue - raise ParseException("分享已删除或资源直链提取失败, 请稍后再试") - - @staticmethod - def _build_iesdouyin_url(ty: str, vid: str) -> str: - return f"https://www.iesdouyin.com/share/{ty}/{vid}" - - @staticmethod - def _build_m_douyin_url(ty: str, vid: str) -> str: - return f"https://m.douyin.com/share/{ty}/{vid}" - - async def parse_video(self, url: str): - from . import video + aweme_id = searched.group("aweme_id") + return await self.parse_aweme(aweme_id) + async def parse_aweme(self, aweme_id: str): async with AsyncClient( - headers=self.ios_headers, + headers=self.headers, timeout=COMMON_TIMEOUT, - follow_redirects=False, + follow_redirects=True, verify=False, ) as client: - response = await client.get(url) + response = await client.get( + "https://www.douyin.com/aweme/v1/web/aweme/detail/", + params={"aweme_id": aweme_id, "aid": "6383"}, + ) if response.status_code != 200: raise ParseException(f"status: {response.status_code}") - text = response.text - - pattern = re.compile( - pattern=r"window\._ROUTER_DATA\s*=\s*(.*?)", - flags=re.DOTALL, - ) - matched = pattern.search(text) - - if not matched or not matched.group(1): - raise ParseException("can't find _ROUTER_DATA in html") - - video_data = video.decoder.decode(matched.group(1).strip()).video_data + aweme = decoder.decode(response.content).aweme_detail # 作者 author = self.create_author( - video_data.author.nickname, - video_data.avatar_url, + aweme.author.nickname, + aweme.author.avatar_thumb.url_list[-1], + aweme.author.signature, ) # 先以部分数据构建结果，后续再填充内容，避免使用临时变量 result = self.result( - title=video_data.desc, + text=aweme.share_info.text, author=author, - timestamp=video_data.create_time, + timestamp=aweme.create_time, + url=aweme.share_url.split("?")[0], ) + if music := aweme.music: + if not music.is_original_sound: + if music.play_url.uri == "": + extra = json.loads(music.extra) + music_url = extra.get("original_song_url") + else: + music_url = music.play_url.uri + if music_url: + result.contents.append( + self.create_audio( + url_or_task=music_url, + duration=music.duration, + ) + ) # 添加图片内容 - if image_urls := video_data.image_urls: - result.contents.extend(self.create_images(image_urls)) + if images := aweme.images: + for image in images: + if image.clip_type == 2 or image.clip_type is None: + result.contents.append( + self.create_image( + url_or_task=image.url_list[-1], + ) + ) + elif image_video := image.video: + result.contents.extend( + [ + self.create_image( + url_or_task=image_video.cover.url_list[-1], + ), + self.create_video(url_or_task=image_video.play_addr.url), + ] + ) # 添加视频内容 - elif video_url := video_data.video_url: + elif video := aweme.video: result.video = self.create_video( - video_url, - video_data.cover_url, - video_data.duration, + video.play_addr.url, + video.cover_original_scale.url_list[-1] if video.cover_original_scale else video.cover.url_list[-1], + video.duration // 1000, ) return result - - async def parse_slides(self, video_id: str): - from . import slides - - url = "https://www.iesdouyin.com/web/api/v2/aweme/slidesinfo/" - params = { - "aweme_ids": f"[{video_id}]", - "request_source": "200", - } - async with AsyncClient(headers=self.android_headers, verify=False) as client: - response = await client.get(url, params=params) - response.raise_for_status() - - slides_data = slides.decoder.decode(response.content).aweme_details[0] - - # 作者 - author = self.create_author(slides_data.name, slides_data.avatar_url) - - # 先以部分数据构建结果，后续再填充内容，避免使用临时变量 - result = self.result( - title=slides_data.desc, - author=author, - timestamp=slides_data.create_time, - ) - - # 优先取动图 - if dynamic_urls := slides_data.dynamic_urls: - for dynamic_url in dynamic_urls: - result.contents.append(self.create_gif(dynamic_url)) - elif image_urls := slides_data.image_urls: - result.contents.extend(self.create_images(image_urls)) - - return result diff --git a/src/nonebot_plugin_parser/parsers/douyin/aweme.py b/src/nonebot_plugin_parser/parsers/douyin/aweme.py new file mode 100644 index 00000000..fe7ba075 --- /dev/null +++ b/src/nonebot_plugin_parser/parsers/douyin/aweme.py @@ -0,0 +1,82 @@ +from msgspec import Struct, field +from msgspec.json import Decoder + + +class Addr(Struct): + uri: str + + @property + def url(self) -> str: + return f"https://aweme.snssdk.com/aweme/v1/play/?video_id={self.uri}&ratio=1080p&line=0" + + +class UrlList(Struct): + url_list: list[str] + + +class Author(Struct): + uid: str + nickname: str + avatar_thumb: UrlList + """头像""" + signature: str | None = None + + +class Video(Struct): + cover: UrlList + duration: int + """视频时长(/1000)""" + play_addr: Addr + cover_original_scale: UrlList | None = None + + @property + def url(self) -> str: + return self.play_addr.url + + +class Image(Struct): + url_list: list[str] + uri: str = field(default="") + clip_type: int | None = field(default=None) + """=2 or None 是普通图片""" + video: Video | None = field(default=None) + """Live Photo 视频""" + + +class MusicPlayUrl(Struct): + uri: str + + +class Music(Struct): + duration: int + mid: str + play_url: MusicPlayUrl + extra: str + is_original_sound: bool + + +class ShareInfo(Struct): + share_desc: str + share_desc_info: str + + @property + def text(self) -> str: + return self.share_desc_info.replace(f"#{self.share_desc}#", "", 1) + + +class Aweme(Struct): + aweme_id: str + author: Author + share_info: ShareInfo + create_time: int + share_url: str + images: list[Image] | None = field(default=None) + music: Music | None = field(default=None) + video: Video | None = field(default=None) + + +class Response(Struct): + aweme_detail: Aweme + + +decoder = Decoder(Response) diff --git a/src/nonebot_plugin_parser/parsers/douyin/slides.py b/src/nonebot_plugin_parser/parsers/douyin/slides.py deleted file mode 100644 index 6e109975..00000000 --- a/src/nonebot_plugin_parser/parsers/douyin/slides.py +++ /dev/null @@ -1,63 +0,0 @@ -from random import choice - -from msgspec import Struct, field -from msgspec.json import Decoder - - -class PlayAddr(Struct): - url_list: list[str] - - -class Cover(Struct): - url_list: list[str] - - -class Video(Struct): - play_addr: PlayAddr - cover: Cover - duration: int - - -class Image(Struct): - video: Video | None = None - url_list: list[str] = field(default_factory=list) - - -class Avatar(Struct): - url_list: list[str] - - -class Author(Struct): - nickname: str - # avatar_larger: Avatar - avatar_thumb: Avatar - - -class SlidesData(Struct): - author: Author - desc: str - create_time: int - images: list[Image] - - @property - def name(self) -> str: - return self.author.nickname - - @property - def avatar_url(self) -> str: - return choice(self.author.avatar_thumb.url_list) - - @property - def image_urls(self) -> list[str]: - return [choice(image.url_list) for image in self.images] - - @property - def dynamic_urls(self) -> list[str]: - return [choice(image.video.play_addr.url_list) for image in self.images if image.video] - - -class SlidesInfo(Struct): - aweme_details: list[SlidesData] = field(default_factory=list) - - -decoder = Decoder(SlidesInfo) diff --git a/src/nonebot_plugin_parser/parsers/douyin/video.py b/src/nonebot_plugin_parser/parsers/douyin/video.py deleted file mode 100644 index 839afd1c..00000000 --- a/src/nonebot_plugin_parser/parsers/douyin/video.py +++ /dev/null @@ -1,103 +0,0 @@ -from random import choice -from typing import Any - -from msgspec import Struct, field -from msgspec.json import Decoder - -from ..base import ParseException - - -class Avatar(Struct): - url_list: list[str] - - -class Author(Struct): - nickname: str - avatar_thumb: Avatar | None = None - avatar_medium: Avatar | None = None - - -class PlayAddr(Struct): - url_list: list[str] - - -class Cover(Struct): - url_list: list[str] - - -class Video(Struct): - play_addr: PlayAddr - cover: Cover - duration: int - - -class Image(Struct): - video: Video | None = None - url_list: list[str] = field(default_factory=list) - - -class VideoData(Struct): - create_time: int - author: Author - desc: str - images: list[Image] | None = None - video: Video | None = None - - @property - def image_urls(self) -> list[str]: - return [choice(image.url_list) for image in self.images] if self.images else [] - - @property - def video_url(self) -> str | None: - return choice(self.video.play_addr.url_list).replace("playwm", "play") if self.video else None - - @property - def cover_url(self) -> str | None: - return choice(self.video.cover.url_list) if self.video else None - - @property - def duration(self) -> int | None: - return self.video.duration // 1000 if self.video else None - - @property - def avatar_url(self) -> str | None: - if avatar := self.author.avatar_thumb: - return choice(avatar.url_list) - elif avatar := self.author.avatar_medium: - return choice(avatar.url_list) - return None - - -class VideoInfoRes(Struct): - item_list: list[VideoData] = field(default_factory=list) - - @property - def video_data(self) -> VideoData: - if len(self.item_list) == 0: - raise ParseException("can't find data in videoInfoRes") - return choice(self.item_list) - - -class VideoOrNotePage(Struct): - video_info_res: VideoInfoRes = field(name="videoInfoRes", default_factory=VideoInfoRes) - - -class LoaderData(Struct): - video_page: VideoOrNotePage | None = field(name="video_(id)/page", default=None) - note_page: VideoOrNotePage | None = field(name="note_(id)/page", default=None) - - -class RouterData(Struct): - loader_data: LoaderData = field(name="loaderData", default_factory=LoaderData) - errors: dict[str, Any] | None = None - - @property - def video_data(self) -> VideoData: - if page := self.loader_data.video_page: - return page.video_info_res.video_data - elif page := self.loader_data.note_page: - return page.video_info_res.video_data - raise ParseException("can't find video_(id)/page or note_(id)/page in router data") - - -decoder = Decoder(RouterData) diff --git a/tests/parsers/test_douyin.py b/tests/parsers/test_douyin.py index d8b4947e..edc509fe 100644 --- a/tests/parsers/test_douyin.py +++ b/tests/parsers/test_douyin.py @@ -23,10 +23,12 @@ async def test_parse(url: str) -> None: result = await parser.parse(keyword, searched) logger.debug(f"{url} | 解析结果: \n{result}") - assert result.title, "标题为空" + # assert result.title, "标题为空" + assert result.author, "作者为空" assert result.author.avatar, "作者头像不存在" assert await result.author.avatar.get(), "头像为空" + assert result.text, "内容为空" assert result.video, "视频内容为空" video_path = await result.video.path_task.get() @@ -83,8 +85,9 @@ async def test_parse(url: str) -> None: result = await parser.parse(keyword, searched) logger.debug(f"{url} | 解析结果: \n{result}") - assert result.title, "标题为空" + # assert result.title, "标题为空" assert result.author, "作者为空" + assert result.text, "内容为空" if img_contents := result.img_contents: for img_content in img_contents: # 容易失败，使用 safe_get @@ -116,7 +119,8 @@ async def test_slides(): assert searched, "无法匹配 URL" result = await parser.parse(keyword, searched) logger.debug(f"{dynamic_image_url} | 解析结果: \n{result}") - assert result.title, "标题为空" + # assert result.title, "标题为空" + assert result.text, "内容为空" video_contents = result.video_contents assert video_contents, "视频内容为空" for video_content in video_contents: @@ -130,7 +134,8 @@ async def test_slides(): assert searched, "无法匹配 URL" result = await parser.parse(keyword, searched) logger.debug(f"{static_image_url} | 解析结果: \n{result}") - assert result.title, "标题为空" + # assert result.title, "标题为空" + assert result.text, "内容为空" img_contents = result.img_contents assert img_contents, "图片内容为空" for img_content in img_contents: