query-risk-user-features/scripts/build_report.py

#!/usr/bin/env python3
"""Build deterministic user and population reports from behavior and social CSVs."""

from __future__ import annotations

import argparse
from pathlib import Path

import pandas as pd


GROUP_ORDER = ["明确举报人", "审核人", "疑似举报人", "正常用户"]
ID_COLUMNS = {"用户分类", "子分类", "统计周期", "用户id"}
TEXT_COLUMNS = {
    "操作系统", "机型", "地域", "来源其他场景值",
    "来源上游用户shareid", "来源群id",
}
GROUP_FEATURES = [
    "有效来源群数",
    "来源群访问UV合计",
    "来源群有效播放UV合计",
    "来源群有效播放用户率",
]


def ordered_groups(detail: pd.DataFrame) -> list[str]:
    present = detail["用户分类"].drop_duplicates().tolist()
    return [group for group in GROUP_ORDER if group in present] + [
        group for group in present if group not in GROUP_ORDER
    ]


def numeric_summary(detail: pd.DataFrame) -> pd.DataFrame:
    numeric = [column for column in detail.columns if column not in ID_COLUMNS | TEXT_COLUMNS]
    rows = []
    for group_name in ordered_groups(detail):
        group = detail[detail["用户分类"] == group_name]
        for feature in numeric:
            values = pd.to_numeric(group[feature], errors="coerce").fillna(0)
            rows.append(
                {
                    "用户分类": group_name,
                    "用户数": group["用户id"].nunique(),
                    "特征": feature,
                    "取值范围": f"{values.min():g}~{values.max():g}",
                    "均值": values.mean(),
                    "中位数": values.median(),
                    "P90": values.quantile(0.9),
                    "覆盖用户数": int(values.gt(0).sum()),
                    "覆盖用户比例": values.gt(0).mean(),
                }
            )
    return pd.DataFrame(rows)


def group_summary(detail: pd.DataFrame) -> pd.DataFrame:
    missing = [column for column in GROUP_FEATURES if column not in detail.columns]
    if missing:
        return pd.DataFrame(columns=["用户分类", "总人数", *GROUP_FEATURES])
    rows = []
    for group_name in ordered_groups(detail):
        group = detail[detail["用户分类"] == group_name].copy()
        for column in GROUP_FEATURES:
            group[column] = pd.to_numeric(group[column], errors="coerce").fillna(0)
        has_group = group["有效来源群数"] > 0
        visit_uv = group["来源群访问UV合计"].sum()
        play_uv = group["来源群有效播放UV合计"].sum()
        rows.append(
            {
                "用户分类": group_name,
                "总人数": group["用户id"].nunique(),
                "有来源群用户数": int(has_group.sum()),
                "有来源群用户占比": has_group.mean(),
                "访问来源群数合计": int(group["有效来源群数"].sum()),
                "来源群访问用户数合计（按群UV求和）": int(visit_uv),
                "来源群有效播放用户数合计（按群UV求和）": int(play_uv),
                "来源群有效播放用户率": play_uv / visit_uv if visit_uv else 0,
                "有来源群但有效播放UV为0用户数": int(
                    (has_group & group["来源群有效播放UV合计"].eq(0)).sum()
                ),
                "有来源群但有效播放UV为0用户占比": (
                    (has_group & group["来源群有效播放UV合计"].eq(0)).sum()
                    / has_group.sum()
                    if has_group.sum()
                    else 0
                ),
            }
        )
    return pd.DataFrame(rows)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--behavior-detail", type=Path, required=True)
    parser.add_argument("--social-dir", type=Path)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()

    behavior = pd.read_csv(args.behavior_detail)
    detail = behavior.copy()
    social = group_detail = relationships = None
    if args.social_dir:
        social = pd.read_csv(args.social_dir / "user_social_features.csv")
        group_detail = pd.read_csv(args.social_dir / "group_detail.csv")
        relationships = pd.read_csv(args.social_dir / "relationship_detail.csv")
        detail = behavior.merge(social, on="用户id", how="left", validate="one_to_one")

    summary = numeric_summary(detail.copy())
    source_group_summary = group_summary(detail.copy())
    args.output.parent.mkdir(parents=True, exist_ok=True)
    with pd.ExcelWriter(args.output, engine="openpyxl") as writer:
        detail.to_excel(writer, sheet_name="用户完整明细", index=False)
        summary.to_excel(writer, sheet_name="人群特征聚合", index=False)
        source_group_summary.to_excel(writer, sheet_name="来源群聚合", index=False)
        if group_detail is not None:
            group_detail.to_excel(writer, sheet_name="来源群逐群明细", index=False)
        if relationships is not None:
            relationships.to_excel(writer, sheet_name="分享关系明细", index=False)
    print(f"users={len(detail)} unique_mids={detail['用户id'].nunique()} output={args.output}")


if __name__ == "__main__":
    main()