新增异常账户基线写库服务

lsfx-mock-server/services/abnormal_account_baseline_service.py

@@ -0,0 +1,114 @@
+from typing import List
+
+from config.settings import settings
+
+
+class AbnormalAccountBaselineService:
+    """异常账户基线写库服务。"""
+
+    UPSERT_SQL = """
+        INSERT INTO ccdi_account_info (
+          account_no,
+          account_type,
+          account_name,
+          owner_type,
+          owner_id,
+          bank,
+          bank_code,
+          currency,
+          is_self_account,
+          trans_risk_level,
+          status,
+          effective_date,
+          invalid_date,
+          create_by,
+          update_by
+        )
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+        ON DUPLICATE KEY UPDATE
+          account_name = VALUES(account_name),
+          owner_type = VALUES(owner_type),
+          owner_id = VALUES(owner_id),
+          bank = VALUES(bank),
+          bank_code = VALUES(bank_code),
+          currency = VALUES(currency),
+          is_self_account = VALUES(is_self_account),
+          trans_risk_level = VALUES(trans_risk_level),
+          status = VALUES(status),
+          effective_date = VALUES(effective_date),
+          invalid_date = VALUES(invalid_date),
+          update_by = VALUES(update_by),
+          update_time = NOW()
+    """
+
+    def __init__(self):
+        self.db_config = {
+            "host": settings.CCDI_DB_HOST,
+            "port": settings.CCDI_DB_PORT,
+            "database": settings.CCDI_DB_NAME,
+            "username": settings.CCDI_DB_USERNAME,
+            "password": settings.CCDI_DB_PASSWORD,
+            "connect_timeout_seconds": settings.CCDI_DB_CONNECT_TIMEOUT_SECONDS,
+        }
+
+    def _connect(self):
+        try:
+            import pymysql
+        except ImportError as exc:
+            raise RuntimeError("缺少 PyMySQL 依赖，无法写入异常账户基线") from exc
+
+        return pymysql.connect(
+            host=settings.CCDI_DB_HOST,
+            port=settings.CCDI_DB_PORT,
+            user=settings.CCDI_DB_USERNAME,
+            password=settings.CCDI_DB_PASSWORD,
+            database=settings.CCDI_DB_NAME,
+            charset="utf8mb4",
+            connect_timeout=settings.CCDI_DB_CONNECT_TIMEOUT_SECONDS,
+            autocommit=False,
+        )
+
+    def _validate_fact_owner(self, staff_id_card: str, abnormal_accounts: List[dict]) -> None:
+        for account_fact in abnormal_accounts:
+            owner_id_card = account_fact.get("owner_id_card")
+            if owner_id_card != staff_id_card:
+                raise RuntimeError(
+                    f"异常账户 owner_id_card 与 staff_id_card 不一致: {owner_id_card}"
+                )
+
+    def _build_upsert_params(self, account_fact: dict) -> tuple:
+        return (
+            account_fact["account_no"],
+            "DEBIT",
+            account_fact["account_name"],
+            "EMPLOYEE",
+            account_fact["owner_id_card"],
+            "兰溪农商银行",
+            "LXNCSY",
+            "CNY",
+            1,
+            "HIGH",
+            account_fact["status"],
+            account_fact["effective_date"],
+            account_fact.get("invalid_date"),
+            "lsfx-mock-server",
+            "lsfx-mock-server",
+        )
+
+    def apply(self, staff_id_card: str, abnormal_accounts: List[dict]) -> None:
+        if not abnormal_accounts:
+            return
+
+        self._validate_fact_owner(staff_id_card, abnormal_accounts)
+
+        connection = self._connect()
+        try:
+            with connection.cursor() as cursor:
+                for account_fact in abnormal_accounts:
+                    cursor.execute(self.UPSERT_SQL, self._build_upsert_params(account_fact))
+            connection.commit()
+        except Exception:
+            connection.rollback()
+            raise
+        finally:
+            connection.close()