修复Mock流水按数据库员工及亲属绑定身份证

lsfx-mock-server/services/staff_identity_repository.py

@@ -0,0 +1,87 @@
+from typing import Dict, List
+
+from config.settings import settings
+
+
+class StaffIdentityRepository:
+    """从员工信息库中读取员工及亲属身份证信息。"""
+
+    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,
+            cursorclass=pymysql.cursors.DictCursor,
+        )
+
+    def select_random_staff_with_families(self) -> Dict[str, object]:
+        """随机选择一个员工，并读取其有效亲属证件号。"""
+        with self._connect() as connection:
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    """
+                    SELECT s.name AS staff_name, s.id_card AS staff_id_card
+                    FROM ccdi_base_staff s
+                    WHERE COALESCE(TRIM(s.id_card), '') <> ''
+                      AND s.status = '0'
+                      AND EXISTS (
+                          SELECT 1
+                          FROM ccdi_staff_fmy_relation r
+                          WHERE r.person_id = s.id_card
+                            AND r.status = 1
+                            AND COALESCE(TRIM(r.relation_cert_no), '') <> ''
+                      )
+                    ORDER BY RAND()
+                    LIMIT 1
+                    """
+                )
+                staff = cursor.fetchone()
+                if not staff:
+                    cursor.execute(
+                        """
+                        SELECT s.name AS staff_name, s.id_card AS staff_id_card
+                        FROM ccdi_base_staff s
+                        WHERE COALESCE(TRIM(s.id_card), '') <> ''
+                          AND s.status = '0'
+                        ORDER BY RAND()
+                        LIMIT 1
+                        """
+                    )
+                    staff = cursor.fetchone()
+
+                if not staff:
+                    raise RuntimeError("员工信息库中不存在可用身份证号")
+
+                cursor.execute(
+                    """
+                    SELECT DISTINCT relation_cert_no
+                    FROM ccdi_staff_fmy_relation
+                    WHERE person_id = %s
+                      AND status = 1
+                      AND COALESCE(TRIM(relation_cert_no), '') <> ''
+                    ORDER BY relation_cert_no
+                    """,
+                    (staff["staff_id_card"],),
+                )
+                family_rows = cursor.fetchall()
+
+        family_id_cards: List[str] = [
+            row["relation_cert_no"]
+            for row in family_rows
+            if row["relation_cert_no"] != staff["staff_id_card"]
+        ]
+
+        return {
+            "staff_name": staff["staff_name"],
+            "staff_id_card": staff["staff_id_card"],
+            "family_id_cards": family_id_cards,
+        }