MediaCrawler/tests/test_no_user_info.py

# -*- coding: utf-8 -*-
"""
教学版回归测试：确保爬取/存储链路不再持久化可定位真人的用户个人信息。

覆盖：
1. ORM 自省 —— database.models 中无禁用列、creator 档案表已删除、内容/评论表含 creator_hash。
2. 提取层 —— 用 mock API/HTML payload 喂各平台提取器，断言输出 dict 不含禁用字段、
   不含原始 user_id、昵称已脱敏且不等于原文。
3. 仓库 grep 断言 —— store/ 与 media_platform/ 不再把禁用字段作为存储 dict 的 key。
"""
import re
import subprocess
import pathlib

import pytest

ROOT = pathlib.Path(__file__).resolve().parent.parent

# 统一的禁用字段名(键)。昵称字段(nickname/user_nickname/screen_name/name/user_name)允许保留(值需脱敏)。
FORBIDDEN_KEYS = {
    "user_id", "sec_uid", "short_user_id", "user_unique_id", "user_signature",
    "avatar", "user_avatar", "face", "sign", "profile_url", "user_link",
    "url_token", "user_url_token", "ip_location", "ip_address", "gender", "sex",
    "up_id", "fan_id", "up_name", "fan_name", "up_avatar", "fan_avatar",
    "up_sign", "fan_sign", "mid",
}
NICK_KEYS = {"nickname", "user_nickname", "screen_name", "name", "user_name"}
MASK_RE = re.compile(r"^.?\*{1,4}.?$")


# ----------------------------- ORM 自省 -----------------------------

def test_orm_has_no_forbidden_columns():
    import database.models as m
    from sqlalchemy.orm import class_mapper
    tables = [c for c in dir(m) if c[0].isupper()
              and c not in ("Base", "Column", "Integer", "BigInteger", "String", "Text")]
    bad = []
    for t in tables:
        cols = {c.name for c in class_mapper(getattr(m, t)).columns}
        hit = cols & FORBIDDEN_KEYS
        if hit:
            bad.append((t, sorted(hit)))
    assert not bad, f"ORM 仍含禁用列: {bad}"


def test_creator_tables_removed():
    import database.models as m
    removed = {"XhsCreator", "DyCreator", "WeiboCreator", "TiebaCreator",
               "ZhihuCreator", "BilibiliUpInfo", "BilibiliContactInfo"}
    for t in removed:
        assert not hasattr(m, t), f"creator 档案表 {t} 仍存在"


def test_content_tables_have_creator_hash():
    import database.models as m
    from sqlalchemy.orm import class_mapper
    content_tables = ["XhsNote", "XhsNoteComment", "WeiboNote", "WeiboNoteComment",
                      "BilibiliVideo", "BilibiliVideoComment", "BilibiliUpDynamic",
                      "DouyinAweme", "DouyinAwemeComment", "KuaishouVideo",
                      "KuaishouVideoComment", "TiebaNote", "TiebaComment",
                      "ZhihuContent", "ZhihuComment"]
    for t in content_tables:
        cols = {c.name for c in class_mapper(getattr(m, t)).columns}
        assert "creator_hash" in cols, f"{t} 缺少 creator_hash 列"


# ----------------------------- 提取层 mock -----------------------------

def _check_no_forbidden_keys(d: dict, label: str):
    keys = set(d.keys())
    hit = keys & FORBIDDEN_KEYS
    assert not hit, f"[{label}] 输出仍含禁用字段键: {hit}"


def _check_nickname_masked(d: dict, raw: str, label: str):
    nick_keys = set(d.keys()) & NICK_KEYS
    assert nick_keys, f"[{label}] 未保留任何昵称字段(应保留并脱敏)"
    for k in nick_keys:
        val = d[k]
        if val in ("", None):
            continue
        assert val != raw, f"[{label}] {k} 未脱敏，仍为原文: {val}"
        assert MASK_RE.match(val) or "*" in val, f"[{label}] {k} 未脱敏: {val}"


def test_mask_and_hash_tools():
    from tools.user_hash import anonymize_user_id, mask_nickname
    h = anonymize_user_id("12345")
    assert h and h != "12345" and re.fullmatch(r"[0-9a-f]{16}", h)
    assert anonymize_user_id(None) == "" and anonymize_user_id("") == ""
    # 昵称脱敏：首尾留1字、中间星号，且不等于原文
    assert mask_nickname("张三丰") != "张三丰"
    assert "*" in mask_nickname("张三丰")
    assert mask_nickname(None) == ""
    assert mask_nickname("a") == "*"


def test_xhs_note_extraction_masks_user_info():
    import asyncio
    import store.xhs as xs
    note_item = {
        "note_id": "abc",
        "type": "normal",
        "title": "t",
        "desc": "d",
        "time": 1,
        "last_update_time": 0,
        "user": {"user_id": "u123", "nickname": "小红同学", "avatar": "http://x/a.jpg"},
        "ip_location": "上海",
        "interact_info": {"liked_count": "1", "collected_count": "0",
                          "comment_count": "0", "share_count": "0"},
        "image_list": [], "tag_list": [], "xsec_token": "tok",
    }
    captured = {}

    class FakeStore:
        async def store_content(self, content_item):
            captured.update(content_item)

    orig = xs.XhsStoreFactory.create_store
    xs.XhsStoreFactory.create_store = staticmethod(lambda: FakeStore())
    try:
        asyncio.run(xs.update_xhs_note(note_item))
    finally:
        xs.XhsStoreFactory.create_store = orig
    _check_no_forbidden_keys(captured, "xhs_note")
    assert captured.get("creator_hash") != "u123"
    _check_nickname_masked(captured, "小红同学", "xhs_note")


def test_tieba_note_extraction_masks_user_info():
    from media_platform.tieba.help import TieBaExtractor
    api_data = {
        "thread": {"id": 1, "title": "tt", "reply_num": 5},
        "first_floor": {"tid": 1, "author_id": 9, "time": 1700000000, "content": "c"},
        "forum": {"name": "test", "id": 1},
        "page": {"total_page": 1},
        "user_list": [{"id": 9, "name_show": "贴吧老哥", "name": "lg", "portrait": "p", "ip_address": "北京"}],
    }
    note = TieBaExtractor().extract_note_detail_from_api(api_data)
    d = note.model_dump()
    _check_no_forbidden_keys(d, "tieba_note")
    assert d.get("creator_hash")  # user_link 已转哈希
    _check_nickname_masked(d, "贴吧老哥", "tieba_note")


def test_tieba_comment_extraction_masks_user_info():
    from media_platform.tieba.help import TieBaExtractor
    from model.m_baidu_tieba import TiebaNote
    api_data = {
        "forum": {"id": 1, "name": "test"},
        "post_list": [{"id": 7, "author_id": 9, "time": 1700000000, "content": "c", "sub_post_number": 0}],
        "user_list": [{"id": 9, "name_show": "评论员", "name": "py", "portrait": "p", "ip_address": "上海"}],
    }
    note_detail = TiebaNote(note_id="1", title="t", note_url="u", tieba_name="test", tieba_link="l")
    comments = TieBaExtractor().extract_tieba_note_parent_comments_from_api(api_data, note_detail)
    assert comments
    d = comments[0].model_dump()
    _check_no_forbidden_keys(d, "tieba_comment")
    _check_nickname_masked(d, "评论员", "tieba_comment")


def test_zhihu_comment_extraction_masks_user_info():
    from media_platform.zhihu.help import ZhihuExtractor
    from model.m_zhihu import ZhihuContent
    comments_raw = [{
        "type": "comment", "id": 1, "content": "c", "created_time": 1700000000,
        "like_count": 1, "dislike_count": 0, "child_comment_count": 0,
        "author": {"id": "z9", "name": "知乎答主", "url_token": "tok", "avatar_url": "http://x/a.jpg"},
        "comment_tag": [{"type": "ip_info", "text": "广东"}],
    }]
    page_content = ZhihuContent(content_id="c1", content_type="answer")
    comments = ZhihuExtractor().extract_comments(page_content, comments_raw)
    assert comments
    d = comments[0].model_dump() if hasattr(comments[0], "model_dump") else vars(comments[0])
    _check_no_forbidden_keys(d, "zhihu_comment")
    assert "creator_hash" in d and d["creator_hash"]
    _check_nickname_masked(d, "知乎答主", "zhihu_comment")


def test_bilibili_video_dict_masks_user_info():
    # 直接测 store/bilibili/__init__.py 的拍平逻辑(不触发网络)
    import asyncio
    from store.bilibili import update_bilibili_video
    video_item = {
        "View": {
            "aid": 100, "title": "t", "desc": "d", "pubdate": 1,
            "owner": {"mid": 777, "name": "UP主大人", "face": "http://x/a.jpg"},
            "stat": {"like": 1, "view": 2},
            "pic": "http://x/cover.jpg",
        }
    }
    # 拦截真实存储：替换工厂返回一个捕获 dict 的假 store
    captured = {}

    class FakeStore:
        async def store_content(self, content_item):
            captured.update(content_item)

    import store.bilibili as bs
    orig = bs.BiliStoreFactory.create_store
    bs.BiliStoreFactory.create_store = staticmethod(lambda: FakeStore())
    try:
        asyncio.get_event_loop().run_until_complete(update_bilibili_video(video_item)) \
            if False else asyncio.run(update_bilibili_video(video_item))
    finally:
        bs.BiliStoreFactory.create_store = orig
    _check_no_forbidden_keys(captured, "bili_video")
    assert captured.get("creator_hash") != 777
    _check_nickname_masked(captured, "UP主大人", "bili_video")


# ----------------------------- 仓库 grep 断言 -----------------------------

def test_store_no_forbidden_dict_keys():
    # store/ 下不得把禁用字段作为存储 dict 的 key("field": value 形式)
    out = subprocess.run(
        ["grep", "-rnE", '"(' + "|".join(FORBIDDEN_KEYS) + r')"\s*:', str(ROOT / "store")],
        capture_output=True, text=True,
    )
    # 允许的例外：Mongo store_creator 里的 query={"user_id": ...} 已全部改为 pass，应为空
    assert out.stdout.strip() == "", f"store/ 仍写入禁用字段键:\n{out.stdout}"


def test_store_no_creator_orm_imports():
    # 已删除的 creator ORM 表(XhsCreator/DyCreator/...)不得再从 database.models 导入。
    # 注意:model/m_*.py 里的同名 pydantic 类是内存类型，允许保留。
    out = subprocess.run(
        ["grep", "-rnE",
         r"from database\.models import.*(XhsCreator|DyCreator|WeiboCreator|TiebaCreator|ZhihuCreator|BilibiliUpInfo|BilibiliContactInfo)",
         str(ROOT / "store")],
        capture_output=True, text=True,
    )
    assert out.stdout.strip() == "", f"store/ 仍 import 已删除的 creator ORM 表:\n{out.stdout}"


if __name__ == "__main__":
    pytest.main([__file__, "-v"])