diff --git a/media_platform/kuaishou/client.py b/media_platform/kuaishou/client.py index 8fde7e9..0f78040 100644 --- a/media_platform/kuaishou/client.py +++ b/media_platform/kuaishou/client.py @@ -240,6 +240,37 @@ class KuaiShouClient(AbstractApiClient, ProxyRefreshMixin): } return await self.post("", post_data) + async def resolve_short_url(self, short_url: str) -> str: + """解析快手分享短链（/f/xxx），返回重定向后的真实 URL。 + + 短链路径里的 share_token 不是视频 ID，只能靠 302 的 Location 拿到 + /short-video/ 形式的真实地址。 + """ + async with make_async_client(proxy=self.proxy, follow_redirects=False) as client: + try: + utils.logger.info( + f"[KuaiShouClient.resolve_short_url] Resolving short URL: {short_url}" + ) + response = await client.get(short_url, timeout=10, headers=self.headers) + + # 短链通常返回 302 + if response.status_code in (301, 302, 303, 307, 308): + redirect_url = response.headers.get("Location", "") + utils.logger.info( + f"[KuaiShouClient.resolve_short_url] Resolved to: {redirect_url}" + ) + return redirect_url + + utils.logger.warning( + f"[KuaiShouClient.resolve_short_url] Unexpected status code: {response.status_code}" + ) + return "" + except Exception as e: + utils.logger.error( + f"[KuaiShouClient.resolve_short_url] Failed to resolve short URL: {e}" + ) + return "" + async def get_video_info(self, photo_id: str) -> Dict: """ Kuaishou web video detail api diff --git a/media_platform/kuaishou/core.py b/media_platform/kuaishou/core.py index 0c06217..d290689 100644 --- a/media_platform/kuaishou/core.py +++ b/media_platform/kuaishou/core.py @@ -204,6 +204,24 @@ class KuaishouCrawler(AbstractCrawler): for video_url in config.KS_SPECIFIED_ID_LIST: try: video_info = parse_video_info_from_url(video_url) + + # 分享短链（/f/xxx）要先跟随 302 重定向，才能拿到 /short-video/ + if video_info.url_type == "short": + utils.logger.info( + f"[KuaishouCrawler.get_specified_videos] Resolving short link: {video_url}" + ) + resolved_url = await self.ks_client.resolve_short_url(video_url) + if resolved_url: + video_info = parse_video_info_from_url(resolved_url) + utils.logger.info( + f"[KuaishouCrawler.get_specified_videos] Short link resolved to video ID: {video_info.video_id}" + ) + else: + utils.logger.error( + f"[KuaishouCrawler.get_specified_videos] Failed to resolve short link: {video_url}" + ) + continue + video_ids.append(video_info.video_id) utils.logger.info(f"Parsed video ID: {video_info.video_id} from {video_url}") except ValueError as e: diff --git a/media_platform/kuaishou/help.py b/media_platform/kuaishou/help.py index a7bca7c..2f93c29 100644 --- a/media_platform/kuaishou/help.py +++ b/media_platform/kuaishou/help.py @@ -92,6 +92,8 @@ def parse_video_info_from_url(url: str) -> VideoUrlInfo: Supports the following formats: 1. Full video URL: "https://www.kuaishou.com/short-video/3x3zxz4mjrsc8ke?authorId=3x84qugg4ch9zhs&streamSource=search" 2. Pure video ID: "3x3zxz4mjrsc8ke" + 3. Share short link: "https://www.kuaishou.com/f/X9Idt15MQb9L2cv" + （路径里是 share_token 而非视频 ID，返回 url_type="short"，由调用方跟随重定向） Args: url: Kuaishou video link or video ID @@ -109,6 +111,14 @@ def parse_video_info_from_url(url: str) -> VideoUrlInfo: video_id = match.group(1) return VideoUrlInfo(video_id=video_id, url_type="normal") + # 分享短链：https://www.kuaishou.com/f/X9Idt15MQb9L2cv + # 路径里的 share_token 不是视频 ID，必须跟随 302 重定向才能拿到真实地址， + # 所以这里只标记类型，交给调用方解析（url_type="short"） + share_pattern = r'kuaishou\.com/f/([a-zA-Z0-9_-]+)' + match = re.search(share_pattern, url) + if match: + return VideoUrlInfo(video_id=match.group(1), url_type="short") + raise ValueError(f"Unable to parse video ID from URL: {url}") diff --git a/tests/test_kuaishou_url_parse.py b/tests/test_kuaishou_url_parse.py new file mode 100644 index 0000000..9ddab62 --- /dev/null +++ b/tests/test_kuaishou_url_parse.py @@ -0,0 +1,73 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2025 relakkes@gmail.com +# +# This file is part of MediaCrawler project. +# Repository: https://github.com/NanmiCoder/MediaCrawler/blob/main/tests/test_kuaishou_url_parse.py +# GitHub: https://github.com/NanmiCoder +# Licensed under NON-COMMERCIAL LEARNING LICENSE 1.1 +# + +"""快手视频输入解析的回归测试。 + +覆盖三种输入形态： +1. 纯视频 ID +2. 标准视频页 ``/short-video/`` +3. 分享短链 ``/f/`` —— 注意路径里是 **share_token 而不是视频 ID**， + 必须跟随 302 重定向才能拿到真实 ID，所以只标记 ``url_type="short"`` 交给调用方。 + +这里只测纯函数，不发网络请求。 +""" + +from __future__ import annotations + +import pytest + +from media_platform.kuaishou.help import parse_video_info_from_url + + +def test_pure_video_id_is_normal(): + info = parse_video_info_from_url("3xf8enb8dbj6uig") + + assert info.video_id == "3xf8enb8dbj6uig" + assert info.url_type == "normal" + + +def test_short_video_url_is_normal(): + info = parse_video_info_from_url( + "https://www.kuaishou.com/short-video/3x3zxz4mjrsc8ke" + "?authorId=3x84qugg4ch9zhs&streamSource=search&area=searchxxnull&searchKey=python" + ) + + assert info.video_id == "3x3zxz4mjrsc8ke" + assert info.url_type == "normal" + + +def test_share_short_link_is_marked_short(): + """分享短链必须标记为 short —— 路径里的 token 不是视频 ID""" + info = parse_video_info_from_url("https://www.kuaishou.com/f/X9Idt15MQb9L2cv") + + assert info.video_id == "X9Idt15MQb9L2cv" + assert info.url_type == "short" + + +def test_share_short_link_with_dash_in_token(): + info = parse_video_info_from_url("https://www.kuaishou.com/f/X-a8vLyTxvEvN2jg") + + assert info.video_id == "X-a8vLyTxvEvN2jg" + assert info.url_type == "short" + + +def test_short_video_url_is_not_confused_with_share_link(): + """带 query 的标准视频页不能被误判成短链""" + info = parse_video_info_from_url( + "https://www.kuaishou.com/short-video/3xyziwesje8e9jg" + "?shareToken=X9Idt15MQb9L2cv&shareObjectId=3xyziwesje8e9jg" + ) + + assert info.video_id == "3xyziwesje8e9jg" + assert info.url_type == "normal" + + +def test_unparsable_url_raises(): + with pytest.raises(ValueError): + parse_video_info_from_url("https://www.kuaishou.com/profile/3x84qugg4ch9zhs")