zhanghongyu_0626
c9cac1cdf9
- 在 `risk_suitability_log` 表中新增 `check_source` 枚举值 `platform`，支持代销平台的适当性检查。 - 更新相关 SQL 脚本以适应新的数据结构，确保数据一致性。 - 修改文档以反映 API 的最新状态和测试基线，确保文档与实现保持一致。 - 测试基线更新至 530 passed, 0 skipped，确保系统稳定性。 此更新为代销平台提供了更全面的适当性检查能力，提升了系统的功能性与可维护性。
112 lines
3.5 KiB
Python
112 lines
3.5 KiB
Python
"""Repair UTF-8 demo seed data (core display_name + risk_aml_list).
|
|
|
|
Windows mysql CLI pipe may corrupt CJK; use SQLAlchemy + utf-8 file reads instead.
|
|
Idempotent: safe to re-run after partial bootstrap.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.config.settings import settings
|
|
from app.utils.db import get_engine
|
|
|
|
CUSTOMER_SEED = ROOT / "scripts/core/03-seed-customers.sql"
|
|
AML_SEED = ROOT / "scripts/agent/seed-aml-list.sql"
|
|
RISK_DEMO = ROOT / "scripts/demo/prepare_risk_demo.sql"
|
|
|
|
_ROW_RE = re.compile(r"\('(CUST-[^']+)',\s*'([^']+)'")
|
|
|
|
|
|
def _parse_customer_names() -> list[tuple[str, str]]:
|
|
content = CUSTOMER_SEED.read_text(encoding="utf-8")
|
|
block = content.split("INSERT INTO core_customer_risk")[0]
|
|
return _ROW_RE.findall(block)
|
|
|
|
|
|
def _parse_aml_rows() -> list[dict[str, str]]:
|
|
content = AML_SEED.read_text(encoding="utf-8")
|
|
rows: list[dict[str, str]] = []
|
|
for m in re.finditer(
|
|
r"\('(AML-\d+)',\s*'(\w+)',\s*'([^']*)',\s*NULL,\s*NULL,\s*([\d.]+),\s*"
|
|
r"'([^']*)',\s*'([^']*)',\s*'(\d{4}-\d{2}-\d{2})'\)",
|
|
content,
|
|
):
|
|
rows.append(
|
|
{
|
|
"list_id": m.group(1),
|
|
"list_type": m.group(2),
|
|
"full_name": m.group(3),
|
|
"match_threshold": m.group(4),
|
|
"source": m.group(5),
|
|
"list_version": m.group(6),
|
|
"effective_date": m.group(7),
|
|
}
|
|
)
|
|
if len(rows) != 8:
|
|
raise RuntimeError(f"expected 8 AML rows, parsed {len(rows)}")
|
|
return rows
|
|
|
|
|
|
def _run_sql_file(engine, path: Path) -> None:
|
|
statements = [s.strip() for s in path.read_text(encoding="utf-8").split(";") if s.strip()]
|
|
with engine.begin() as conn:
|
|
for stmt in statements:
|
|
if stmt.upper().startswith("USE "):
|
|
continue
|
|
conn.execute(text(stmt))
|
|
|
|
|
|
def main() -> None:
|
|
core = get_engine(settings.mysql_core_database)
|
|
agent = get_engine(settings.mysql_database)
|
|
|
|
names = _parse_customer_names()
|
|
with core.begin() as conn:
|
|
for customer_id, display_name in names:
|
|
conn.execute(
|
|
text("UPDATE core_customer SET display_name = :n WHERE customer_id = :id"),
|
|
{"id": customer_id, "n": display_name},
|
|
)
|
|
print(f"updated {len(names)} core_customer.display_name")
|
|
|
|
aml_rows = _parse_aml_rows()
|
|
with agent.begin() as conn:
|
|
conn.execute(text("DELETE FROM risk_aml_list"))
|
|
for row in aml_rows:
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO risk_aml_list
|
|
(list_id, list_type, full_name, id_no, bank_card_no,
|
|
match_threshold, source, list_version, effective_date, is_active)
|
|
VALUES
|
|
(:list_id, :list_type, :full_name, NULL, NULL,
|
|
:match_threshold, :source, :list_version, :effective_date, 1)
|
|
"""
|
|
),
|
|
row,
|
|
)
|
|
print(f"reseeded {len(aml_rows)} risk_aml_list rows")
|
|
|
|
_run_sql_file(core, RISK_DEMO)
|
|
print("ran prepare_risk_demo.sql")
|
|
|
|
with core.connect() as conn:
|
|
sample = conn.execute(
|
|
text("SELECT customer_id, display_name FROM core_customer WHERE customer_id = 'CUST-3001'")
|
|
).one()
|
|
print("verify CUST-3001:", sample)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|