262 lines
13 KiB
Python
262 lines
13 KiB
Python
"""画像模块演示/测试数据种子（5 条，三张表都写）。
|
||
|
||
用法：
|
||
|
||
.\\.venv\\Scripts\\python.exe tools\\seed_profile_demo.py
|
||
|
||
## 为什么是"5 条 + 边界覆盖"
|
||
|
||
用户 2026-09-10 决定：**画像虚拟数据先做 5 条**，用于画像读取、适当性判定与后续演示。
|
||
5 条不是随机凑数，而是覆盖**风险等级与测评有效期的边界**：
|
||
|
||
| customer_id | 画像 | 等级 | 测评有效期 | 覆盖目的 |
|
||
|---|---|---|---|---|
|
||
| 9101 | 企业主 / 高净值 | C5 | 有效 | 最高风险等级 |
|
||
| 9102 | 中学教师 / 中产 | C3 | 有效 | 中间等级（R3 产品可买、R5 不可） |
|
||
| 9103 | 退休职工 | C1 | 有效 | 最低风险承受能力 |
|
||
| 9104 | 互联网产品经理 / 年轻白领 | C2 | 有效 | 新手 + 低资产 |
|
||
| 9001 | 既有测试客户 | C1 | **已过期 45 天** | **失败关闭**：过期测评不得放行 |
|
||
|
||
## 采用"底座版"画像字段（不是老师需求文档的四维加权版）
|
||
|
||
用户已裁定：画像字段与采集规则**按底座做**。老师版要求**新增 7 个字段**，
|
||
而项目铁律禁止改已有字段定义 —— 故只能走底座既有的 16 字段。
|
||
来源：`docs/superpowers/analysis/2026-09-10-范围重定位与画像讨论记录.md`。
|
||
|
||
## 两个已实测的技术坑（实现时必须遵守）
|
||
|
||
1. **`profile_snapshots.current_customer_id` 不是生成列**，而是带唯一键
|
||
`uk_profile_snapshot_current` 的普通列 —— `is_current=1` 时**必须显式写入**且等于 `customer_id`，
|
||
否则同一个 NULL 会在第二条记录上撞唯一键。（与 `agent_reply_template.active_key` 那个**真生成列**不同。）
|
||
2. `uk_profile_snapshot_version (customer_id, version)`：版本号按客户唯一，本脚本统一写 `version=1`。
|
||
|
||
## 幂等性
|
||
|
||
重复执行先删除**本脚本所造客户**（9101-9104 + 9001）在这三张表里的数据再重建。
|
||
**不删 `interaction_audit`**（审计留痕不可删），也**不动其它客户**。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import json
|
||
import sys
|
||
from datetime import UTC, datetime, timedelta
|
||
from pathlib import Path
|
||
from uuid import uuid4
|
||
|
||
from sqlalchemy import bindparam, text as sql
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
|
||
from app.infrastructure.db import SessionFactory # noqa: E402
|
||
|
||
CUSTOMER_ROLE_ID = 9001
|
||
|
||
#: 5 条虚拟画像。`valid_days` 为负表示测评**已过期**（用于验证失败关闭）。
|
||
PROFILES: tuple[dict[str, object], ...] = (
|
||
{
|
||
"user_id": 9101, "user_no": "T-PRF-01", "username": "profile_c5",
|
||
"real_name": "陈宏远", "birth_year": 1972, "occupation": "企业主",
|
||
"investor_type": "C5", "valid_days": 300,
|
||
"horizon": "long_term", "assets": ["equity_fund", "private_equity", "structured"],
|
||
"frequency": "high", "total_asset": "12800000.00", "behavior": 88,
|
||
"risk_tags": ["aggressive", "high_net_worth", "experienced"],
|
||
"tier": "diamond", "score": 92,
|
||
},
|
||
{
|
||
"user_id": 9102, "user_no": "T-PRF-02", "username": "profile_c3",
|
||
"real_name": "李静怡", "birth_year": 1985, "occupation": "中学教师",
|
||
"investor_type": "C3", "valid_days": 200,
|
||
"horizon": "medium_term", "assets": ["bond_fund", "money_fund", "bank_wm"],
|
||
"frequency": "medium", "total_asset": "860000.00", "behavior": 61,
|
||
"risk_tags": ["balanced", "stable_income"],
|
||
"tier": "platinum", "score": 63,
|
||
},
|
||
{
|
||
"user_id": 9103, "user_no": "T-PRF-03", "username": "profile_c1",
|
||
"real_name": "王秀兰", "birth_year": 1958, "occupation": "退休职工",
|
||
"investor_type": "C1", "valid_days": 120,
|
||
"horizon": "short_term", "assets": ["money_fund", "bank_wm"],
|
||
"frequency": "low", "total_asset": "420000.00", "behavior": 33,
|
||
"risk_tags": ["conservative", "retired", "capital_preservation"],
|
||
"tier": "gold", "score": 31,
|
||
},
|
||
{
|
||
"user_id": 9104, "user_no": "T-PRF-04", "username": "profile_c2",
|
||
"real_name": "张一鸣", "birth_year": 1996, "occupation": "互联网产品经理",
|
||
"investor_type": "C2", "valid_days": 180,
|
||
"horizon": "medium_term", "assets": ["money_fund", "index_fund"],
|
||
"frequency": "medium", "total_asset": "150000.00", "behavior": 47,
|
||
"risk_tags": ["cautious", "young_professional", "new_investor"],
|
||
"tier": "gold", "score": 42,
|
||
},
|
||
{
|
||
# 关键边界：测评**已过期**。用于验证画像读取与适配置性的失败关闭。
|
||
"user_id": 9001, "user_no": None, "username": None,
|
||
"real_name": "测试客户", "birth_year": 1990, "occupation": "软件工程师",
|
||
"investor_type": "C1", "valid_days": -45,
|
||
"horizon": "short_term", "assets": ["money_fund"],
|
||
"frequency": "low", "total_asset": "60000.00", "behavior": 22,
|
||
"risk_tags": ["expired_assessment", "conservative"],
|
||
"tier": "gold", "score": 35,
|
||
},
|
||
)
|
||
|
||
PROFILE_TABLE = "fin_customer_profile"
|
||
ASSESSMENT_TABLE = "fin_risk_assessment"
|
||
SNAPSHOT_TABLE = "profile_snapshots"
|
||
|
||
|
||
async def _ensure_customer(session: object, spec: dict[str, object], now: datetime) -> None:
|
||
"""为新造客户建 `sys_user` 行与 customer 角色（9001 是既有账号，跳过）。"""
|
||
if spec["user_no"] is None:
|
||
return
|
||
exists = await session.scalar( # type: ignore[attr-defined]
|
||
sql("SELECT id FROM sys_user WHERE id = :i"), {"i": spec["user_id"]}
|
||
)
|
||
if exists:
|
||
return
|
||
await session.execute( # type: ignore[attr-defined]
|
||
sql("""
|
||
INSERT INTO sys_user (id, user_no, username, password_hash, user_type,
|
||
professional_investor_status, fund_account_status, status, created_at, updated_at)
|
||
VALUES (:id, :no, :name, :pwd, 'customer', 'none', 'opened', '正常', :now, :now)
|
||
"""),
|
||
{"id": spec["user_id"], "no": spec["user_no"], "name": spec["username"],
|
||
"pwd": hashlib.sha256(f"profile-{spec['user_id']}".encode()).hexdigest(), "now": now},
|
||
)
|
||
await session.execute( # type: ignore[attr-defined]
|
||
# 列名是 `assigned_at`（不是 `created_at`）——按现库 DDL 口径写入。
|
||
sql("INSERT INTO sys_user_role (user_id, role_id, assigned_at) VALUES (:u, :r, :now)"),
|
||
{"u": spec["user_id"], "r": CUSTOMER_ROLE_ID, "now": now},
|
||
)
|
||
|
||
|
||
async def _fetch_count(session: object, table: str, ids: list[int]) -> int:
|
||
"""按 customer_id 集合计数（`IN` 的绑定值必须是**元组**，asyncmy 不接受 list）。"""
|
||
return int(await session.scalar( # type: ignore[attr-defined]
|
||
sql(f"SELECT COUNT(*) FROM {table} WHERE customer_id IN :ids").bindparams(
|
||
bindparam("ids", value=tuple(ids), expanding=True)
|
||
)
|
||
))
|
||
|
||
|
||
async def seed() -> list[int]:
|
||
"""幂等写入 5 条画像数据，返回涉及的 customer_id 列表。"""
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
ids = [int(spec["user_id"]) for spec in PROFILES]
|
||
id_tuple = tuple(ids)
|
||
async with SessionFactory() as session:
|
||
for table in (SNAPSHOT_TABLE, ASSESSMENT_TABLE, PROFILE_TABLE):
|
||
await session.execute(
|
||
sql(f"DELETE FROM {table} WHERE customer_id IN :ids").bindparams(
|
||
bindparam("ids", value=id_tuple, expanding=True)
|
||
)
|
||
)
|
||
|
||
for spec in PROFILES:
|
||
cid = int(spec["user_id"])
|
||
await _ensure_customer(session, spec, now)
|
||
valid_until = now + timedelta(days=int(spec["valid_days"]))
|
||
|
||
await session.execute(sql(f"""
|
||
INSERT INTO {PROFILE_TABLE} (customer_id, trade_account, real_name, birth_date,
|
||
occupation, mobile_masked, investor_type, investment_horizon,
|
||
preferred_asset_class, trading_frequency, last_active_at, total_asset,
|
||
behavior_score, risk_tags, opened_at, updated_at)
|
||
VALUES (:cid, :acct, :name, :birth, :occ, :mobile, :itype, :horizon,
|
||
:assets, :freq, :last_active, :total, :behavior, :tags, :opened, :now)
|
||
"""), {
|
||
"cid": cid, "acct": f"TA{cid}0001", "name": spec["real_name"],
|
||
"birth": f"{spec['birth_year']}-06-15", "occ": spec["occupation"],
|
||
"mobile": f"138****{cid % 10000:04d}", "itype": spec["investor_type"],
|
||
"horizon": spec["horizon"], "assets": json.dumps(spec["assets"]),
|
||
"freq": spec["frequency"],
|
||
"last_active": now - timedelta(days=3), "total": spec["total_asset"],
|
||
"behavior": spec["behavior"],
|
||
"tags": json.dumps(spec["risk_tags"], ensure_ascii=False),
|
||
"opened": now - timedelta(days=800), "now": now,
|
||
})
|
||
|
||
# `fin_risk_assessment.id` **没有 AUTO_INCREMENT**（实测 AUTO_INCREMENT=None），
|
||
# 必须显式给值；用 99xxx 段，避开任何自增序列。
|
||
await session.execute(sql(f"""
|
||
INSERT INTO {ASSESSMENT_TABLE} (id, customer_id, questionnaire_version, answers,
|
||
total_score, investor_type, assessed_at, valid_until, created_at)
|
||
VALUES (:aid, :cid, 'v2026.1', :answers, :score, :itype, :assessed, :valid, :now)
|
||
"""), {
|
||
"aid": 99000 + cid % 1000, "cid": cid,
|
||
"answers": json.dumps({"q1": "A", "q2": "B", "q3": "C"}, ensure_ascii=False),
|
||
"score": spec["score"], "itype": spec["investor_type"],
|
||
"assessed": valid_until - timedelta(days=365), "valid": valid_until, "now": now,
|
||
})
|
||
|
||
snapshot = {
|
||
"investor_type": spec["investor_type"],
|
||
"investment_horizon": spec["horizon"],
|
||
"preferred_asset_class": spec["assets"],
|
||
"trading_frequency": spec["frequency"],
|
||
"total_asset": spec["total_asset"],
|
||
"behavior_score": spec["behavior"],
|
||
"risk_tags": spec["risk_tags"],
|
||
"customer_tier": spec["tier"],
|
||
"assessment_valid_until": valid_until.isoformat(),
|
||
"assessment_expired": valid_until <= now,
|
||
}
|
||
body = json.dumps(snapshot, ensure_ascii=False, sort_keys=True)
|
||
await session.execute(sql(f"""
|
||
INSERT INTO {SNAPSHOT_TABLE} (profile_uuid, customer_id, version, snapshot,
|
||
generation_basis, snapshot_hash, is_current, current_customer_id,
|
||
generated_at, created_at, updated_at)
|
||
VALUES (:uuid, :cid, 1, :snap, :basis, :hash, 1, :cid, :now, :now, :now)
|
||
"""), {
|
||
"uuid": str(uuid4()), "cid": cid, "snap": body,
|
||
# `current_customer_id` **必须显式写入**（不是生成列，见模块 docstring 坑 1）。
|
||
"basis": json.dumps(
|
||
{"sources": [ASSESSMENT_TABLE, PROFILE_TABLE], "seed": "profile_demo"},
|
||
ensure_ascii=False),
|
||
"hash": hashlib.sha256(body.encode()).hexdigest(), "now": now,
|
||
})
|
||
await session.commit()
|
||
return ids
|
||
|
||
|
||
async def verify(ids: list[int]) -> None:
|
||
"""自校验：三张表各有 5 条、每客户恰好一条 is_current=1、过期边界确实存在。"""
|
||
id_tuple = tuple(ids)
|
||
async with SessionFactory() as session:
|
||
for table in (PROFILE_TABLE, ASSESSMENT_TABLE, SNAPSHOT_TABLE):
|
||
n = await _fetch_count(session, table, ids)
|
||
print(f" {table:<24} {n} 行（期望 5）")
|
||
|
||
dup = (await session.execute(sql(f"""
|
||
SELECT customer_id, COUNT(*) AS n FROM {SNAPSHOT_TABLE}
|
||
WHERE customer_id IN :ids AND is_current = 1
|
||
GROUP BY customer_id HAVING n <> 1
|
||
""").bindparams(bindparam("ids", value=id_tuple, expanding=True)))).mappings().all()
|
||
print(f" is_current=1 非唯一的客户: {[dict(d) for d in dup] or '无'}")
|
||
|
||
expired = (await session.execute(sql(f"""
|
||
SELECT customer_id FROM {ASSESSMENT_TABLE}
|
||
WHERE customer_id IN :ids AND valid_until <= NOW()
|
||
""").bindparams(bindparam("ids", value=id_tuple, expanding=True)))).scalars().all()
|
||
print(f" 测评已过期的客户: {sorted(expired)}（期望 [9001]）")
|
||
|
||
levels = (await session.execute(sql(f"""
|
||
SELECT investor_type, COUNT(*) AS n FROM {PROFILE_TABLE}
|
||
WHERE customer_id IN :ids GROUP BY investor_type ORDER BY investor_type
|
||
""").bindparams(bindparam("ids", value=id_tuple, expanding=True)))).mappings().all()
|
||
print(f" 风险等级分布: {[(r['investor_type'], r['n']) for r in levels]}")
|
||
|
||
|
||
async def main() -> None:
|
||
ids = await seed()
|
||
print(f"已写入 {len(ids)} 条画像数据: {ids}")
|
||
await verify(ids)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|