wiki_retrieve/data_mock/mock_dashboards.py

from __future__ import annotations

import argparse
import hashlib
import random
from datetime import datetime, timedelta
from pathlib import Path

import pandas as pd


ROOT = Path(__file__).resolve().parents[1]
OUTPUT_PATH = ROOT / "data" / "dashboard.parquet"
RANDOM_SEED = 20260506


BRANCH_ID_MAP = {
    "总行": "00100",
    "北京分行": "00110",
    "广东分行": "00120",
}


BRANCH_DM_MAP = {
    "总行": "zh",
    "北京分行": "bjfh",
    "广东分行": "gdfh",
}


DEFAULT_BRANCH_NAMES = ["总行", "北京分行", "广东分行"]


DOMAIN_CONFIGS = [
    {
        "domain": "零售",
        "level1": "零售金融部",
        "level2": "经营分析室",
        "level3": "客户经营团队",
        "topics": [
            "零售客户经营总览",
            "高价值客户增长分析",
            "私人银行客户资产监测",
            "私钻客户AUM提升分析",
            "财富客户资产配置看板",
            "养老金客户经营跟踪",
            "代发客群转化跟踪",
            "零售存款日均分析",
            "储蓄存款客群结构分析",
            "零售贷款投放监控",
            "个人住房贷款质量看板",
            "消费贷获客转化分析",
            "汽车分期贷款经营分析",
            "理财产品销售分析",
            "基金产品持仓透视",
            "保险产品销售监测",
            "客户资产提升作战图",
            "网点零售业绩排名",
            "MGM获客转化分析",
        ],
        "folder": "零售数据门户",
    },
    {
        "domain": "对公",
        "level1": "公司金融部",
        "level2": "公司客户经营室",
        "level3": "对公经营分析团队",
        "topics": [
            "对公客户经营总览",
            "战略客户贡献分析",
            "机构客户综合贡献分析",
            "上市公司客群经营看板",
            "对公存款增长看板",
            "单位活期存款结构分析",
            "保证金存款监测",
            "代发业务拓展监测",
            "对公信贷投放监控",
            "项目融资投放分析",
            "小微企业贷款分析",
            "供应链金融业务看板",
            "信用证业务分析",
            "保函业务风险监测",
            "票据贴现经营分析",
            "现金管理客户活跃分析",
            "交易银行客户活跃分析",
            "公司客户流失预警",
        ],
        "folder": "公司金融数据门户",
    },
    {
        "domain": "信用卡",
        "level1": "信用卡中心",
        "level2": "经营管理部",
        "level3": "数据分析团队",
        "topics": [
            "信用卡新增客户分析",
            "信用卡交易额监测",
            "信用卡分期业务看板",
            "信用卡账单分期转化分析",
            "信用卡逾期风险预警",
            "信用卡活跃客户经营",
            "信用卡渠道获客分析",
            "信用卡额度使用分析",
            "信用卡客群画像看板",
            "信用卡权益活动效果分析",
            "信用卡商户交易监测",
            "信用卡睡眠客户唤醒看板",
        ],
        "folder": "信用卡经营分析",
    },
    {
        "domain": "风险",
        "level1": "风险管理部",
        "level2": "组合风险管理室",
        "level3": "模型监测团队",
        "topics": [
            "信贷资产质量监测",
            "不良贷款迁徙分析",
            "逾期客户风险预警",
            "关注类贷款压降监控",
            "拨备覆盖率经营看板",
            "重点行业风险排查",
            "授信集中度监控",
            "贷后检查进度跟踪",
            "资产分类变动分析",
            "抵质押品价值重估分析",
            "风险预警信号处置看板",
            "大额风险暴露监测",
        ],
        "folder": "风险管理驾驶舱",
    },
    {
        "domain": "运营",
        "level1": "运营管理部",
        "level2": "运营监控室",
        "level3": "流程管理团队",
        "topics": [
            "网点运营效率看板",
            "柜面业务量监测",
            "远程银行服务分析",
            "账户开立质量监测",
            "企业账户年检进度监控",
            "支付结算业务分析",
            "反洗钱可疑交易跟踪",
            "客户投诉处理看板",
            "运营风险事件监控",
            "集中作业处理时效分析",
            "现金库存与调拨监测",
            "电子回单服务分析",
        ],
        "folder": "运营管理专区",
    },
    {
        "domain": "财务",
        "level1": "计划财务部",
        "level2": "管理会计室",
        "level3": "财务分析团队",
        "topics": [
            "分行利润贡献分析",
            "FTP收支测算看板",
            "费用预算执行监控",
            "中间业务收入分析",
            "净利息收入贡献分析",
            "资本占用收益分析",
            "经营计划完成率看板",
            "资产负债结构分析",
            "税务成本监测看板",
            "经济利润EVA分析",
            "分产品收益率分析",
            "管理会计分摊结果看板",
        ],
        "folder": "财务管理驾驶舱",
    },
    {
        "domain": "渠道",
        "level1": "网络经营服务部",
        "level2": "渠道管理室",
        "level3": "数字渠道团队",
        "topics": [
            "手机银行活跃分析",
            "网银交易监测",
            "网点客流热力分析",
            "自助设备运行看板",
            "远程渠道转化分析",
            "企业网银客户经营",
            "开放银行接口监测",
            "渠道协同营销看板",
            "数字人民币交易分析",
            "小程序渠道转化分析",
            "线上预约到店分析",
            "渠道客户体验监测",
        ],
        "folder": "渠道经营分析",
    },
    {
        "domain": "普惠",
        "level1": "普惠金融部",
        "level2": "普惠经营管理室",
        "level3": "小微客户团队",
        "topics": [
            "普惠贷款投放监控",
            "小微客户增长看板",
            "普惠首贷户拓展分析",
            "涉农贷款经营监测",
            "科技型企业贷款分析",
            "普惠风险补偿跟踪",
            "普惠延期还本付息监测",
            "科创小微客户经营分析",
            "个体工商户贷款看板",
            "普惠贷款利率定价分析",
        ],
        "folder": "普惠金融专区",
    },
    {
        "domain": "金融市场",
        "level1": "金融市场部",
        "level2": "投资交易管理室",
        "level3": "市场风险与经营分析团队",
        "topics": [
            "债券投资组合分析",
            "同业负债成本监测",
            "资金头寸预测看板",
            "外汇交易损益分析",
            "衍生品估值监测",
            "票据转贴现业务分析",
            "理财投资资产穿透看板",
            "金融市场限额占用监控",
            "市场价格波动预警",
        ],
        "folder": "金融市场经营专区",
    },
    {
        "domain": "国际业务",
        "level1": "国际业务部",
        "level2": "跨境金融管理室",
        "level3": "国际结算分析团队",
        "topics": [
            "跨境结算业务分析",
            "贸易融资投放监控",
            "进口信用证业务看板",
            "出口托收业务跟踪",
            "外汇存款结构分析",
            "结售汇客户贡献分析",
            "跨境人民币业务监测",
            "国际业务风险预警",
        ],
        "folder": "国际业务分析专区",
    },
    {
        "domain": "合规",
        "level1": "法律合规部",
        "level2": "反洗钱与制裁合规管理团队",
        "level3": "合规数据应用室",
        "topics": [
            "受益所有人信息缺失监测",
            "客户尽职调查进度看板",
            "反洗钱名单命中分析",
            "可疑交易报告质效分析",
            "员工异常行为排查",
            "监管报送质量监控",
            "合规检查问题整改跟踪",
            "制裁筛查处理时效分析",
        ],
        "folder": "合规管理专区",
    },
    {
        "domain": "人力",
        "level1": "人力资源部",
        "level2": "绩效薪酬管理室",
        "level3": "人力数据分析团队",
        "topics": [
            "机构人均产能分析",
            "客户经理绩效看板",
            "支行奖金分配测算",
            "岗位编制与人员缺口分析",
            "员工培训完成率监测",
            "人才盘点与梯队建设看板",
            "一线人员工作量分析",
        ],
        "folder": "人力资源分析专区",
    },
]


VERSION_SUFFIXES = [
    "",
    " V1.0",
    " V2.0",
    " V3.0",
    " V4.0",
    "（机构版）",
    "（管理端）",
    "（支行版）",
    "（客户经理版）",
    "（测试版）",
    "月报",
    "日报",
    "周报",
    "快报",
    "大屏",
]


def make_id(prefix: str, ordinal: int) -> str:
    raw = f"{prefix}-{ordinal}-{RANDOM_SEED}".encode("utf-8")
    return hashlib.md5(raw).hexdigest()[:24]


def parse_branch_names(raw_branch_names: list[str] | None) -> list[str]:
    if not raw_branch_names:
        return DEFAULT_BRANCH_NAMES

    branch_names: list[str] = []
    for raw_value in raw_branch_names:
        for branch_name in raw_value.split(","):
            branch_name = branch_name.strip()
            if branch_name and branch_name not in branch_names:
                branch_names.append(branch_name)

    unknown_branch_names = [name for name in branch_names if name not in BRANCH_DM_MAP]
    if unknown_branch_names:
        supported = "、".join(BRANCH_DM_MAP)
        unknown = "、".join(unknown_branch_names)
        raise ValueError(f"unsupported branch name: {unknown}. Supported branch names: {supported}")
    return branch_names


def build_branch_pool(branch_names: list[str]) -> list[tuple[str, str, str]]:
    return [(BRANCH_ID_MAP[name], name, BRANCH_DM_MAP[name]) for name in branch_names]


def choose_branch(
    rng: random.Random,
    domain: str,
    branch_pool: list[tuple[str, str, str]],
) -> tuple[str, str, str]:
    total_branch = next((branch for branch in branch_pool if branch[1] == "总行"), None)
    if total_branch and domain in {"风险", "财务", "信用卡"} and rng.random() < 0.36:
        return total_branch
    return rng.choice(branch_pool)


def make_timestamp(rng: random.Random, start: datetime, end: datetime) -> str:
    seconds = int((end - start).total_seconds())
    value = start + timedelta(seconds=rng.randint(0, seconds))
    microsecond = rng.choice([0, 152000, 281000, 371000, 495000, 659000, 749000, 911000])
    return value.replace(microsecond=microsecond).strftime("%Y-%m-%d %H:%M:%S.%f")


def build_folder_route(rng: random.Random, branch_name: str, config: dict[str, object], topic: str) -> str:
    roots = [
        f"根目录/{config['folder']}/{config['domain']}/{topic}",
        f"根目录/业务网仪表板/{config['domain']}条线/{branch_name}/{topic}",
        f"根目录/总行/经营管理驾驶舱/{config['folder']}/{topic}",
        f"根目录/用户开发报表区/{branch_name}/{config['folder']}/{topic}",
        f"根目录/应用市场下载目录/分析模板/{config['domain']}经营/{topic}",
    ]
    return rng.choice(roots)


def make_dashboard_name(
    rng: random.Random,
    ordinal: int,
    branch_name: str,
    topic: str,
    used_names: set[str],
) -> str:
    suffix = rng.choice(VERSION_SUFFIXES)
    branch_prefix = branch_name if rng.random() < 0.34 and branch_name != "总行" else ""
    base_name = f"{branch_prefix}{topic}{suffix}"
    dash_name = base_name
    if dash_name in used_names:
        dash_name = f"{base_name}（第{ordinal:03d}期）"

    retry_index = 2
    while dash_name in used_names:
        dash_name = f"{base_name}（第{ordinal:03d}-{retry_index}期）"
        retry_index += 1

    used_names.add(dash_name)
    return dash_name


def build_record(
    rng: random.Random,
    ordinal: int,
    config: dict[str, object],
    topic: str,
    branch_pool: list[tuple[str, str, str]],
    used_names: set[str],
) -> dict[str, object]:
    bbk_id, bbk_name, dm_nm = choose_branch(rng, str(config["domain"]), branch_pool)
    dash_name = make_dashboard_name(rng, ordinal, bbk_name, topic, used_names)

    level1 = str(config["level1"])
    level2 = str(config["level2"])
    level3 = str(config["level3"])
    level4_candidates = ["数据应用室", "经营推动组", "业务支持组", "指标管理组", ""]
    level4 = rng.choice(level4_candidates)
    dept_fname_parts = ["XX银行", bbk_name, level1, level2, level3]
    if level4:
        dept_fname_parts.append(level4)

    chart_cnt = rng.randint(20, 50)
    author_cnt = rng.randint(8, 360)
    first_visit = make_timestamp(rng, datetime(2023, 1, 1), datetime(2026, 2, 28))
    recent_start = datetime.strptime(first_visit, "%Y-%m-%d %H:%M:%S.%f") + timedelta(days=1)
    recent_end = datetime(2026, 5, 5, 23, 59, 59)
    recent_visit = make_timestamp(rng, recent_start, recent_end) if recent_start < recent_end else first_visit

    efficiency = rng.choices(["高效", "中效", "低效"], weights=[0.34, 0.46, 0.20], k=1)[0]
    visit_ranges = {
        "高效": (360, 6800),
        "中效": (80, 1200),
        "低效": (0, 180),
    }
    visit_low, visit_high = visit_ranges[efficiency]

    return {
        "dash_id": make_id("dash", ordinal),
        "dash_dsply_name": dash_name,
        "folder_all_route": build_folder_route(rng, bbk_name, config, topic),
        "build_main": rng.choices(["总行建设", "分行自建"], weights=[0.48, 0.52], k=1)[0],
        "dept_cls": rng.choices(["业务部门", "信息技术部", "风险合规部门", "管理部门"], weights=[0.66, 0.18, 0.09, 0.07], k=1)[0],
        "dash_rat": efficiency,
        "bbk_name": bbk_name,
        "bbk_id": bbk_id,
        "dept_fname": "/".join(dept_fname_parts),
        "level1_dept_name": level1,
        "level2_dept_name": level2,
        "level3_dept_name": level3,
        "level4_dept_name": level4,
        "chart_cnt": chart_cnt,
        "be_author_cnt": author_cnt,
        "fir_be_visit_tm": first_visit,
        "recnt_be_visit_tm": recent_visit,
        "m3_vis_cnt": rng.randint(visit_low, visit_high),
        "mon_add_idf": rng.choices(["是", "否"], weights=[0.08, 0.92], k=1)[0],
        "dm_nm": dm_nm,
    }


def make_dashboards(
    target_count: int = 200,
    start_ordinal: int = 1,
    branch_pool: list[tuple[str, str, str]] | None = None,
    used_names: set[str] | None = None,
) -> pd.DataFrame:
    rng = random.Random(RANDOM_SEED + start_ordinal - 1)
    branch_pool = branch_pool or build_branch_pool(DEFAULT_BRANCH_NAMES)
    used_names = used_names if used_names is not None else set()
    records: list[dict[str, object]] = []
    topic_pairs: list[tuple[dict[str, object], str]] = []
    for config in DOMAIN_CONFIGS:
        for topic in config["topics"]:
            topic_pairs.append((config, topic))

    while len(records) < target_count:
        rng.shuffle(topic_pairs)
        for config, topic in topic_pairs:
            records.append(
                build_record(
                    rng,
                    start_ordinal + len(records),
                    config,
                    topic,
                    branch_pool,
                    used_names,
                )
            )
            if len(records) >= target_count:
                break

    columns = [
        "dash_id",
        "dash_dsply_name",
        "folder_all_route",
        "build_main",
        "dept_cls",
        "dash_rat",
        "bbk_name",
        "bbk_id",
        "dept_fname",
        "level1_dept_name",
        "level2_dept_name",
        "level3_dept_name",
        "level4_dept_name",
        "chart_cnt",
        "be_author_cnt",
        "fir_be_visit_tm",
        "recnt_be_visit_tm",
        "m3_vis_cnt",
        "mon_add_idf",
        "dm_nm",
    ]
    return pd.DataFrame(records, columns=columns)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Generate mock BI dashboard parquet data.")
    parser.add_argument(
        "--mode",
        choices=["overwrite", "append"],
        default="overwrite",
        help="overwrite replaces the parquet file; append inserts generated rows into the existing parquet file.",
    )
    parser.add_argument(
        "--count",
        type=int,
        default=200,
        help="number of dashboard rows to generate in this run.",
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=OUTPUT_PATH,
        help="target parquet path.",
    )
    parser.add_argument(
        "--branch-name",
        action="append",
        help=(
            "branch names to generate, separated by comma or passed repeatedly. "
            "Defaults to 总行,北京分行,广东分行."
        ),
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    if args.count <= 0:
        raise ValueError("--count must be a positive integer")

    output_path = args.output.resolve()
    output_path.parent.mkdir(parents=True, exist_ok=True)
    branch_names = parse_branch_names(args.branch_name)
    branch_pool = build_branch_pool(branch_names)

    existing = pd.DataFrame()
    if args.mode == "append" and output_path.exists():
        existing = pd.read_parquet(output_path)

    used_names = set(existing["dash_dsply_name"]) if not existing.empty else set()
    new_dashboards = make_dashboards(
        args.count,
        start_ordinal=len(existing) + 1,
        branch_pool=branch_pool,
        used_names=used_names,
    )
    dashboards = pd.concat([existing, new_dashboards], ignore_index=True) if not existing.empty else new_dashboards
    dashboards.to_parquet(output_path, index=False)

    duplicate_count = int(dashboards["dash_id"].duplicated().sum())
    duplicate_name_count = int(dashboards["dash_dsply_name"].duplicated().sum())
    print(f"mode: {args.mode}")
    print(f"branch names: {', '.join(branch_names)}")
    print(f"generated rows: {len(new_dashboards)}")
    print(f"wrote total rows: {len(dashboards)} to {output_path}")
    print(f"duplicate dash_id: {duplicate_count}")
    print(f"duplicate dash_dsply_name: {duplicate_name_count}")
    print(f"columns: {', '.join(dashboards.columns)}")
    print("dash_rat:", dashboards["dash_rat"].value_counts().to_dict())
    print("chart_cnt:", int(dashboards["chart_cnt"].min()), int(dashboards["chart_cnt"].max()))
    print("level1_dept_name:", dashboards["level1_dept_name"].value_counts().to_dict())


if __name__ == "__main__":
    main()