HeXi/tests/test_kuaishou_parse.py

"""快手分享页解析（kuaishou_content.py 纯函数）单元测试

只测不碰网络的解析/命名部分：INIT_STATE 提取、视频/图集 URL 组装、媒体命名。
样本仿自 2026-10-08 实测的分享页（v.kuaishou.com 短链 → v.m.chenzhongtech.com/fw/photo）。

kuaishou_content.py 有插件内相对导入（`...models` / `...utils`），无法裸加载，
故在 sys.modules 里预置父包替身 + 真实加载 utils/models 后按路径加载目标模块
（与 test_video_sender_link.py 同法）。
"""

import importlib.util
import json
import sys
import types
from pathlib import Path

import pytest

_REPO = Path(__file__).resolve().parents[1]
_PLUGIN = _REPO / "hexi" / "plugins" / "nonebot_plugin_video_analysis"
_PKG = "hexi.plugins.nonebot_plugin_video_analysis"

#: 真实加载的插件内模块（utils 纯标准库、models 只有一个异常类）
_REAL_MODULES = {
    f"{_PKG}.models": _PLUGIN / "models.py",
    f"{_PKG}.utils": _PLUGIN / "utils.py",
}
#: 只当命名空间用的父包（避免触发插件包 __init__ 的 NoneBot 初始化）
_STUB_PACKAGES = (_PKG, f"{_PKG}.services", f"{_PKG}.services.fetchers")


def _load_real(name: str, path: Path):
    spec = importlib.util.spec_from_file_location(name, path)
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    spec.loader.exec_module(module)
    return module


@pytest.fixture(scope="module")
def ks():
    saved = {name: sys.modules.get(name) for name in _STUB_PACKAGES}
    for name in _STUB_PACKAGES:
        sys.modules[name] = types.ModuleType(name)
    for name, path in _REAL_MODULES.items():
        saved[name] = sys.modules.get(name)
        _load_real(name, path)

    target = f"{_PKG}.services.fetchers.kuaishou_content"
    saved[target] = sys.modules.get(target)
    module = _load_real(target, _PLUGIN / "services" / "fetchers" / "kuaishou_content.py")

    yield module

    for name, previous in saved.items():
        if previous is None:
            sys.modules.pop(name, None)
        else:
            sys.modules[name] = previous


@pytest.fixture(scope="module")
def u(ks):
    """插件 utils（命名断言用同一套函数，别在测试里手拼）"""
    return sys.modules[f"{_PKG}.utils"]


# ───────────────────────── 样本 ─────────────────────────

KEY = "string/rest/wd/ugH5App/photo/simple/info@kpn>KUAISHOU"


def _photo(**over) -> dict:
    photo = {
        "caption": "欧巴欧巴哦 #蔚蓝档案 #砂狼白子",
        "userName": "砂狼白子",
        "userId": 2329823245,
        "photoId": "5206161376908999738",
        "duration": 15150,
        "timestamp": 1789916802288,
        "coverUrls": [
            {"cdn": "p2.a.yximgs.com", "url": "https://p2.a.yximgs.com/cover.jpg"}
        ],
        "mainMvUrls": [
            {"cdn": "tymov2.a.kwimgs.com", "url": "https://tymov2.a.kwimgs.com/a_b.mp4"},
            {"cdn": "txmov2.a.kwimgs.com", "url": "https://txmov2.a.kwimgs.com/a_b.mp4"},
        ],
        "ext_params": {"mtype": 3, "video": 15150},
    }
    photo.update(over)
    return photo


def _page(data: dict) -> str:
    payload = json.dumps(data, ensure_ascii=False)
    return f"<!doctype html><html><script>window.INIT_STATE={payload}</script></html>"


def _page_with(photo: dict | None, prelude: dict | None = None) -> str:
    data = dict(prelude or {})
    data[KEY] = {"result": 1, "photo": photo}
    return _page(data)


# ───────────────────────── INIT_STATE 提取 ─────────────────────────


def test_extract_photo_skips_empty_entries(ks):
    """INIT_STATE 是一堆缓存键的 dict，photo 在第一个非空项里（顺序不定）"""
    page = _page(
        {
            "some/cache/key": {"result": 1, "photo": None},
            "another/key": {"result": 2},
            KEY: {"result": 1, "photo": _photo()},
        }
    )
    assert ks._extract_photo(page)["photoId"] == "5206161376908999738"


def test_extract_photo_tolerates_js_literals(ks):
    """页面偶有 undefined 之类非 JSON 字面量（同小红书预处理）"""
    page = (
        "<script>window.INIT_STATE = "
        '{"k": {"result": 1, "photo": {"photoId": "1", "userName": undefined}}}'
        "</script>"
    )
    photo = ks._extract_photo(page)
    assert photo["photoId"] == "1"
    assert photo["userName"] is None


def test_extract_photo_missing_state(ks):
    """风控壳页（桌面 UA / www 站）没有 INIT_STATE → 报错而不是静默返回空"""
    err = sys.modules[f"{_PKG}.models"].ContentFetchError
    with pytest.raises(err):
        ks._extract_photo("<html><body>risk control</body></html>")


def test_extract_photo_without_photo(ks):
    """INIT_STATE 在但没有作品（已删除/私密）→ 报错"""
    err = sys.modules[f"{_PKG}.models"].ContentFetchError
    with pytest.raises(err):
        ks._extract_photo(_page({"k": {"result": 2, "photo": None}}))


# ───────────────────────── 媒体 URL ─────────────────────────


def test_first_url_prefers_first_available(ks):
    assert ks._first_url([{"cdn": "a", "url": "https://a/1.mp4"}]) == "https://a/1.mp4"
    # 首项没有 url（快手同一资源的备用项可能缺字段）时继续往后找
    assert ks._first_url([{"cdn": "a"}, {"url": "https://b/2.mp4"}]) == "https://b/2.mp4"
    assert ks._first_url([]) is None
    assert ks._first_url(None) is None


def _rep(width, height, fps, codec, bitrate, url, **over) -> dict:
    rep = {
        "width": width,
        "height": height,
        "frameRate": fps,
        "videoCodec": codec,
        "avgBitrate": bitrate,
        "url": url,
    }
    rep.update(over)
    return rep


def test_manifest_video_url_picks_highest_quality(ks):
    """清晰度梯取最优：先比分辨率，再帧率、编码（avc 优先）、码率"""
    photo = _photo(
        manifest={
            "adaptationSet": [
                {
                    "representation": [
                        # 1080p 30fps hevc 也要赢过 720p 60fps avc（分辨率优先）
                        _rep(1920, 1080, 30, "hevc", 2000, "https://c/1080.mp4"),
                        _rep(1280, 720, 60, "avc", 1597, "https://c/720p60.mp4"),
                        _rep(1280, 720, 30, "hevc", 726, "https://c/720p30.mp4"),
                    ]
                }
            ]
        }
    )
    assert ks._manifest_video_url(photo) == "https://c/1080.mp4"


def test_manifest_video_url_prefers_avc_on_tie(ks):
    """同分辨率同帧率 → avc 优先（兼容性），hevc 只在画质更高时才选"""
    photo = _photo(
        manifest={
            "adaptationSet": [
                {
                    "representation": [
                        _rep(1280, 720, 60, "avc", 1597, "https://c/avc.mp4"),
                        # 码率更高但编码是 hevc，仍输给 avc
                        _rep(1280, 720, 60, "hevc", 2000, "https://c/hevc.mp4"),
                    ]
                }
            ]
        }
    )
    assert ks._manifest_video_url(photo) == "https://c/avc.mp4"


def test_manifest_video_url_skips_broken_entries(ks):
    """缺 url / 结构不对的档跳过，不因为一条脏数据整体失败"""
    photo = _photo(
        manifest={
            "adaptationSet": [
                {"representation": [{"width": 1920, "height": 1080}, "junk"]},
                {"representation": [_rep(1280, 720, 30, "avc", 900, "https://c/720.mp4")]},
            ]
        }
    )
    assert ks._manifest_video_url(photo) == "https://c/720.mp4"
    assert ks._manifest_video_url(_photo()) is None
    assert ks._manifest_video_url(_photo(manifest={"adaptationSet": []})) is None


def test_video_url_falls_back_to_main_mv_urls(ks):
    """老页面没有 manifest → 退回 mainMvUrls（参考实现的默认档路径）"""
    assert ks._video_url(_photo()) == "https://tymov2.a.kwimgs.com/a_b.mp4"
    photo = _photo(
        manifest={"adaptationSet": [{"representation": [_rep(1920, 1080, 30, "avc", 2000, "https://c/best.mp4")]}]}
    )
    assert ks._video_url(photo) == "https://c/best.mp4"


def test_atlas_image_urls(ks):
    photo = _photo(
        ext_params={
            "atlas": {
                "cdnList": [{"cdn": "p1.a.yximgs.com"}, {"cdn": "p2.a.yximgs.com"}],
                "list": ["upic/a.jpg", "upic/b.jpg"],
            }
        }
    )
    assert ks._atlas_image_urls(photo) == [
        "https://p1.a.yximgs.com/upic/a.jpg",
        "https://p1.a.yximgs.com/upic/b.jpg",
    ]


def test_atlas_image_urls_absent(ks):
    """视频作品的 ext_params 里没有 atlas → 不是图集"""
    assert ks._atlas_image_urls(_photo()) == []
    assert ks._atlas_image_urls(_photo(ext_params={})) == []
    # 有 atlas 但缺图床/路径 → 空（宁可当视频处理也不要拼出坏链接）
    assert ks._atlas_image_urls(_photo(ext_params={"atlas": {"list": ["a.jpg"]}})) == []
    assert ks._atlas_image_urls(
        _photo(ext_params={"atlas": {"cdnList": [{"cdn": "c"}]}})
    ) == []
    # 老字段名 extParams 也认（ext_params 缺失时才会去看它）
    assert ks._atlas_image_urls(
        _photo(
            ext_params=None,
            extParams={"atlas": {"cdnList": [{"cdn": "c"}], "list": ["a.jpg"]}},
        )
    ) == ["https://c/a.jpg"]


# ───────────────────────── 命名 ─────────────────────────


def test_rel_stem_uses_author_id(ks, u):
    """有 userId（数字）→ 稳定作者目录，标题去掉 #话题"""
    assert ks._build_rel_stem(_photo()) == "砂狼白子_2329823245/欧巴欧巴哦"


def test_rel_stem_without_author(ks, u):
    """昵称是兜底串（"快手用户"）→ 按"未知作者"规则 + 作品来源码"""
    stem = ks._build_rel_stem(_photo(userName="快手用户", userId=0))
    code = u.short_source_code("5206161376908999738")
    assert stem == f"未知作者_{code}/欧巴欧巴哦"


def test_rel_stem_empty_caption(ks, u):
    """标题只有 #话题（slugify 后为空）→ 作品_来源码，同一作品稳定可复现"""
    stem = ks._build_rel_stem(_photo(caption="#蔚蓝档案"))
    code = u.short_source_code("5206161376908999738")
    assert stem == f"砂狼白子_2329823245/作品_{code}"


# ───────────────────────── URL 识别 ─────────────────────────


def test_web_short_video_regex(ks):
    """www 站形态（分享出来的网页链接，实际会重写成移动端分享页）"""
    for url, expected in (
        ("https://www.kuaishou.com/short-video/3xhjgcmir24m4nm", "3xhjgcmir24m4nm"),
        ("https://www.kuaishou.com/short-video/5206161376908999738", "5206161376908999738"),
        ("https://kuaishou.com/short-video/abc123", "abc123"),
    ):
        m = ks.WEB_SHORT_VIDEO_RE.search(url)
        assert m and m.group(1) == expected


def test_short_link_regex(ks):
    m = ks.SHORT_LINK_RE.search("看看这个 https://v.kuaishou.com/nUqhAiDI 好看")
    assert m and m.group(0) == "v.kuaishou.com/nUqhAiDI"