MediaCrawler/api/monitor/adapters.py

# -*- coding: utf-8 -*-
# Copyright (c) 2025 [email protected]
#
# This file is part of MediaCrawler project.
# Repository: https://github.com/NanmiCoder/MediaCrawler/blob/main/api/monitor/adapters.py
# GitHub: https://github.com/NanmiCoder
# Licensed under NON-COMMERCIAL LEARNING LICENSE 1.1
#
# 声明：本代码仅供学习和研究目的使用。使用者应遵守以下原则：
# 1. 不得用于任何商业用途。
# 2. 使用时应遵守对应平台的使用条款和robots.txt规则。
# 3. 不得进行大规模爬取或对平台造成运营干扰。
# 4. 应合理控制请求频率，避免给目标平台带来不必要的负担。
# 5. 不得用于任何非法或不当的用途。
#
# 详细许可条款请参阅项目根目录下的LICENSE文件。
# 使用本代码即表示您同意遵守上述原则和LICENSE中的所有条款。

"""平台适配：两个平台之间**不一样**的那些管子。

监控层的大部分是平台中立的 —— 调度、入库、差分、报表、封面缓存、通知发送都与平台无关。
真正随平台变化的只有四样东西：

1. 爬虫把产物**落在哪个目录**（这里有个坑，见 ``artifact_dir``）
2. jsonl 里**字段叫什么**（抖音的作品没有 ``note_id``，叫 ``aweme_id``）
3. **目标链接**长什么样（怎么拼、怎么从链接里抠出 id）
4. 通知里的作品链接怎么拼

集中在这里，是为了让「加一个平台」变成在一处补一份数据，而不是去五个文件里找硬编码。

**为什么不放进 platforms.py**：那个模块被 ``describe_all()`` 整个序列化进
``GET /api/config/platforms`` 交给前端（连 ``**capability`` 一起），把正则、目录名、字段别名
塞进去会让爬虫的内部细节漏进 API 载荷，也会让「改适配」有动到接口形状的风险。
分工与既有的 schedule.py（算术）↔ scheduler.py（循环）一致。
"""

import re
from dataclasses import dataclass
from typing import Any, Dict, Mapping, Pattern, Tuple

from .platforms import PLATFORM_XHS

PLATFORM_DY = "dy"


def _first_cover(record: Dict[str, Any], fields: Tuple[str, ...]) -> str:
    """封面地址：取第一个非空字段，再取逗号分隔的第一段。

    一条规则同时适配两边，所以不需要 per-platform 的函数：
    小红书的 ``image_list`` 是 ``"url1,url2,..."``（要切第一段），
    抖音的 ``cover_url`` 本身就是单个地址（切了等于没切）。
    """
    for name in fields:
        raw = record.get(name)
        if raw:
            return str(raw).split(",")[0].strip()
    return ""


@dataclass(frozen=True)
class PlatformAdapter:
    """一个平台的全部「管子」。

    字段别名的方向是**规范名 -> 该平台 jsonl 里的键**，读作「我们的列 ← 他们的键」。
    """

    # 爬虫落盘用的目录名。**不等于平台 id**：抖音的平台 id 是 ``dy`` 而目录是 ``douyin``。
    # 这不是笔误，是上游 store 里写死的（store/douyin/_store_impl.py:47）。改错这里的
    # 后果是 ingest 一个文件都找不到 —— 它不会报错，只会落进「没抓到数据」分支，
    # 然后被误报成「疑似登录失效」。
    artifact_dir: str

    web_base: str
    creator_path: str
    note_path: str

    # 从链接里抠 id。是元组而不是单个正则，因为同一个平台可能有多种链接形态
    # （抖音的作品链接还带 ?modal_id= 那种），按顺序试，第一个匹配的胜出。
    # 每个正则必须恰好有一个捕获组。
    creator_url_res: Tuple[Pattern, ...]
    note_url_res: Tuple[Pattern, ...]
    # 也允许直接粘贴裸 id —— 但两边的 id 形状不同，所以分开。
    creator_bare_re: Pattern
    note_bare_re: Pattern
    # 短链（v.douyin.com 这种）无法在不发请求的情况下还原出 id，解析时单独报错，
    # 好过存一个聚不出目标的值进去。
    short_link_hosts: Tuple[str, ...]

    note_fields: Mapping[str, str]
    comment_fields: Mapping[str, str]
    cover_fields: Tuple[str, ...]

    def note_url(self, note_id: str) -> str:
        """作品的可点击链接。拼法与监控目标的链接是同一个形状 —— 通知里给的就是
        人能直接点开看的那一个。"""
        return f"{self.web_base}{self.note_path}/{note_id}"

    def note_field(self, record: Dict[str, Any], name: str) -> Any:
        """按规范名读作品记录里的原始值（没有就是 None）。"""
        return record.get(self.note_fields.get(name, name))

    def comment_field(self, record: Dict[str, Any], name: str) -> Any:
        return record.get(self.comment_fields.get(name, name))

    def cover(self, record: Dict[str, Any]) -> str:
        return _first_cover(record, self.cover_fields)

    def parent_comment_id(self, record: Dict[str, Any]) -> str:
        """父评论 id，顶层评论一律归一成空串。

        抖音顶层评论的 ``reply_id`` 是字符串 ``"0"``，小红书是 ``""`` —— 把 "0" 原样
        存进去，前端就会多出一堆指向不存在的父评论的边。
        """
        raw = self.comment_field(record, "parent_comment_id")
        if raw is None:
            return ""
        raw = str(raw).strip()
        return "" if raw in ("", "0") else raw


# 小红书 id 是 24 位 hex，允许稍宽一点，让格式变化退化成「仍然接受」而不是「拒绝」。
_XHS_BARE_RE = re.compile(r"^[A-Za-z0-9_-]{8,64}$")

XHS = PlatformAdapter(
    artifact_dir="xhs",
    web_base="https://www.xiaohongshu.com",
    creator_path="/user/profile",
    note_path="/explore",
    creator_url_res=(re.compile(r"xiaohongshu\.com/user/profile/([A-Za-z0-9_-]+)"),),
    note_url_res=(
        re.compile(r"xiaohongshu\.com/(?:explore|discovery/item)/([A-Za-z0-9_-]+)"),
    ),
    creator_bare_re=_XHS_BARE_RE,
    note_bare_re=_XHS_BARE_RE,
    short_link_hosts=(),
    note_fields={
        "note_id": "note_id",
        "title": "title",
        "note_url": "note_url",
        "creator_hash": "creator_hash",
        "creator_name": "nickname",
        "source_kind": "type",
        "published_at": "time",
    },
    comment_fields={
        "comment_id": "comment_id",
        "note_id": "note_id",
        "content": "content",
        "creator_hash": "creator_hash",
        "creator_name": "nickname",
        "create_time": "create_time",
        "like_count": "like_count",
        "sub_comment_count": "sub_comment_count",
        "parent_comment_id": "parent_comment_id",
    },
    cover_fields=("image_list",),
)

# 抖音的 id 形状与小红书完全不同（见 media_platform/douyin/help.py:101-164）：
#   作品  aweme_id    纯数字，如 7525082444551310602
#   博主  sec_user_id 形如 MS4wLjABAAAA...，含 - 和 _，**变长**（实测样本 55 字符，更长的也常见），
#        而小红书那条裸 id 规则封顶 64 —— 所以两条规则必须分开，否则长一点的 sec_uid
#        会被拒，表现为「粘贴了一个完全正确的链接却说无法识别」。
# 另外抖音**不需要 xsec_token**，裸链接就能用，比小红书简单。
DY = PlatformAdapter(
    artifact_dir="douyin",
    web_base="https://www.douyin.com",
    creator_path="/user",
    note_path="/video",
    creator_url_res=(re.compile(r"douyin\.com/user/([A-Za-z0-9_-]+)"),),
    note_url_res=(
        re.compile(r"douyin\.com/video/(\d+)"),
        # 带 modal_id 的链接：在别人主页或搜索结果里点开视频就是这个形态。
        re.compile(r"[?&]modal_id=(\d+)"),
    ),
    # 用长度而不是前缀来区分两者：sec_uid 是 20 字符以上的变长串，作品 id 是 19 位数字。
    # 用前缀（MS4wLjABAAAA）更精确，但上游的 parse_creator_info_from_url 对裸 id 一律
    # 照单全收，万一有别的前缀就会被我这里挡掉 —— 门槛设在长度上，两边都放得进，
    # 又不会把 19 位的作品号误当成博主。
    creator_bare_re=re.compile(r"^[A-Za-z0-9_-]{20,128}$"),
    note_bare_re=re.compile(r"^\d{8,25}$"),
    short_link_hosts=("v.douyin.com",),
    note_fields={
        "note_id": "aweme_id",
        "title": "title",
        "note_url": "aweme_url",
        "creator_hash": "creator_hash",
        "creator_name": "nickname",
        "source_kind": "aweme_type",
        "published_at": "create_time",
    },
    comment_fields={
        "comment_id": "comment_id",
        "note_id": "aweme_id",
        "content": "content",
        "creator_hash": "creator_hash",
        "creator_name": "nickname",
        "create_time": "create_time",
        "like_count": "like_count",
        "sub_comment_count": "sub_comment_count",
        "parent_comment_id": "parent_comment_id",
    },
    cover_fields=("cover_url",),
)

ADAPTERS: Dict[str, PlatformAdapter] = {
    PLATFORM_XHS: XHS,
    PLATFORM_DY: DY,
}


class UnknownPlatformError(ValueError):
    """平台还没有适配器。"""


def adapter(platform: str) -> PlatformAdapter:
    try:
        return ADAPTERS[platform]
    except KeyError as exc:
        raise UnknownPlatformError(f"平台 {platform} 还没有适配器") from exc


def has_adapter(platform: str) -> bool:
    return platform in ADAPTERS


def artifact_dir(platform: str) -> str:
    """该平台的爬虫会把 jsonl 落在哪个子目录下。

    ``runner`` 用它判断产物是否真的出现过，``ingest`` 用它定位文件 —— 两处必须用
    同一个值，否则会出现「文件在，但两边找的目录不是同一个」这种最难查的错。
    """
    return adapter(platform).artifact_dir