feat(skill): 新增 find-field —— 跨库盘点字段 + 行数 + 视图溯源

新建 .claude/skills/find-field/：
- SKILL.md：调用方式 / 参数 / 输出格式 / 6 条踩坑 / 安全约束
- scripts/find_field.py：从 web3 connection_preset 读连接，三方言
  (MySQL/达梦/Oracle) 扫列名 + COUNT(*) + USER_DEPENDENCIES 溯源 V_* 视图

安全：
- 数据库连接只能选 web3 已存的（13 个预设），不允许临时输入密码
- 只读 SELECT，不发 DML（CLAUDE.md 原则）

附带 web3/tests/_probe_gjj80_bankcard.py：
GJJ80 一次性探测脚本，作为本次 skill 命名的原型被保留作历史对照。
SKILL.md 里点名「已被本 skill 取代」，但保留 git 历史。

验证：find-field huanggang-gjj 银行卡号 → 与一次性脚本结果完全一致
(V_JCFX_DWXX 11768 + V_JCFX_GRXX 567523 + 基表 GJ_DWXX/GJ_GRXX)

.claude/skills/find-field/SKILL.md

+---
+name: find-field
+description: 在 web3 已存储的数据库连接里，按关键字扫字段名 + 取行数 + 视图溯源。专为「跨库盘点字段 / 数据量」场景设计，例如「GJJ80 哪些表有银行卡号」「mysql.smart-build 里哪些表带 phone 字段」。数据库连接只能选 web3 里存的（不允许临时输入），保证安全 + 可审计。
+metadata:
+  argument-hint: "<db-alias> <keyword> [more keywords ...]"
+  allowed-tools: Bash
+---
+
+# find-field —— 跨库盘点字段 + 行数
+
+## 何时使用
+
+- 用户说「找一下 XX 库里哪些表有 YY 字段」「YY 库有多少数据」
+- 建数据质量任务前盘点字段位置 + 数据量
+- 跨库对比（同一字段名在多个数据库里的分布）
+- 视图溯源（V_* 视图背后是哪张基表 + 基表行数）
+
+**适用数据库**：MySQL / 达梦 / Oracle 三种方言都支持，统一走 web3 的 `DBConnection` 封装。
+
+## 调用方式
+
+**先通过 Bash 调用脚本**，把脚本输出原样回给用户（脚本输出已经按表格排好）。
+
+```bash
+# Windows + Git Bash / PowerShell / cmd 通用
+python .claude/skills/find-field/scripts/find_field.py --db <db-alias> --keywords <kw1> [kw2 ...]
+
+# 只列已存储的连接（用户没指定时，先 --list 拿清单让用户挑）
+python .claude/skills/find-field/scripts/find_field.py --list
+
+# Windows cmd / PowerShell 中文乱码 → 加 PYTHONUTF8=1
+set PYTHONUTF8=1
+python .claude/skills/find-field/scripts/find_field.py --db huanggang-gjj --keywords 银行卡号
+```
+
+## 参数
+
+| 参数 | 必填 | 说明 |
+|---|---|---|
+| `--db` | 是（除非 `--list`） | web3 已存储的连接别名。允许：①精确 `name`（如 `huanggang-gjj`）；②模糊 `display`（如「黄冈公积金」）；找不到时报错并列出所有 |
+| `--keywords` | 是（除非 `--list`） | 1+ 个列名关键字，大小写不敏感，子串匹配（`LIKE %kw%`）。中英文都行 |
+| `--list` | 否 | 只列出 web3 已存的连接，不查库 |
+| `--db-path` | 否 | 覆盖 web3.db 路径（默认读 `web3.backend.config.DB_PATH`，可用环境变量 `WEB3_DB_PATH` 改） |
+
+## 输出格式
+
+```
+================================================================================
+DB      : <display>（<db_type>）
+URI     : <db_type>://<user>@<host>:<port>/<database>
+关键字  : [<kw1>, <kw2>, ...]
+================================================================================
+
+[1/3] 扫字段...
+  → 命中 N 个（表, 列）
+
+[2/3] 取每张表行数...
+
+================================================================================
+结果汇总
+================================================================================
+表名                          行数      列名                类型              可空   表注释
+------------------------------------------------------------------------------------------------------------------------
+V_JCFX_DWXX              11,768    银行账号              VARCHAR2(30)    Y
+V_JCFX_GRXX             567,523    银行账号              VARCHAR2(30)    Y
+------------------------------------------------------------------------------------------------------------------------
+匹配表数: 2   匹配列数: 2   行数总计: 579,291
+
+[3/3] 视图依赖溯源...
+
+  VIEW  V_JCFX_DWXX  →  引用：
+    BASE TABLE  GJJ80.GJ_DWXX                          行数=11,768
+  VIEW  V_JCFX_GRXX  →  引用：
+    BASE TABLE  GJJ80.GJ_GRXX                         行数=567,523
+```
+
+三段：
+1. **结果汇总**：每个（表, 列）一行，行数就是 COUNT(*)，按表聚合展示
+2. **视图依赖溯源**（仅 Oracle/达梦）：命中以 `V_` 开头的视图时，`USER_DEPENDENCIES` 找背后基表 + 行数
+3. 子串匹配会带干扰项（如「账号」会同时命中「银行账号/单位账号/个人账号」），用户在表格里挑真的那张
+
+## 安全约束
+
+- **只能用 web3 里已存的连接**：脚本启动先读 `connection_preset`，找不到 --db 报错并列出全部，不允许临时输入 host/port/密码
+- **只读**：用 `DBConfig` + `DBConnection`，只跑 SELECT，没碰 DML（项目原则 CLAUDE.md「永远不要使用数据操作语句」）
+- **密码脱敏**：脚本输出不打印 password（DBConfig 里也不会带出来）；URI 行只打 `user@host:port/database`
+
+## 调用流程（Claude 该怎么做）
+
+1. 用户调用 skill，参数格式：`<db-alias> <keyword> [<keyword> ...]`
+2. **如果用户没给 `--db`**：先跑 `--list` 拿连接清单，列出 `name` / `display` 给用户挑，等用户回
+3. **如果用户给的 db 名字找不到**：脚本会自己报错并列出全部 13 个连接 → 把脚本错误输出原样回给用户，让他/她挑
+4. 拿到正确 --db 和 --keywords 后，跑脚本 → 把输出原样回给用户（不要二次解析、不要省略表格、不要自己写总结）
+5. **可选**：跑完后帮用户做一个 1~2 行的中文要点总结（命中几表 / 哪张最大 / 干扰项提示），但表格内容必须照搬脚本输出
+
+## 踩坑
+
+1. **Windows cmd / PowerShell 中文乱码**：cmd 默认 GBK，脚本输出里中文是 UTF-8 → 表格列对不齐。
+   解决：调用前 `set PYTHONUTF8=1`（cmd）或 `$env:PYTHONUTF8=1`（PowerShell），或直接用 Git Bash。
+2. **Oracle thin 模式连不上老 Oracle（11g 及以下）**：脚本读 `connection_preset.oracle_client_dir` 自动切 thick 模式，**不需要用户配置**。
+3. **db_adapter 归一化列名为小写**：脚本内访问 dict 都用 lowercase key（`r["table_name"]` 不是 `r["TABLE_NAME"]`），跟 web3 后端一致
+4. **大量「账号/帐号」类干扰项**：「账号」做关键字会同时命中「银行账号」「单位账号」「社保账号」「个人账号」等——子串匹配通病，靠人工看表注释/列注释筛；模糊关键字越短，干扰越多
+5. **视图行数 = COUNT(*)**：有些 V_* 是带 WHERE 的视图，行数不等于底层基表行数；脚本默认报视图行数，基表行数在「依赖溯源」段再单独打
+6. **GJJ80 类用户视图权限有限**：`ALL_DEPENDENCIES` 通常查不到，脚本默认走 `USER_DEPENDENCIES`（用户视角），基表不在当前 schema 时会 0 行——这是预期行为
+
+## 实现要点
+
+- 路径：`scripts/find_field.py`（独立脚本，sys.path 注入 `<project_root>` 后 import `web3.backend`）
+- 数据字典查询走方言分流：
+  - Oracle: `ALL_TAB_COLUMNS` + `ALL_COL_COMMENTS`（去重 LEFT JOIN，跟 [list_columns.oracle.sql](../web3/backend/core/sql_templates/info_schema/list_columns.oracle.sql) 同源思路）
+  - MySQL: `INFORMATION_SCHEMA.COLUMNS`
+  - 达梦: `ALL_TAB_COLUMNS`（跟 Oracle 高度一致，占位符 `?`）
+- 行数 COUNT(*) 用方言的 `quote_ident()` 包识别符，避开关键字冲突
+- 视图溯源：仅对 `V_` / `V$` 开头的命中视图跑 `USER_DEPENDENCIES`，附带基表行数
+- 数据治理原则：只读 SELECT，不发 DML
+
+## 相关文件
+
+- 脚本：[scripts/find_field.py](scripts/find_field.py)
+- web3 底层连接管理：[web3/backend/core/db_adapter.py](../web3/backend/core/db_adapter.py)
+- web3 连接预设表 schema：[web3/backend/db/schema.sql](../web3/backend/db/schema.sql)（`connection_preset` 表）
+- 一次性的 GJJ80 探测脚本（已被本 skill 取代）：[web3/tests/_probe_gjj80_bankcard.py](../web3/tests/_probe_gjj80_bankcard.py)

.claude/skills/find-field/scripts/find_field.py

+#!/usr/bin/env python
+"""
+find_field.py —— 在 web3 已保存的数据库连接里，按关键字扫字段名 + 取行数 + 视图溯源。
+
+跟一次性探测脚本 _probe_gjj80_bankcard.py 的区别：
+  - 不在代码里硬编码 GJJ80 连接，而是从 web3 数据库的 connection_preset 读
+  - 用户只能传「已在 web3 里存的连接别名」，传错立刻报错并列出所有可用别名
+  - 支持 MySQL / 达梦 / Oracle 三种方言
+
+用法（由 SKILL.md 触发，调用方负责传 cwd 和环境变量）：
+    python find_field.py --db GJJ80 --keywords 银行卡号 银行账号
+    python find_field.py --db huanggang-gjj --keywords bank_card_no card_no
+    python find_field.py --list           # 只列出所有可用连接
+
+数据治理原则：只用 SELECT，不发 DML。
+"""
+from __future__ import annotations
+
+import argparse
+import sqlite3
+import sys
+from pathlib import Path
+
+# 复用 web3 后端的连接管理
+#   __file__ = <project_root>/.claude/skills/find-field/scripts/find_field.py
+#   parents[4] = project_root → /web3 是其下的子包
+#   sys.path 要加 project_root（不是 web3/），这样 `import web3.backend.xxx` 才合法
+_PROJECT_ROOT = Path(__file__).resolve().parents[4]
+sys.path.insert(0, str(_PROJECT_ROOT))
+
+from web3.backend.core.db_adapter import DBConfig, DBConnection  # noqa: E402
+from web3.backend.config import DB_PATH as WEB3_DB_PATH          # noqa: E402
+
+
+# ── 1. 读 connection_preset ─────────────────────────────────
+def list_presets(db_path: Path) -> list[dict]:
+    if not db_path.exists():
+        raise SystemExit(
+            f"[!] web3 数据库不存在: {db_path}\n"
+            "    请先启动 web3 后端一次（python -m web3.backend.app），会自动建库 + 灌预设连接"
+        )
+    con = sqlite3.connect(str(db_path))
+    try:
+        rows = con.execute(
+            """
+            SELECT db_type, name, display, host, port, user, password, database, oracle_client_dir, ord
+            FROM connection_preset
+            ORDER BY db_type, ord
+            """
+        ).fetchall()
+    finally:
+        con.close()
+    cols = ["db_type", "name", "display", "host", "port", "user",
+            "password", "database", "oracle_client_dir", "ord"]
+    return [dict(zip(cols, r)) for r in rows]
+
+
+def find_preset(presets: list[dict], db_alias: str) -> dict:
+    """按 db_type 内唯一的 name 字段找。允许用户传「name」「db_type/name」「display(模糊)」"""
+    db_alias = db_alias.strip()
+    # 1) 精确 name 匹配
+    for p in presets:
+        if p["name"] == db_alias:
+            return p
+    # 2) display 模糊匹配（用户口语化输入兜底）
+    matches = [p for p in presets if p["display"] and db_alias in p["display"]]
+    if len(matches) == 1:
+        return matches[0]
+    if len(matches) > 1:
+        raise SystemExit(
+            f"[!] 显示名「{db_alias}」匹配到多个连接，请用 name 精确指定：\n"
+            + "\n".join(f"    - {m['name']}（{m['display']}）" for m in matches)
+        )
+    # 3) 没找到 → 列出所有
+    available = "\n".join(
+        f"    {p['db_type']:<8} {p['name']:<25} {p['display'] or ''}" for p in presets
+    )
+    raise SystemExit(
+        f"[!] 没找到连接「{db_alias}」。\n"
+        f"    web3 已存储的连接（共 {len(presets)} 个）：\n{available}"
+    )
+
+
+# ── 2. 关键字搜索 SQL（按方言切换） ─────────────────────────
+def _build_oracle_match_sql(keywords: list[str]) -> tuple[str, tuple]:
+    """Oracle 走 ALL_TAB_COLUMNS + ALL_COL_COMMENTS（去重 LEFT JOIN）"""
+    like_clause = " OR ".join([f"LOWER(c.COLUMN_NAME) LIKE :{i+2}" for i in range(len(keywords))])
+    sql = f"""
+        SELECT
+            c.OWNER                                          AS owner,
+            c.TABLE_NAME                                     AS table_name,
+            c.COLUMN_NAME                                    AS column_name,
+            CASE
+                WHEN c.DATA_TYPE IN ('VARCHAR2','NVARCHAR2','CHAR','RAW')
+                    THEN c.DATA_TYPE || '(' || c.DATA_LENGTH || ')'
+                ELSE c.DATA_TYPE
+            END                                              AS column_type,
+            c.NULLABLE                                       AS nullable,
+            cc.COMMENTS                                      AS column_comment,
+            (
+                SELECT tc.COMMENTS
+                FROM ALL_TAB_COMMENTS tc
+                WHERE tc.OWNER = c.OWNER AND tc.TABLE_NAME = c.TABLE_NAME
+            )                                                AS table_comment
+        FROM ALL_TAB_COLUMNS c
+        LEFT JOIN (
+            SELECT OWNER, TABLE_NAME, COLUMN_NAME, COMMENTS
+            FROM (
+                SELECT cc.OWNER, cc.TABLE_NAME, cc.COLUMN_NAME, cc.COMMENTS,
+                       ROW_NUMBER() OVER (
+                           PARTITION BY cc.OWNER, cc.TABLE_NAME, cc.COLUMN_NAME
+                           ORDER BY cc.COMMENTS DESC NULLS LAST
+                       ) AS rn
+                FROM ALL_COL_COMMENTS cc
+            )
+            WHERE rn = 1
+        ) cc
+            ON  cc.OWNER      = c.OWNER
+            AND cc.TABLE_NAME = c.TABLE_NAME
+            AND cc.COLUMN_NAME = c.COLUMN_NAME
+        WHERE c.OWNER = UPPER(:1)
+          AND ({like_clause})
+        ORDER BY c.TABLE_NAME, c.COLUMN_ID
+    """
+    params = (":PLACEHOLDER_DUMMY",) + tuple(f"%{kw.lower()}%" for kw in keywords)
+    return sql, params
+
+
+def _build_mysql_match_sql(keywords: list[str], database: str) -> tuple[str, tuple]:
+    """MySQL 走 INFORMATION_SCHEMA.COLUMNS（database_name 不需要 UPPER，默认就是）"""
+    like_clause = " OR ".join([f"LOWER(COLUMN_NAME) LIKE %s" for _ in keywords])
+    sql = f"""
+        SELECT
+            TABLE_SCHEMA                                       AS owner,
+            TABLE_NAME                                         AS table_name,
+            COLUMN_NAME                                        AS column_name,
+            COLUMN_TYPE                                        AS column_type,
+            IS_NULLABLE                                        AS nullable,
+            COLUMN_DEFAULT                                     AS column_default,
+            COLUMN_COMMENT                                     AS column_comment,
+            TABLE_COMMENT                                      AS table_comment
+        FROM INFORMATION_SCHEMA.COLUMNS
+        WHERE TABLE_SCHEMA = %s
+          AND ({like_clause})
+        ORDER BY TABLE_NAME, ORDINAL_POSITION
+    """
+    params = (database,) + tuple(f"%{kw.lower()}%" for kw in keywords)
+    return sql, params
+
+
+def _build_dameng_match_sql(keywords: list[str]) -> tuple[str, tuple]:
+    """达梦走 ALL_TAB_COLUMNS（与 Oracle 高度一致，但占位符是 ? 而不是 :1）"""
+    like_clause = " OR ".join([f"LOWER(c.COLUMN_NAME) LIKE ?" for _ in keywords])
+    sql = f"""
+        SELECT
+            c.OWNER                                            AS owner,
+            c.TABLE_NAME                                       AS table_name,
+            c.COLUMN_NAME                                      AS column_name,
+            c.DATA_TYPE                                        AS data_type,
+            c.NULLABLE                                         AS nullable,
+            cc.COMMENTS                                        AS column_comment,
+            tc.COMMENTS                                        AS table_comment
+        FROM ALL_TAB_COLUMNS c
+        LEFT JOIN (
+            SELECT OWNER, TABLE_NAME, COLUMN_NAME, COMMENTS
+            FROM (
+                SELECT cc.OWNER, cc.TABLE_NAME, cc.COLUMN_NAME, cc.COMMENTS,
+                       ROW_NUMBER() OVER (
+                           PARTITION BY cc.OWNER, cc.TABLE_NAME, cc.COLUMN_NAME
+                           ORDER BY cc.COMMENTS DESC NULLS LAST
+                       ) AS rn
+                FROM ALL_COL_COMMENTS cc
+            )
+            WHERE rn = 1
+        ) cc
+            ON  cc.OWNER      = c.OWNER
+            AND cc.TABLE_NAME = c.TABLE_NAME
+            AND cc.COLUMN_NAME = c.COLUMN_NAME
+        LEFT JOIN ALL_TAB_COMMENTS tc
+            ON  tc.OWNER      = c.OWNER
+            AND tc.TABLE_NAME = c.TABLE_NAME
+        WHERE c.OWNER = UPPER(?)
+          AND ({like_clause})
+        ORDER BY c.TABLE_NAME, c.COLUMN_ID
+    """
+    params = ("PLACEHOLDER_DUMMY",) + tuple(f"%{kw.lower()}%" for kw in keywords)
+    return sql, params
+
+
+# ── 3. 行数 + 视图溯源 ─────────────────────────
+def table_row_count(db: DBConnection, owner: str, table_name: str, db_type: str) -> int | str:
+    """跨方言 COUNT(*)。owner.table 用对应方言的 quote。"""
+    from web3.backend.core.db_adapter import quote_ident
+    ident_owner = quote_ident(owner, db_type)
+    ident_tbl = quote_ident(table_name, db_type)
+    sql = f"SELECT COUNT(*) AS cnt FROM {ident_owner}.{ident_tbl}"
+    try:
+        row = db.fetchone(sql)
+        return row["cnt"]
+    except Exception as e:
+        return f"ERR: {e.__class__.__name__}: {str(e)[:80]}"
+
+
+def trace_view_deps(db: DBConnection, view_names: list[str], schema_for_user: str) -> dict[str, list]:
+    """Oracle / 达梦 USER_DEPENDENCIES 找 view → 基表依赖，附带基表行数。仅对 view 类型生效。"""
+    if not view_names:
+        return {}
+    placeholders = ",".join([f":{i+1}" if _is_oracledb(db) else "%s"
+                             for i in range(len(view_names))])
+    src = "USER_DEPENDENCIES"   # GJJ80/普通用户视角，权限要求低；ALL_DEPENDENCIES 通常查不到
+    sql = f"""
+        SELECT
+            d.NAME            AS dependent_name,
+            d.REFERENCED_OWNER,
+            d.REFERENCED_NAME,
+            d.REFERENCED_TYPE
+        FROM {src} d
+        WHERE d.NAME IN ({placeholders})
+          AND d.REFERENCED_TYPE IN ('TABLE', 'VIEW')
+    """
+    try:
+        rows = db.fetchall(sql, tuple(view_names))
+    except Exception:
+        return {}
+
+    from collections import defaultdict
+    dep = defaultdict(list)
+    for r in rows:
+        dep[r["dependent_name"]].append(r)
+    return dep
+
+
+def _is_oracledb(db: DBConnection) -> bool:
+    return db._driver == "oracledb"
+
+
+# ── 4. 主流程 ─────────────────────────
+def main():
+    ap = argparse.ArgumentParser(
+        description="在 web3 已存的数据库连接里，按关键字扫字段名 + 取行数 + 视图溯源",
+    )
+    ap.add_argument("--db", help="web3 里存的连接别名（name 或 display）")
+    ap.add_argument("--keywords", nargs="+", help="列名关键字（LIKE %kw%，不区分大小写）")
+    ap.add_argument("--list", action="store_true", help="只列出 web3 里的所有连接")
+    ap.add_argument("--db-path", default=str(WEB3_DB_PATH),
+                    help=f"覆盖 web3.db 路径（默认 {WEB3_DB_PATH}）")
+    args = ap.parse_args()
+
+    presets = list_presets(Path(args.db_path))
+
+    if args.list:
+        print(f"web3 已存储的连接（共 {len(presets)} 个）：")
+        print(f"  {'DBTYPE':<8} {'NAME':<25} {'DISPLAY':<20} {'HOST':<18} {'SCHEMA'}")
+        for p in presets:
+            print(f"  {p['db_type']:<8} {p['name']:<25} "
+                  f"{(p['display'] or ''):<20} {p['host']:<18} {p['database']}")
+        return
+
+    if not args.db or not args.keywords:
+        ap.error("需要 --db 和 --keywords；先用 --list 看可用连接")
+
+    preset = find_preset(presets, args.db)
+    cfg = DBConfig(
+        db_type=preset["db_type"],
+        host=preset["host"],
+        port=preset["port"],
+        user=preset["user"],
+        password=preset["password"],
+        database=preset["database"],
+        oracle_client_dir=(preset["oracle_client_dir"] or None) if preset["db_type"] == "oracle" else None,
+    )
+
+    print("=" * 80)
+    print(f"DB      : {preset['display'] or preset['name']}（{preset['db_type']}）")
+    print(f"URI     : {preset['db_type']}://{preset['user']}@{preset['host']}:{preset['port']}/{preset['database']}")
+    print(f"关键字  : {args.keywords}")
+    print("=" * 80)
+
+    with DBConnection(cfg) as db:
+        # 1) 匹配
+        if preset["db_type"] == "oracle":
+            sql, params = _build_oracle_match_sql(args.keywords)
+            # 第一个参数替换为当前 user（owner 大写）
+            params = (preset["user"],) + tuple(f"%{kw.lower()}%" for kw in args.keywords)
+        elif preset["db_type"] == "dameng":
+            sql, params = _build_dameng_match_sql(args.keywords)
+            params = (preset["user"],) + tuple(f"%{kw.lower()}%" for kw in args.keywords)
+        elif preset["db_type"] == "mysql":
+            sql, params = _build_mysql_match_sql(args.keywords, preset["database"])
+        else:
+            raise SystemExit(f"[!] 暂不支持的 db_type: {preset['db_type']}")
+
+        print("\n[1/3] 扫字段...")
+        rows = db.fetchall(sql, params)
+        if not rows:
+            print("  ⚠ 没有匹配字段")
+            return
+
+        # 去重
+        seen = set()
+        unique = []
+        for r in rows:
+            k = (r["table_name"], r["column_name"])
+            if k in seen:
+                continue
+            seen.add(k)
+            unique.append(r)
+        rows = unique
+        print(f"  → 命中 {len(rows)} 个（表, 列）")
+
+        # 2) 行数
+        print("\n[2/3] 取每张表行数...")
+        results = []
+        for r in rows:
+            cnt = table_row_count(db, r["owner"], r["table_name"], preset["db_type"])
+            results.append({**r, "row_count": cnt})
+
+        # 3) 输出
+        print("\n" + "=" * 80)
+        print("结果汇总")
+        print("=" * 80)
+        print(f"{'表名':<32} {'行数':>14} {'列名':<24} {'类型':<14} {'可空':<4} {'表注释':<28}")
+        print("-" * 120)
+        from collections import defaultdict
+        by_table = defaultdict(list)
+        for r in results:
+            by_table[r["table_name"]].append(r)
+        for tbl, items in sorted(by_table.items()):
+            tc = items[0].get("table_comment") or ""
+            tc = str(tc)[:28]
+            rc = items[0]["row_count"]
+            rc_str = f"{rc:,}" if isinstance(rc, int) else str(rc)
+            for i, it in enumerate(items):
+                print(
+                    f"{tbl[:32]:<32} "
+                    f"{(rc_str if i == 0 else ''):>14} "
+                    f"{it['column_name'][:24]:<24} "
+                    f"{str(it.get('column_type') or it.get('data_type') or '')[:14]:<14} "
+                    f"{str(it.get('nullable') or ''):<4} {tc}"
+                )
+        total_rows = sum(r for r in [it["row_count"] for items in by_table.values() for it in items[:1]]
+                         if isinstance(r, int))
+        print("-" * 120)
+        print(f"匹配表数: {len(by_table)}   匹配列数: {len(results)}   行数总计: {total_rows:,}")
+
+        # 4) 视图溯源（Oracle / 达梦才有 USER_DEPENDENCIES）
+        if preset["db_type"] in ("oracle", "dameng"):
+            views = [r["table_name"] for r in results
+                     if r["table_name"].startswith(("V_", "V$"))]
+            if not views:
+                print("\n  （无视图命中，跳过依赖溯源）")
+                return
+            print("\n[3/3] 视图依赖溯源...")
+            # deps: dict[view_name -> list[{referenced_owner, referenced_name, referenced_type}]]
+            deps = trace_view_deps(db, views, preset["user"])
+            if not deps:
+                print("  （USER_DEPENDENCIES 0 行 —— 用户可能没权限 / 视图无依赖）")
+                return
+            for vname, refs in sorted(deps.items()):
+                base_tables  = [r for r in refs if (r["referenced_type"] or "").upper() == "TABLE"]
+                nested_views = [r for r in refs if (r["referenced_type"] or "").upper() == "VIEW"]
+                print(f"\n  VIEW  {vname}  →  引用：")
+                for r in base_tables:
+                    cnt = table_row_count(db, r["referenced_owner"],
+                                          r["referenced_name"], preset["db_type"])
+                    cnt_s = f"{cnt:,}" if isinstance(cnt, int) else str(cnt)
+                    print(f"    BASE TABLE  {r['referenced_owner']}.{r['referenced_name']:<30}  行数={cnt_s}")
+                for r in nested_views:
+                    print(f"    VIEW        {r['referenced_owner']}.{r['referenced_name']}  （嵌套）")
+
+
+if __name__ == "__main__":
+    main()

web3/tests/_probe_gjj80_bankcard.py

+"""
+一次性探测脚本：在 GJJ80 (ZFGJJ) Oracle 库中找含有「银行卡号」相关字段的表 + 行数。
+
+用法（在 web3/ 目录下）：
+    PYTHONPATH=. python tests/_probe_gjj80_bankcard.py
+
+字段匹配规则（覆盖常见命名）：
+    - 中文：银行卡号 / 银行卡 / 卡号 / 银行账号
+    - 英文：BANK_CARD_NO / BANKCARD / BANK_NO / BANK_ACCOUNT / CARD_NO
+    - 区分大小写都查（ALL_TAB_COLUMNS.COLUMN_NAME 默认大写；中文大小写无意义）
+"""
+from __future__ import annotations
+
+import logging
+import sys
+from pathlib import Path
+
+# 让 import "web3.backend.xxx" 生效 —— parents[2] 是项目根，
+# 根下有 web3/ 子包。
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from web3.backend.core.db_adapter import DBConfig, DBConnection
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+
+CFG = DBConfig(
+    db_type="oracle",
+    host="192.168.20.196",
+    port=1521,
+    user="GJJ80",
+    password="GJJ80_Hg41RG",
+    database="ZFGJJ",  # Oracle 里是 service_name
+    # 本机 PL/SQL Developer 自带的 Instant Client 23.0；thin 模式不支持老 Oracle，必须切 thick
+    oracle_client_dir=r"C:\Program Files\PLSQL Developer 16\instantclient_23_0",
+)
+
+# 关键字列表 —— LIKE 大小写不敏感（Oracle NLS 默认区分，这里显式 LOWER 兼容）
+# %不要前缀，避免全表扫；先全库扫一次 cost 还行（ALL_TAB_COLUMNS 不大）
+KEYWORDS = [
+    "银行卡号",
+    "银行卡",
+    "卡号",
+    "银行账号",
+    "银行账户",
+    "账号",
+    "帐号",
+    "bank_card_no",
+    "bankcard",
+    "bank_no",
+    "bank_account",
+    "card_no",
+    "cardnum",
+    "cardaccount",
+]
+
+SQL_MATCH = """
+SELECT
+    c.OWNER,
+    c.TABLE_NAME,
+    c.COLUMN_NAME,
+    c.DATA_TYPE,
+    c.DATA_LENGTH,
+    c.NULLABLE,
+    cc.COMMENTS AS column_comment,
+    (
+        SELECT tc.COMMENTS
+        FROM ALL_TAB_COMMENTS tc
+        WHERE tc.OWNER = c.OWNER AND tc.TABLE_NAME = c.TABLE_NAME
+    ) AS table_comment
+FROM ALL_TAB_COLUMNS c
+LEFT JOIN ALL_COL_COMMENTS cc
+    ON  cc.OWNER      = c.OWNER
+    AND cc.TABLE_NAME = c.TABLE_NAME
+    AND cc.COLUMN_NAME = c.COLUMN_NAME
+WHERE c.OWNER = UPPER(:1)
+  AND (
+""" + " OR ".join(
+    [
+        "LOWER(c.COLUMN_NAME) LIKE :%d" % (i + 2)
+        for i, kw in enumerate(KEYWORDS)
+    ]
+) + """
+  )
+ORDER BY c.TABLE_NAME, c.COLUMN_ID
+"""
+
+# 构造 LIKE 串，% 在 bind 里传入；每个关键字一个小写
+LIKE_PARAMS = ["%" + kw.lower() + "%" for kw in KEYWORDS]
+PARAMS = (CFG.user,) + tuple(LIKE_PARAMS)
+
+
+def main():
+    with DBConnection(CFG) as db:
+        print("=" * 80)
+        print(f"连接: {CFG.user}@{CFG.host}:{CFG.port}/{CFG.database}")
+        print(f"匹配关键字: {KEYWORDS}")
+        print("=" * 80)
+
+        # 1) 命中字段
+        print("\n[1/2] 扫 ALL_TAB_COLUMNS，找匹配的字段...")
+        rows = db.fetchall(SQL_MATCH, PARAMS)
+        if not rows:
+            print("⚠ 没有匹配字段")
+            return
+
+        # 去重（ALL_COL_COMMENTS 多份可能导致重复行，先按表+列 dedup）
+        # 注意：db_adapter 归一化列名为小写，所以用 lowercase key
+        seen = set()
+        unique_rows = []
+        for r in rows:
+            key = (r["table_name"], r["column_name"])
+            if key in seen:
+                continue
+            seen.add(key)
+            unique_rows.append(r)
+        rows = unique_rows
+        print(f"  → 命中 {len(rows)} 个（表, 列）")
+
+        # 2) 批量取行数 —— 用 owner.table 拼 COUNT(*)
+        # Oracle 里可以双引号包识别符，但 owner.table 是数据字典里的标准写法
+        print("\n[2/2] 对每张表跑 COUNT(*)...")
+        results = []
+        for r in rows:
+            owner = r["owner"]
+            tbl = r["table_name"]
+            ident = f'{owner}."{tbl}"'  # 表名用双引号兜底关键字
+            count_sql = f"SELECT COUNT(*) AS cnt FROM {ident}"
+            try:
+                cnt_row = db.fetchone(count_sql)
+                cnt = cnt_row["cnt"] if cnt_row else "?"
+            except Exception as e:
+                cnt = f"ERR: {e.__class__.__name__}: {str(e)[:80]}"
+            results.append({**r, "row_count": cnt})
+
+        # 3) 汇总输出
+        print("\n" + "=" * 80)
+        print("结果汇总")
+        print("=" * 80)
+        # 按 (table_name, column_name) 排序，按表聚合
+        from collections import defaultdict
+
+        by_table = defaultdict(list)
+        for r in results:
+            by_table[r["table_name"]].append(r)
+
+        # 列宽
+        print(f"{'表名':<35} {'行数':>12} {'列名':<25} {'类型':<12} {'可空':<4} {'表注释':<30}")
+        print("-" * 130)
+        total_rows_sum = 0
+        for tbl, items in sorted(by_table.items()):
+            tbl_comment = items[0].get("table_comment") or ""
+            tbl_comment = str(tbl_comment)[:30]
+            # 行数只显示一次（如果多个列命中同一表，行数展示第一行的即可）
+            row_count = items[0]["row_count"]
+            if isinstance(row_count, int):
+                total_rows_sum += row_count
+            for i, it in enumerate(items):
+                col_name = it["column_name"][:25]
+                col_type = (it["data_type"] or "")[:12]
+                nullable = it["nullable"] or ""
+                rc_str = f"{row_count:,}" if isinstance(row_count, int) else str(row_count)
+                print(
+                    f"{tbl[:35]:<35} {rc_str if i == 0 else '':>12} "
+                    f"{col_name:<25} {col_type:<12} {nullable:<4} {tbl_comment}"
+                )
+        print("-" * 130)
+        print(f"匹配表数: {len(by_table)}   匹配列数: {len(results)}   行数总计: {total_rows_sum:,}")
+
+        # 4) 详细列表（包含列注释，方便人工确认是不是真「银行卡号」）
+        print("\n" + "=" * 80)
+        print("详细列表（含列注释）")
+        print("=" * 80)
+        for r in results:
+            cc = r.get("column_comment") or ""
+            cc = str(cc)[:50]
+            rc = r["row_count"]
+            rc_str = f"{rc:,}" if isinstance(rc, int) else str(rc)
+            print(
+                f"  {r['owner']}.{r['table_name']}.{r['column_name']:<25} "
+                f"类型={r['data_type']:<10} 行数={rc_str:>14}  列注释={cc}"
+            )
+
+        # 5) 视图依赖溯源：V_* 通常是 view，背后是 T_* / E_* 等基表
+        #    先试 USER_DEPENDENCIES（GJJ80 用户视角，权限要求低一些）
+        #    再试 ALL_DEPENDENCIES（如果上面 0 行）
+        print("\n" + "=" * 80)
+        print("视图依赖溯源（V_* 引用的基表）")
+        print("=" * 80)
+        view_names = list(by_table.keys())
+        if not view_names:
+            print("  （无视图命中）")
+            return
+
+        # 先看 V_* 视图下是不是真有基表 T_JCFX_* / E_JCFX_*
+        # GJJ80 用户在自己 schema 下找（OWNER=UPPER(GJJ80) AND TYPE='TABLE' AND NAME LIKE 'JCFX%'）
+        guess_sql = """
+            SELECT OWNER, OBJECT_TYPE, OBJECT_NAME, STATUS
+            FROM ALL_OBJECTS
+            WHERE OWNER = UPPER(:1)
+              AND OBJECT_TYPE = 'TABLE'
+              AND (OBJECT_NAME LIKE '%JCFX%' OR OBJECT_NAME LIKE '%GR%' OR OBJECT_NAME LIKE '%DW%')
+            ORDER BY OBJECT_NAME
+        """
+        guess_rows = db.fetchall(guess_sql, (CFG.user,))
+        if guess_rows:
+            print(f"\n  GJJ80 用户下所有 JCFX/GR/DW 命名的基表（共 {len(guess_rows)} 张）：")
+            for r in guess_rows:
+                print(f"    TABLE  {r['owner']}.{r['object_name']:<35}  状态={r['status']}")
+        else:
+            print("  （没找到 JCFX/GR/DW 命名的基表）")
+
+        # 真正的 view → table 依赖（用 USER_DEPENDENCIES + ALL_DEPENDENCIES 两种试）
+        for src in ("USER_DEPENDENCIES", "ALL_DEPENDENCIES"):
+            placeholders = ",".join([f":{i+1}" for i in range(len(view_names))])
+            dep_sql = f"""
+                SELECT
+                    d.NAME            AS dependent_name,
+                    d.REFERENCED_OWNER,
+                    d.REFERENCED_NAME,
+                    d.REFERENCED_TYPE
+                FROM {src} d
+                WHERE d.NAME IN ({placeholders})
+                  AND d.REFERENCED_TYPE IN ('TABLE', 'VIEW')
+                ORDER BY d.NAME, d.REFERENCED_NAME
+            """
+            try:
+                dep_rows = db.fetchall(dep_sql, tuple(view_names))
+            except Exception as e:
+                print(f"\n  {src} 查询失败：{e.__class__.__name__}: {str(e)[:120]}")
+                continue
+            if not dep_rows:
+                print(f"\n  {src} 0 行（可能没权限 / 视图无依赖记录）")
+                continue
+            print(f"\n  {src} 命中 {len(dep_rows)} 行：")
+            from collections import defaultdict
+            dep_by_view = defaultdict(list)
+            for d in dep_rows:
+                dep_by_view[d["dependent_name"]].append(d)
+            for vname in view_names:
+                refs = dep_by_view.get(vname, [])
+                if not refs:
+                    continue
+                base_tables = [r for r in refs if (r["referenced_type"] or "").upper() == "TABLE"]
+                nested_views = [r for r in refs if (r["referenced_type"] or "").upper() == "VIEW"]
+                print(f"\n  VIEW  {vname}  →  引用：")
+                for r in base_tables:
+                    ro = r["referenced_owner"]
+                    rn = r["referenced_name"]
+                    # 顺手把基表行数也打一下
+                    try:
+                        rcnt = db.fetchone(f'SELECT COUNT(*) AS cnt FROM {ro}."{rn}"')["cnt"]
+                        rcnt_s = f"{rcnt:,}" if rcnt is not None else "(空)"
+                    except Exception as e:
+                        rcnt_s = f"ERR:{str(e)[:40]}"
+                    print(f"    BASE TABLE  {ro}.{rn:<35}  行数={rcnt_s}")
+                for r in nested_views:
+                    ro = r["referenced_owner"]
+                    rn = r["referenced_name"]
+                    print(f"    VIEW        {ro}.{rn}  （嵌套视图）")
+            break  # 一种查到就退出
+
+
+if __name__ == "__main__":
+    main()