mirror of
https://github.com/NanmiCoder/MediaCrawler.git
synced 2026-09-22 01:48:11 +08:00
feat(ks): 支持分享短链 /f/<token> 形式的视频输入
快手分享短链 https://www.kuaishou.com/f/X9Idt15MQb9L2cv 路径里是 share_token 而不是视频 ID,它只做 302 跳转,真实地址在 Location 里。原先的解析器只认 /short-video/<id> 和纯 ID,遇到短链会抛 ValueError 被 continue 掉——只有一行 ERROR 日志,看起来像"爬了但没数据"。 - help.py: 新增 /f/<token> 分支,返回 url_type="short"(token 不是视频 ID, 必须跟随重定向),并补上第三种形式的文档 - client.py: 新增 resolve_short_url,GET 时 follow_redirects=False,读 301/302/303/307/308 的 Location - core.py: url_type == "short" 时先解析短链再解析一次,失败则跳过该条 - 新增 tests/test_kuaishou_url_parse.py(6 条,不发网络请求) 验证:两条真实短链分别解析到 3xyziwesje8e9jg / 3xbbkdxtxqm8sae,详情均成功; 带 query 的标准视频页不会被误判成短链。
This commit is contained in:
@@ -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/<id> 形式的真实地址。
|
||||
"""
|
||||
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
|
||||
|
||||
@@ -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/<id>
|
||||
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:
|
||||
|
||||
@@ -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}")
|
||||
|
||||
|
||||
|
||||
@@ -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/<id>``
|
||||
3. 分享短链 ``/f/<share_token>`` —— 注意路径里是 **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")
|
||||
Reference in New Issue
Block a user