query-risk-user-features/tests/test_offline_regression.py

from __future__ import annotations

import sys
import unittest
from pathlib import Path

import pandas as pd


PROJECT_DIR = Path(__file__).resolve().parents[1]
WORKSPACE_ROOT = PROJECT_DIR.parent
for path in (PROJECT_DIR, WORKSPACE_ROOT):
    if str(path) not in sys.path:
        sys.path.insert(0, str(path))

from scripts.build_report import group_summary
from src.social_logic import aggregate_groups, aggregate_relationships
from src.user_input import load_users


FIXTURE_DIR = PROJECT_DIR / "fixtures" / "v2_220"
SAMPLE_PATH = PROJECT_DIR / "samples" / "v2_220_20260810.csv"


class OfflineRegressionTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.users = load_users(SAMPLE_PATH, lookback_days=180)
        cls.behavior = pd.read_csv(FIXTURE_DIR / "user_detail.csv")
        cls.relationships = pd.read_csv(FIXTURE_DIR / "relationships.csv")
        cls.groups = pd.read_csv(FIXTURE_DIR / "social_group_features.csv")
        cls.expected_social = pd.read_csv(FIXTURE_DIR / "social_user_features.csv")

    def test_sample_contract(self) -> None:
        self.assertEqual(len(self.users), 220)
        self.assertEqual(self.users["mid"].nunique(), 220)
        self.assertTrue((self.users["window_end_dt"] < self.users["anchor_dt"]).all())
        counts = self.users.groupby(["user_type", "sub_category"]).size().to_dict()
        self.assertEqual(counts[("明确举报人", "明确举报人")], 7)
        self.assertEqual(counts[("审核人", "审核人")], 13)
        self.assertEqual(counts[("正常用户", "正常用户")], 100)
        suspected = sum(value for (group, _), value in counts.items() if group == "疑似举报人")
        self.assertEqual(suspected, 100)

    def test_long_retention_play_and_share_sources(self) -> None:
        builder = (PROJECT_DIR / "src" / "feature_sql.py").read_text(encoding="utf-8")
        dataworks = (
            PROJECT_DIR / "sql" / "dataworks_single_user_180d_all_in_one.sql"
        ).read_text(encoding="utf-8")
        for sql in (builder, dataworks):
            self.assertIn("loghubods.ods_video_play_log_day", sql)
            self.assertIn("s.topic = 'share'", sql)
            self.assertNotIn("LEFT JOIN loghubods.video_play_log p", sql)

    def test_social_aggregation_matches_saved_result(self) -> None:
        relationships = aggregate_relationships(self.users, self.relationships)
        groups = aggregate_groups(self.users, self.groups)
        actual = relationships.merge(groups, on="用户id", validate="one_to_one")
        expected = self.expected_social
        self.assertEqual(len(actual), 220)
        self.assertEqual(actual["用户id"].nunique(), 220)
        columns = [
            "社交关系点击次数",
            "风险来源点击次数",
            "同一卡片最大连续点击次数",
            "有效来源群数",
            "来源群访问UV合计",
            "来源群有效播放UV合计",
        ]
        merged = actual[["用户id", *columns]].merge(
            expected[["用户id", *columns]],
            on="用户id",
            suffixes=("_actual", "_expected"),
            validate="one_to_one",
        )
        for column in columns:
            pd.testing.assert_series_equal(
                pd.to_numeric(merged[f"{column}_actual"], errors="coerce").fillna(0),
                pd.to_numeric(merged[f"{column}_expected"], errors="coerce").fillna(0),
                check_names=False,
            )

    def test_population_group_facts(self) -> None:
        detail = self.behavior.merge(
            self.expected_social,
            on="用户id",
            how="left",
            validate="one_to_one",
        )
        summary = group_summary(detail).set_index("用户分类")
        expected = {
            "明确举报人": (7, 8, 149, 0),
            "审核人": (13, 2, 54, 19),
            "疑似举报人": (100, 316, 5560, 2543),
            "正常用户": (100, 310, 6404, 2760),
        }
        for group, values in expected.items():
            self.assertEqual(int(summary.at[group, "总人数"]), values[0])
            self.assertEqual(int(summary.at[group, "访问来源群数合计"]), values[1])
            self.assertEqual(
                int(summary.at[group, "来源群访问用户数合计（按群UV求和）"]),
                values[2],
            )
            self.assertEqual(
                int(summary.at[group, "来源群有效播放用户数合计（按群UV求和）"]),
                values[3],
            )


if __name__ == "__main__":
    unittest.main()