Skip to content

D
db-tools
Commit 31911f51 authored by Data Governance Dev's avatar Data Governance Dev
Browse files
Options

feat(data-dict): Phase 1 - 单文件脚本完成 smart-build 数据字典与首轮分析 

第一阶段，针对 smart-build 数据库做一次性手工治理：
- fetch_data_dictionary.py：拉取 information_schema，输出 data_dictionary.json
- check_empty_fields.py：逐表扫 NULL/空字符串比例，输出高空字段报告
- verify_merge_redundancy.py + .sql：连库验证合并候选的实际行数
- validate_standard_fields.py：对身份证/手机/统一社会信用代码等抽样校验
- check_field_length.py：识别 VARCHAR 超出固定长度的字段（基于国标）
- check_uncommented_fields.py：扫描无注释字段，按字段名启发式推测
- generate_report.py：把以上 JSON 汇总成 Word 报告

Why：先把\"用什么工具看什么数据\"跑通，再抽象成可复用工作流。
How：每个脚本独立运行，输出落 JSON 到 data_dictionary/，最终由
     generate_report.py 汇总成 数据治理报告_smart-build.docx。
parent 20dc9503
Show whitespace changes
Inline Side-by-side
Showing with 3702 additions and 0 deletions
data_dictionary/README.md 0 → 100644
View file @ 31911f51
# 数据字典获取工具
## 用途
连接 `smart-build` 数据库（MySQL），自动提取全部表和字段的元数据信息，输出为 JSON 文件供后续数据治理分析使用。
## 依赖
- Python 3.7+
- pymysql
```bash
pip install pymysql
```
## 使用方法
```bash
cd data_dictionary
python fetch_data_dictionary.py
```
执行时会提示输入数据库密码（输入不显示，按 Enter 确认）。
## 输出文件
执行成功后在当前目录生成两个 JSON 文件：
| 文件 | 内容 |
|------|------|
| `data_dictionary.json` | 所有表的所有字段详情 |
| `table_summary.json` | 表级别汇总信息 |
### data_dictionary.json 字段说明
| 字段 | 含义 |
|------|------|
| `table_name` | 表名 |
| `column_name` | 字段名 |
| `ordinal_position` | 字段序号（在表中的位置） |
| `column_type` | 完整列类型（如 `varchar(50)`） |
| `data_type` | 数据类型（如 `varchar`） |
| `char_max_length` | 字符最大长度 |
| `numeric_precision` | 数值精度 |
| `numeric_scale` | 小数位数 |
| `is_nullable` | 是否可为空（YES/NO） |
| `column_default` | 默认值 |
| `column_comment` | 字段注释 |
| `extra` | 额外信息（如 `auto_increment`） |
| `table_comment` | 所属表的注释 |
### table_summary.json 字段说明
| 字段 | 含义 |
|------|------|
| `TABLE_NAME` | 表名 |
| `TABLE_TYPE` | 表类型（BASE TABLE / VIEW） |
| `ENGINE` | 存储引擎 |
| `TABLE_ROWS` | 估计行数 |
| `DATA_LENGTH` | 数据大小（字节） |
| `INDEX_LENGTH` | 索引大小（字节） |
| `TABLE_COMMENT` | 表注释 |
| `CREATE_TIME` | 创建时间 |
| `UPDATE_TIME` | 最后更新时间 |
## 数据库连接信息
| 项目 | 值 |
|------|-----|
| 主机 | 192.168.20.10 |
| 端口 | 3306 |
| 用户 | root |
| 数据库 | smart-build |
> 连接信息硬编码在脚本的 `DB_CONFIG` 字典中，如需修改请编辑脚本。
---
## 操作记录
### Step 1: 获取数据字典 ✅ (2026-08-03)
执行 `fetch_data_dictionary.py`，获取 smart-build 库全部 149 张表、3148 个字段的元数据。
### Step 2: 表合并与冗余字段分析 ✅ (2026-08-03)
基于数据字典分析发现：
**合并候选（3组）：**
| ID | 表 | 相似度 | 建议 |
|----|-----|--------|------|
| MERGE-001 | `em_gas` + `em_water_supply` | 100% | 24字段完全一致，合并为 `em_infrastructure` 表，加 `type` 字段区分 |
| MERGE-002 | `t_project_*_record` ×5 | 100% | 31字段完全一致，合并为 `t_project_business_record`，加 `record_type` 字段区分 |
| MERGE-003 | `biz_history_field_label` + `biz_history_focus_field` | 83% | 高度相似，考虑合并 |
**废弃候选：**
- **强候选**：19张旧版 `t_*` 人才系统表（全部行数=0，新版 `t_talent_*` 已替代）
- **弱候选**：12张行数为0且无替代的表（mall、project、ai 等模块）
**冗余字段（Top 5）：**
| 字段 | 出现表数 | 风险 |
|------|----------|------|
| `project_name` | 30 | 高 — 应通过 project_id 关联 |
| `enterprise_name` | 13 | 高 — 应通过 enterprise_id 关联 |
| `site_code` + `site_name` | 10 | 中 — 成对冗余 |
| `xzqhbm` | 20 | 中 — 已有 c_bri_xzqh 字典表 |
| `user_id` / `talent_id` | 14 / 15 | 中 — 两套ID体系并存 |
详细分析结果见 `findings_table_merge_redundancy.json`。
### Step 3: 数据验证 ✅ (2026-08-03)
执行 `verify_merge_redundancy.py`，11项检查全部通过。原始数据见 `verify_results.json`，最终结论见 `findings_verified.json`。
**确认合并（3组）：**
| ID | 表 | 数据量 | 结论 |
|----|-----|--------|------|
| MERGE-001 | `em_gas` + `em_water_supply` | 8行 + 7行 | ✅ 确认合并，加 `infrastructure_type` 字段 |
| MERGE-002 | `t_project_*_record` ×5 | 合计7行 | ✅ 确认合并，加 `record_type` 字段 |
| MERGE-003 | `biz_history_field_label` + `biz_history_focus_field` | 38行 + 13行 | ✅ 建议评估后合并 |
**确认废弃：**
- **强候选**：19张旧版 `t_*` 表全部0行 ✅，新版 `t_talent_*` 正常运行
- **弱候选**：11张0行表（`t_ai_llm_provider` 实际有1行，排除）
**确认冗余字段（5个）：**
| 字段 | 表数 | 关键发现 |
|------|------|----------|
| `project_name` | 30 | 13张表同时存 project_id+project_name，冗余维护 |
| `enterprise_name` | 13 | `t_talent_archive`(37155行) 填充率仅1.7% |
| `site_code/site_name` | 10 | **数据已不一致**：同code不同name |
| `xzqhbm` | 20 | 发现孤儿编码；字符集不匹配致部分表无法JOIN |
| `user_id/talent_id` | 14/15 | 两套ID并存；`t_talent_evaluation`/`t_talent_offer` 的 talent_id 填充率0% |
**数据质量问题（3个）：**
| ID | 问题 | 严重程度 |
|----|------|----------|
| DQ-001 | em_*/ex_* 与 c_bri_xzqh 字符集不一致(unicode_ci vs general_ci) | 中 |
| DQ-002 | `c_bri_geo_boundary` 20字段 + 表注释全部为空（1804行数据） | 高 |
| DQ-003 | INFORMATION_SCHEMA.TABLE_ROWS 估算不准（0行实际有数据） | 低 |
**下一步：** 进入 Step 4 — 字段规范检查（身份证号、手机号、统一社会信用代码、地区编码等）。
data_dictionary/check_empty_fields.py 0 → 100644
View file @ 31911f51
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据治理 - 大范围空字段检查脚本
功能：逐表扫描所有字段，统计 NULL 率和空字符串率，
识别出填充率低的字段（大范围空的字段）。
输出文件: empty_fields_report.json
"""
import json
import os
import sys
from datetime import datetime
from getpass import getpass
try:
import pymysql
except ImportError:
print("错误：缺少 pymysql 库，请执行: pip install pymysql")
sys.exit(1)
# ── 配置 ──────────────────────────────────────────────
DB_CONFIG = {
"host": "192.168.20.10",
"port": 3306,
"user": "root",
"database": "smart-build",
"charset": "utf8mb4",
}
OUTPUT_FILE = "empty_fields_report.json"
# 阈值：NULL+空字符串占比超过此值视为"大范围空"
HIGH_EMPTY_THRESHOLD = 0.8 # >=80% 标记为"高空"
MID_EMPTY_THRESHOLD = 0.5 # >=50% 标记为"中空"
# 行数少于该值的表跳过（样本太小没意义）
MIN_ROWS_TO_CHECK = 5
# ── 获取密码 ──────────────────────────────────────────
def get_db_password():
print("=" * 50)
print(f" 连接目标: {DB_CONFIG['user']}@{DB_CONFIG['host']}:{DB_CONFIG['port']}")
print(f" 数据库: {DB_CONFIG['database']}")
print("=" * 50)
password = getpass("请输入数据库密码（输入不显示）: ")
if not password:
print("错误：密码不能为空")
sys.exit(1)
return password
# ── 连接数据库 ────────────────────────────────────────
def connect_db(password):
try:
conn = pymysql.connect(
host=DB_CONFIG["host"],
port=DB_CONFIG["port"],
user=DB_CONFIG["user"],
password=password,
database=DB_CONFIG["database"],
charset=DB_CONFIG["charset"],
connect_timeout=10,
)
print("✓ 数据库连接成功\n")
return conn
except pymysql.err.OperationalError as e:
print(f"错误：无法连接数据库 — {e}")
sys.exit(1)
# ── 辅助函数 ──────────────────────────────────────────
def query_all(cursor, sql, params=None):
cursor.execute(sql, params)
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
return [dict(zip(columns, row)) for row in rows]
def query_single(cursor, sql, params=None):
cursor.execute(sql, params)
row = cursor.fetchone()
return row[0] if row else None
def default_serializer(obj):
if hasattr(obj, "isoformat"):
return obj.isoformat()
if isinstance(obj, bytes):
return obj.decode("utf-8", errors="replace")
return str(obj)
# ── 获取所有表的列信息 ────────────────────────────────
def get_all_columns(cursor):
"""获取所有表的列信息（表名、列名、数据类型、行数估算）"""
print("正在获取表结构信息...")
return query_all(
cursor,
"""
SELECT
c.TABLE_NAME,
c.COLUMN_NAME,
c.DATA_TYPE,
c.CHARACTER_MAXIMUM_LENGTH,
c.IS_NULLABLE,
c.COLUMN_COMMENT,
c.ORDINAL_POSITION,
t.TABLE_ROWS,
t.TABLE_COMMENT
FROM INFORMATION_SCHEMA.COLUMNS c
JOIN INFORMATION_SCHEMA.TABLES t
ON c.TABLE_SCHEMA = t.TABLE_SCHEMA
AND c.TABLE_NAME = t.TABLE_NAME
WHERE c.TABLE_SCHEMA = %s
AND t.TABLE_TYPE = 'BASE TABLE'
ORDER BY c.TABLE_NAME, c.ORDINAL_POSITION
""",
(DB_CONFIG["database"],),
)
# ── 构建单表的批量空值检查SQL ─────────────────────────
def build_table_check_sql(table_name, columns):
"""
为一整张表构建一条 SQL，同时统计所有列的空值情况。
返回 SQL 字符串和列名顺序列表。
"""
exprs = []
col_names = []
for col in columns:
cname = col["COLUMN_NAME"]
dtype = col["DATA_TYPE"].lower()
col_names.append(cname)
# 对于字符串类型，检查 NULL 和空字符串''
if dtype in ("varchar", "char", "text", "longtext", "mediumtext",
"tinytext", "enum", "set"):
exprs.append(
f"SUM(CASE WHEN `{cname}` IS NULL OR `{cname}` = '' THEN 1 ELSE 0 END) AS `{cname}__empty`"
)
else:
# 数值/日期/其他类型，仅检查 NULL
exprs.append(
f"SUM(CASE WHEN `{cname}` IS NULL THEN 1 ELSE 0 END) AS `{cname}__null`"
)
sql = f"SELECT COUNT(*) AS __total__, {', '.join(exprs)} FROM `{table_name}`"
return sql, col_names
# ── 主扫描逻辑 ────────────────────────────────────────
def scan_all_tables(cursor):
"""扫描所有表，返回每个字段的空值率"""
columns_info = get_all_columns(cursor)
# 按表分组
tables = {}
for col in columns_info:
tname = col["TABLE_NAME"]
if tname not in tables:
tables[tname] = {
"table_comment": col["TABLE_COMMENT"],
"estimated_rows": col["TABLE_ROWS"],
"columns": [],
}
tables[tname]["columns"].append(col)
total_tables = len(tables)
print(f"共 {total_tables} 张表待扫描\n")
all_results = [] # 所有字段的空值率
skipped_tables = [] # 因行数太少跳过的表
empty_field_summary = [] # 高/中空字段汇总
for idx, (tname, tinfo) in enumerate(tables.items(), 1):
estimated = tinfo["estimated_rows"]
cols = tinfo["columns"]
# 跳过行数太少的表
if estimated < MIN_ROWS_TO_CHECK:
skipped_tables.append({
"table_name": tname,
"estimated_rows": estimated,
"reason": f"估算行数 < {MIN_ROWS_TO_CHECK}，样本太小跳过",
})
continue
# 获取实际行数
actual_rows = query_single(cursor, f"SELECT COUNT(*) FROM `{tname}`")
if actual_rows is None or actual_rows < MIN_ROWS_TO_CHECK:
skipped_tables.append({
"table_name": tname,
"estimated_rows": estimated,
"actual_rows": actual_rows,
"reason": f"实际行数 < {MIN_ROWS_TO_CHECK}，样本太小跳过",
})
continue
# 构建并执行批量检查SQL
sql, col_order = build_table_check_sql(tname, cols)
try:
row = query_all(cursor, sql)
if not row:
continue
stats = row[0]
total = stats["__total__"]
if total == 0:
skipped_tables.append({
"table_name": tname,
"actual_rows": 0,
"reason": "实际行数为0",
})
continue
# 解析每个字段的空值率
table_empty_fields = []
for cname in col_order:
empty_key = f"{cname}__empty"
null_key = f"{cname}__null"
if empty_key in stats:
empty_count = stats[empty_key] or 0
label = "NULL或空字符串"
elif null_key in stats:
empty_count = stats[null_key] or 0
label = "NULL"
else:
continue
rate = round(empty_count / total, 4) if total > 0 else 0
# 找对应的列信息
col_info = next((c for c in cols if c["COLUMN_NAME"] == cname), {})
field_result = {
"table_name": tname,
"table_comment": tinfo["table_comment"],
"column_name": cname,
"data_type": col_info.get("DATA_TYPE", ""),
"char_max_length": col_info.get("CHARACTER_MAXIMUM_LENGTH"),
"is_nullable": col_info.get("IS_NULLABLE", ""),
"column_comment": col_info.get("COLUMN_COMMENT", ""),
"total_rows": total,
"empty_count": empty_count,
"empty_rate": rate,
"empty_type": label,
"level": "高空" if rate >= HIGH_EMPTY_THRESHOLD else
("中空" if rate >= MID_EMPTY_THRESHOLD else "正常"),
}
all_results.append(field_result)
if rate >= MID_EMPTY_THRESHOLD:
table_empty_fields.append(field_result)
if table_empty_fields:
empty_field_summary.append({
"table_name": tname,
"total_rows": total,
"table_comment": tinfo["table_comment"],
"high_empty_fields": len([f for f in table_empty_fields if f["level"] == "高空"]),
"mid_empty_fields": len([f for f in table_empty_fields if f["level"] == "中空"]),
})
except Exception as e:
print(f" ⚠ {tname} 扫描失败: {e}")
continue
# 进度
if idx % 10 == 0 or idx == total_tables:
print(f" 进度: {idx}/{total_tables} 张表已扫描")
return {
"all_fields": all_results,
"empty_field_summary": empty_field_summary,
"skipped_tables": skipped_tables,
}
# ── 汇总统计 ──────────────────────────────────────────
def build_summary(all_fields, empty_field_summary, skipped_tables):
"""生成汇总统计"""
high_empty = [f for f in all_fields if f["level"] == "高空"]
mid_empty = [f for f in all_fields if f["level"] == "中空"]
normal = [f for f in all_fields if f["level"] == "正常"]
# 按表统计高空字段数（Top 20）
high_by_table = {}
for f in high_empty:
tname = f["table_name"]
high_by_table[tname] = high_by_table.get(tname, 0) + 1
top_high_tables = sorted(high_by_table.items(), key=lambda x: x[1], reverse=True)[:20]
# 按表统计中空字段数（Top 20）
mid_by_table = {}
for f in mid_empty:
tname = f["table_name"]
mid_by_table[tname] = mid_by_table.get(tname, 0) + 1
top_mid_tables = sorted(mid_by_table.items(), key=lambda x: x[1], reverse=True)[:20]
return {
"total_tables_scanned": len(empty_field_summary) + len(skipped_tables),
"total_fields_checked": len(all_fields),
"high_empty_fields_count": len(high_empty),
"mid_empty_fields_count": len(mid_empty),
"normal_fields_count": len(normal),
"skipped_tables_count": len(skipped_tables),
"tables_with_issues_count": len(empty_field_summary),
"thresholds": {
"high_empty": f">={int(HIGH_EMPTY_THRESHOLD * 100)}%",
"mid_empty": f">={int(MID_EMPTY_THRESHOLD * 100)}%",
"min_rows_to_check": MIN_ROWS_TO_CHECK,
},
"top_tables_by_high_empty_fields": [
{"table_name": t, "high_empty_field_count": c} for t, c in top_high_tables
],
"top_tables_by_mid_empty_fields": [
{"table_name": t, "mid_empty_field_count": c} for t, c in top_mid_tables
],
"high_empty_fields": [
{
"table_name": f["table_name"],
"column_name": f["column_name"],
"data_type": f["data_type"],
"column_comment": f["column_comment"],
"total_rows": f["total_rows"],
"empty_count": f["empty_count"],
"empty_rate": f["empty_rate"],
}
for f in high_empty
],
}
# ── 保存 ──────────────────────────────────────────────
def save_json(data, filepath):
script_dir = os.path.dirname(os.path.abspath(__file__))
full_path = os.path.join(script_dir, filepath)
with open(full_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2, default=default_serializer)
print(f"\n✓ 结果已保存至: {full_path}")
# ── 主流程 ────────────────────────────────────────────
def main():
password = get_db_password()
conn = connect_db(password)
try:
with conn.cursor() as cursor:
print("开始扫描大范围空字段...\n")
scan_result = scan_all_tables(cursor)
summary = build_summary(
scan_result["all_fields"],
scan_result["empty_field_summary"],
scan_result["skipped_tables"],
)
output = {
"meta": {
"script": "check_empty_fields.py",
"database": DB_CONFIG["database"],
"host": DB_CONFIG["host"],
"executed_at": datetime.now().isoformat(),
},
"summary": summary,
"empty_field_summary_by_table": scan_result["empty_field_summary"],
"skipped_tables": scan_result["skipped_tables"],
}
save_json(output, OUTPUT_FILE)
# ── 控制台输出 ──
s = summary
print(f"\n{'=' * 50}")
print(f" 扫描完成")
print(f"{'=' * 50}")
print(f" 已扫描表数: {s['total_tables_scanned']}")
print(f" 已检查字段总数: {s['total_fields_checked']}")
print(f" 高空字段 (≥80%空): {s['high_empty_fields_count']}")
print(f" 中空字段 (≥50%空): {s['mid_empty_fields_count']}")
print(f" 正常字段: {s['normal_fields_count']}")
print(f" 跳过表数: {s['skipped_tables_count']} (行数<{MIN_ROWS_TO_CHECK})")
print(f" 有问题的表数: {s['tables_with_issues_count']}")
print(f"\n 📊 高空字段最多的表 Top 10:")
for i, t in enumerate(s['top_tables_by_high_empty_fields'][:10], 1):
print(f" {i}. {t['table_name']}: {t['high_empty_field_count']} 个高空字段")
finally:
conn.close()
print(f"\n数据库连接已关闭。")
if __name__ == "__main__":
main()
data_dictionary/check_field_length.py 0 → 100644
View file @ 31911f51
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据治理 - 字段长度超标检查脚本
功能：根据已知国家标准/规范的固定长度字段，检查数据字典中
哪些字段的存储长度超出了实际所需长度。
规则来源：national_standards/field_standards_reference.json
数据来源：data_dictionary.json
输出文件: field_length_violations.json
"""
import json
import os
import re
import sys
# ── 配置 ──────────────────────────────────────────────
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DICT_FILE = os.path.join(SCRIPT_DIR, "data_dictionary.json")
STANDARDS_FILE = os.path.join(SCRIPT_DIR, "..", "national_standards", "field_standards_reference.json")
OUTPUT_FILE = os.path.join(SCRIPT_DIR, "field_length_violations.json")
# ── 固定长度字段规范 ──────────────────────────────────
# 格式: 字段名匹配模式 → (所需长度, 所需类型, 标准名称, 备注)
FIXED_LENGTH_RULES = {
# ── 人员身份类 ──
"id_card": (18, "char", "GB 11643-1999", "身份证号码，固定18位"),
"id_card_no": (18, "char", "GB 11643-1999", "身份证号码，固定18位"),
# ── 通信类 ──
"mobile": (11, "char", "工信部号码规划", "手机号码，固定11位"),
"phone": (11, "char", "工信部号码规划", "手机号11位；如存固话则需区号+号码，最多13位"),
"contact_phone": (11, "char", "工信部号码规划", "联系电话，手机号11位"),
"customer_mobile": (11, "char", "工信部号码规划", "客户手机号，固定11位"),
"legal_person_mobile":(11, "char", "工信部号码规划", "法人手机号，固定11位"),
"notify_phone": (11, "char", "工信部号码规划", "通知手机号，固定11位"),
"maintainer_phone": (11, "char", "工信部号码规划", "维护人手机号，固定11位"),
"manager_phone": (11, "char", "工信部号码规划", "负责人手机号，固定11位"),
"receiver_phone": (11, "char", "工信部号码规划", "收件人手机号，固定11位"),
# ── 企业/组织类 ──
"uscc": (18, "char", "GB 32100-2015", "统一社会信用代码，固定18位"),
"credit_code": (18, "char", "GB 32100-2015", "统一社会信用代码，固定18位"),
"business_license_no":(18, "char", "GB 32100-2015", "营业执照号/统一社会信用代码"),
# ── 地理/行政类 ──
"xzqhbm": (6, "char", "GB/T 2260", "行政区划代码，标准6位"),
"xzqhjb": (6, "char", "GB/T 2260", "行政区划代码（上级），标准6位"),
"sjxzqhbm": (6, "char", "GB/T 2260", "上级行政区划代码，标准6位"),
"adcode": (6, "char", "GB/T 2260", "行政区划代码，标准6位（高德扩展12位）"),
"district_code": (6, "char", "GB/T 2260", "区县代码，标准6位"),
"province_code": (2, "char", "GB/T 2260", "省级代码，标准2位"),
"city_code": (4, "char", "GB/T 2260", "城市代码，标准4位（前6位的一部分）"),
"region_code": (6, "char", "GB/T 2260", "区域代码，标准6位"),
"postal_code": (6, "char", "GB/T 2260", "邮政编码，固定6位"),
# ── 国际代码类 ──
"country_iso3": (3, "char", "ISO 3166-1", "国家ISO3代码，固定3位字母"),
"country_id": (3, "char", "ISO 3166-1", "国家数字代码，3位数字"),
"currency": (3, "char", "ISO 4217", "货币字母代码，固定3位字母"),
# ── 人员属性代码类 ──
"gender": (1, "char", "GB/T 2261.1", "性别代码，1位数字(0-2)"),
"sex": (1, "char", "GB/T 2261.1", "性别代码，1位数字"),
"nation": (2, "char", "GB/T 3304", "民族代码，2位数字(01-56)"),
"ethnicity": (2, "char", "GB/T 3304", "民族代码，2位数字"),
"education": (2, "char", "GB/T 4658", "学历代码，2位数字"),
"education_level": (2, "char", "GB/T 4658", "学历代码，2位数字"),
"educate_level": (2, "char", "GB/T 4658", "学历代码，2位数字"),
"political_status": (2, "char", "GB/T 4762", "政治面貌代码，2位数字"),
"marital_status": (1, "char", "GB/T 2261.2", "婚姻状况代码，1位数字"),
# ── 日期类 - 应该是 DATE 类型 ──
"year": (4, "int", "GB/T 7408", "年份，4位数字"),
# ── 语言类 ──
"language": (2, "char", "ISO 639-1", "语言代码，2位字母(alpha-2)"),
# ── 经纬度 - 应该是 DOUBLE ──
"lng": (0, "double", "ISO 6709", "经度，应使用DOUBLE而非VARCHAR"),
"longitude": (0, "double", "ISO 6709", "经度，应使用DOUBLE而非VARCHAR"),
"lat": (0, "double", "ISO 6709", "纬度，应使用DOUBLE而非VARCHAR"),
"latitude": (0, "double", "ISO 6709", "纬度，应使用DOUBLE而非VARCHAR"),
"centroid_lng": (0, "double", "ISO 6709", "中心点经度，应使用DOUBLE"),
"centroid_lat": (0, "double", "ISO 6709", "中心点纬度，应使用DOUBLE"),
}
# 如果存储的是 VARCHAR/CHAR 而非 DATE/DATETIME，也属于违规
DATE_FIELD_PATTERNS = [
r".*_date$", r".*_time$", r"^date_.*",
"birth_date", "date_of_birth", "established_date",
"create_time", "update_time", "issue_date", "expire_date",
"start_date", "end_date", "start_time", "end_time",
]
# 如果存储的是 VARCHAR 而非 INT/BIGINT，也提示
INT_FIELD_PATTERNS = [
"id", "user_id", "talent_id", "enterprise_id", "project_id",
"order_id", "shop_id", "course_id", "exam_id", "job_id",
"dept_id", "org_id", "customer_id", "institution_id",
"template_id", "policy_id", "reserve_id", "config_id",
]
def load_json(filepath):
with open(filepath, "r", encoding="utf-8") as f:
return json.load(f)
def match_field(column_name, rule_key):
"""精确匹配或前缀/后缀匹配"""
c = column_name.lower().strip()
r = rule_key.lower().strip()
if c == r:
return True
# 允许前后缀匹配：如 receiver_phone 匹配 phone 规则
# 但需要避免过度匹配：phone 不应匹配 phone_number_suffix
# 策略：规则key是字段名的结尾部分，或字段名是规则的超集
if c.endswith("_" + r) or c.startswith(r + "_"):
return True
return False
def check_type_mismatch(data_type, required_type):
"""检查类型是否不匹配"""
dtype = data_type.lower().strip()
rtype = required_type.lower().strip()
if rtype == "char":
# 要求的char类型，目前是varchar/char就对（后面再检查长度）
return dtype in ("varchar", "char", "text", "longtext", "mediumtext", "tinytext")
elif rtype == "double":
# 应该是数值型
return dtype in ("double", "float", "decimal", "real")
elif rtype == "int":
return dtype in ("int", "bigint", "smallint", "tinyint", "mediumint", "integer")
return True
def check_date_type_mismatch(column_name, data_type):
"""日期字段是否使用了非日期类型"""
dtype = data_type.lower().strip()
if dtype in ("date", "datetime", "timestamp", "time", "year"):
return None # 正确
for pat in DATE_FIELD_PATTERNS:
if re.match(pat, column_name, re.IGNORECASE):
return f"日期字段使用了 {data_type} 类型，建议使用 DATE/DATETIME"
return None
def check_int_type_mismatch(column_name, data_type):
"""ID类字段是否使用了字符串类型"""
dtype = data_type.lower().strip()
if dtype in ("int", "bigint", "smallint", "tinyint", "mediumint", "integer"):
return None # 正确
c = column_name.lower().strip()
for pat in INT_FIELD_PATTERNS:
if c == pat or c.endswith("_" + pat):
if dtype in ("varchar", "char", "text"):
return f"ID字段 {column_name} 使用了 {data_type} 类型，建议使用 BIGINT"
return None
def analyze():
print("=" * 50)
print(" 字段长度超标检查")
print("=" * 50)
data_dict = load_json(DATA_DICT_FILE)
violations = []
type_mismatches = []
date_type_issues = []
id_type_issues = []
# 按表+字段去重
seen = set()
for col in data_dict:
tname = col["table_name"]
cname = col["column_name"]
dtype = col.get("data_type", "").strip().lower()
col_type = col.get("column_type", "").strip()
char_len = col.get("char_max_length")
comment = col.get("column_comment", "")
key = (tname, cname)
if key in seen:
continue
seen.add(key)
# ── 1. 检查固定长度字段 ──
for rule_key, (required_len, required_type, standard, note) in FIXED_LENGTH_RULES.items():
if not match_field(cname, rule_key):
continue
# 如果要求 char 类型但当前字段是 varchar/char/text 系列
if required_type == "char" and dtype in ("varchar", "char"):
# VARCHAR(N) / CHAR(N) — 检查 N 是否超出
if char_len is not None and required_len > 0 and char_len > required_len:
violations.append({
"table_name": tname,
"column_name": cname,
"column_type": col_type,
"data_type": dtype,
"actual_length": char_len,
"required_length": required_len,
"excess": char_len - required_len,
"standard": standard,
"rule": note,
"column_comment": comment,
"suggestion": f"将 {col_type} 改为 CHAR({required_len}) 或 VARCHAR({required_len})",
})
# CHAR(N) 且 N != required_len
if dtype == "char" and required_len > 0:
# CHAR 本身就是定长，从 column_type 中提取长度
match = re.match(r"char\((\d+)\)", col_type.lower())
if match:
actual = int(match.group(1))
if actual > required_len:
violations.append({
"table_name": tname,
"column_name": cname,
"column_type": col_type,
"data_type": dtype,
"actual_length": actual,
"required_length": required_len,
"excess": actual - required_len,
"standard": standard,
"rule": note,
"column_comment": comment,
"suggestion": f"将 CHAR({actual}) 改为 CHAR({required_len})",
})
# 如果要求 numeric/double 类型但当前是 varchar/char
if required_type in ("double", "int") and dtype in ("varchar", "char", "text"):
type_mismatches.append({
"table_name": tname,
"column_name": cname,
"column_type": col_type,
"data_type": dtype,
"required_type": required_type,
"standard": standard,
"rule": note,
"column_comment": comment,
"suggestion": f"应将 {cname} 类型从 {col_type} 改为 {required_type.upper()}",
})
# ── 2. 日期字段类型检查 ──
date_issue = check_date_type_mismatch(cname, dtype)
if date_issue:
date_type_issues.append({
"table_name": tname,
"column_name": cname,
"column_type": col_type,
"data_type": dtype,
"issue": date_issue,
"column_comment": comment,
})
# ── 3. ID字段类型检查 ──
id_issue = check_int_type_mismatch(cname, dtype)
if id_issue:
id_type_issues.append({
"table_name": tname,
"column_name": cname,
"column_type": col_type,
"data_type": dtype,
"issue": id_issue,
"column_comment": comment,
})
# ── 汇总 ──
print(f" 长度超标字段: {len(violations)} 个")
print(f" 类型不匹配(经纬度等应为数值型): {len(type_mismatches)} 个")
print(f" 日期字段使用非日期类型: {len(date_type_issues)} 个")
print(f" ID字段使用字符串类型: {len(id_type_issues)} 个")
# 按超标程度排序
violations.sort(key=lambda x: x["excess"], reverse=True)
type_mismatches.sort(key=lambda x: x["table_name"])
# 统计摘要
tables_affected = set(v["table_name"] for v in violations)
by_standard = {}
for v in violations:
std = v["standard"]
by_standard[std] = by_standard.get(std, 0) + 1
result = {
"meta": {
"script": "check_field_length.py",
"description": "字段长度超标检查 - 对比国家标准/规范中的固定长度要求",
"total_violations": len(violations),
"total_type_mismatches": len(type_mismatches),
"total_date_type_issues": len(date_type_issues),
"total_id_type_issues": len(id_type_issues),
"tables_affected": len(tables_affected),
},
"summary_by_standard": [
{"standard": std, "violation_count": cnt}
for std, cnt in sorted(by_standard.items(), key=lambda x: x[1], reverse=True)
],
"length_violations": violations,
"type_mismatches": type_mismatches,
"date_type_issues": date_type_issues,
"id_type_issues": id_type_issues,
}
return result
def main():
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
result = analyze()
with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
json.dump(result, f, ensure_ascii=False, indent=2)
print(f"\n✓ 结果已保存至: {OUTPUT_FILE}")
# Top 10 超标
print(f"\n{'─' * 50}")
print(" 长度超标 Top 10（按超出位数排序）")
print(f"{'─' * 50}")
for v in result["length_violations"][:10]:
print(f" {v['table_name']}.{v['column_name']}: "
f"{v['column_type']} (需{v['required_length']}位, 超出{v['excess']}位) "
f"— {v['standard']}")
if __name__ == "__main__":
main()
data_dictionary/check_uncommented_fields.py 0 → 100644
View file @ 31911f51
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据治理 - 无注释字段检查脚本
功能：扫描数据字典，找出所有没有注释的字段，
并根据字段名、数据类型、所在表名等信息推测可能的注释。
数据来源：data_dictionary.json
输出文件: uncommented_fields.json
"""
import json
import os
import re
import sys
# ── 配置 ──────────────────────────────────────────────
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DICT_FILE = os.path.join(SCRIPT_DIR, "data_dictionary.json")
OUTPUT_FILE = os.path.join(SCRIPT_DIR, "uncommented_fields.json")
# ── 字段注释推断规则库 ─────────────────────────────────
# 规则优先级：精确匹配 > 后缀匹配 > 前缀匹配 > 模糊匹配 > 表名推断
# 格式: "匹配模式" → ("推测注释", 置信度: high/medium/low)
INFERENCE_RULES = [
# ========== 审计/通用字段 (high confidence) ==========
(r"^create_time$", "创建时间", "high"),
(r"^update_time$", "更新时间", "high"),
(r"^create_by$", "创建人", "high"),
(r"^update_by$", "更新人", "high"),
(r"^create_user$", "创建人", "high"),
(r"^update_user$", "更新人", "high"),
(r"^remark$", "备注", "high"),
(r"^remarks$", "备注", "high"),
(r"^status$", "状态", "medium"),
(r"^version$", "版本号", "high"),
(r"^sort_order$", "排序号", "high"),
(r"^sort$", "排序号", "medium"),
(r"^is_del(eted)?$", "是否删除（0-未删除，1-已删除）", "high"),
(r"^is_deleted$", "是否删除（0-未删除，1-已删除）", "high"),
(r"^is_.*$", "是否标记", "low"), # 兜底
# ========== 主键/外键ID (high confidence) ==========
(r"^id$", "主键ID", "high"),
(r"^guid$", "全局唯一标识(GUID)", "high"),
(r"^uuid$", "全局唯一标识(UUID)", "high"),
(r".+_id$", "关联ID", "medium"), # enterprise_id, credit_id 等 → 具体化在下表名推断
(r"^parent_id$", "父级ID", "high"),
(r"^parent_code$", "父级编码", "high"),
(r"^pid$", "父级ID", "high"),
# ========== 时间字段 (high confidence) ==========
(r"^review_time$", "审核时间", "high"),
(r"^evaluation_time$", "评价时间", "high"),
(r"^audit_time$", "审批时间", "high"),
(r"^approve_time$", "审批时间", "high"),
(r"^submit_time$", "提交时间", "high"),
(r"^publish_time$", "发布时间", "high"),
(r"^expire_time$", "过期时间", "high"),
(r"^start_time$", "开始时间", "high"),
(r"^end_time$", "结束时间", "high"),
(r"^.+(date|time)$", "时间字段", "low"),
# ========== 人员字段 (high confidence) ==========
(r"^creator$", "创建人", "high"),
(r"^updater$", "更新人", "high"),
(r"^operator$", "操作人", "high"),
(r"^submitter$", "提交人", "high"),
(r"^approver$", "审批人", "high"),
(r"^reviewer$", "审核人", "high"),
(r"^user_name$", "用户名", "high"),
(r"^real_name$", "真实姓名", "high"),
(r"^nick_name$", "昵称", "high"),
# ========== 名称字段 (high confidence) ==========
(r"^name$", "名称", "high"),
(r"^full_name$", "全称", "high"),
(r"^short_name$", "简称", "high"),
(r"^title$", "标题", "high"),
(r"^description$", "描述", "high"),
(r"^desc$", "描述", "high"),
# ========== 组织/企业字段 (medium-high confidence) ==========
(r"^enterprise_id$", "企业ID", "high"),
(r"^enterprise_name$", "企业名称", "high"),
(r"^company_id$", "公司ID", "high"),
(r"^company_name$", "公司名称", "high"),
(r"^org_id$", "组织/机构ID", "high"),
(r"^dept_id$", "部门ID", "high"),
(r"^credit_id$", "信用记录ID", "high"),
# ========== 地理/区域字段 (high confidence) ==========
(r"^country_id$", "国家ID", "high"),
(r"^country_name$", "国家名称", "high"),
(r"^country_iso3$", "国家ISO3代码（ISO 3166-1）", "high"),
(r"^continent_id$", "大洲ID", "high"),
(r"^province_id$", "省份ID", "high"),
(r"^city_id$", "城市ID", "high"),
(r"^district_id$", "区县ID", "high"),
(r"^community_id$", "社区ID", "high"),
(r"^region_code$", "行政区划代码", "high"),
(r"^adcode$", "行政区划代码（高德）", "high"),
(r"^scope_type$", "范围类型", "medium"),
(r"^region_name$", "区域名称", "high"),
(r"^region_level$", "区域级别（国家/省/市/区县）", "high"),
(r"^address$", "地址", "high"),
(r"^detail_address$", "详细地址", "high"),
# ========== 地理空间字段 (high confidence) ==========
(r"^(geojson|geo_json)$", "GeoJSON地理数据", "high"),
(r"^geom(etry)?$", "几何地理数据", "high"),
(r"^centroid_lng$", "中心点经度（ISO 6709）", "high"),
(r"^centroid_lat$", "中心点纬度（ISO 6709）", "high"),
(r"^(lng|longitude)$", "经度（ISO 6709）", "high"),
(r"^(lat|latitude)$", "纬度（ISO 6709）", "high"),
(r"^bbox$", "边界框坐标", "high"),
# ========== 来源/类型字段 ==========
(r"^source_url$", "数据来源URL", "high"),
(r"^source_type$", "数据来源类型", "high"),
(r"^source$", "数据来源", "medium"),
(r"^type$", "类型", "medium"),
# ========== 年份/数量字段 ==========
(r"^year$", "年份", "high"),
(r"^month$", "月份", "high"),
(r"^count$", "数量", "medium"),
(r"^total$", "总计", "medium"),
(r"^amount$", "金额", "high"),
# ========== AI相关字段 ==========
(r"^conversation_id$", "AI对话ID", "high"),
(r"^provider_id$", "AI服务商ID", "high"),
(r"^model_id$", "AI模型ID", "high"),
(r"^model_name$", "AI模型名称", "high"),
(r"^prompt$", "提示词/AI输入", "high"),
(r"^response$", "AI响应/输出", "high"),
# ========== 其他常见字段 ==========
(r"^candidate_desc$", "候选人描述", "high"),
(r"^url$", "链接地址", "high"),
(r"^link$", "链接", "high"),
(r"^email$", "邮箱地址", "high"),
(r"^phone$", "联系电话", "high"),
(r"^mobile$", "手机号码", "high"),
(r"^file_path$", "文件路径", "high"),
(r"^file_url$", "文件链接", "high"),
(r"^attachment$", "附件", "high"),
(r"^content$", "内容", "medium"),
(r"^body$", "正文内容", "medium"),
(r"^config$", "配置信息", "medium"),
(r"^params$", "参数", "medium"),
(r"^extra$", "扩展信息", "medium"),
(r"^meta$", "元数据", "high"),
(r"^metadata$", "元数据", "high"),
]
# ── 表名→业务领域映射，辅助推断 _id 字段 ──
TABLE_DOMAIN_MAP = {
"enterprise": "企业",
"company": "公司",
"project": "项目",
"user": "用户",
"talent": "人才",
"order": "订单",
"course": "课程",
"exam": "考试",
"credit": "信用",
"complaint": "投诉",
"appeal": "申诉",
"compliance": "合规",
"reward": "奖励",
"risk": "风险",
"performance": "绩效",
"profile": "企业画像",
"country": "国家",
"continent": "大洲",
"region": "区域",
"community": "社区",
"street": "街道",
"county": "县",
"plot": "地块",
"ai": "AI",
"bri": "一带一路",
"doc": "文档",
"tender": "投标",
"construction": "施工",
"plan": "方案",
"meteorological": "气象",
}
def load_json(filepath):
with open(filepath, "r", encoding="utf-8") as f:
return json.load(f)
def infer_comment(column_name, table_name, data_type, column_type):
"""
根据字段名、表名、数据类型推测字段注释。
返回 (推测注释, 置信度) 或 (None, None)
"""
cname = column_name.lower().strip()
tname = table_name.lower().strip()
# ── 第一轮：精确匹配和后缀匹配 ──
for pattern, comment, confidence in INFERENCE_RULES:
if re.match(pattern, cname, re.IGNORECASE):
# 对于 _id 结尾的字段，尝试结合表名给出更精确的推断
if cname.endswith("_id") and confidence == "medium":
specific = refine_id_field(cname, tname)
if specific:
return (specific, "medium")
return (comment, confidence)
# ── 第二轮：根据数据类型推断 ──
dtype = data_type.lower().strip()
if dtype in ("datetime", "timestamp"):
if "create" in cname:
return ("创建时间", "high")
if "update" in cname:
return ("更新时间", "high")
return ("时间字段（具体含义待确认）", "low")
if dtype in ("date",):
return ("日期字段（具体含义待确认）", "low")
# ── 第三轮：根据表名推断关联字段 ──
# 例如 t_enterprise_credit_appeal 中的 appeal_id 可能不存在，但 enterprise_id 已经匹配
# 这部分主要是处理没匹配到的通用字段
return (None, None)
def refine_id_field(column_name, table_name):
"""
将 _id 结尾的字段结合表名细化。
例如 enterprise_id 在 enterprise_credit 表中 → "企业ID"
"""
# 去掉 _id 后缀得到实体名
entity = re.sub(r"_id$", "", column_name)
# 如果实体名已经能说明问题，直接翻译
known_entities = {
"enterprise": "企业ID",
"company": "公司ID",
"project": "项目ID",
"user": "用户ID",
"talent": "人才ID",
"order": "订单ID",
"course": "课程ID",
"exam": "考试ID",
"credit": "信用记录ID",
"shop": "店铺ID",
"job": "职位ID",
"dept": "部门ID",
"org": "组织/机构ID",
"customer": "客户ID",
"institution": "机构ID",
"template": "模板ID",
"policy": "政策ID",
"reserve": "储备ID",
"config": "配置ID",
"country": "国家ID",
"continent": "大洲ID",
"region": "区域ID",
"community": "社区ID",
"street": "街道ID",
"county": "县ID",
"provider": "服务商ID",
"conversation": "对话ID",
"attachment": "附件ID",
"plan": "方案ID",
"document": "文档ID",
"tender": "投标ID",
"construction": "施工ID",
"complaint": "投诉ID",
"appeal": "申诉ID",
"reward": "奖励ID",
"risk": "风险ID",
"performance": "绩效ID",
"profile": "企业画像ID",
}
if entity in known_entities:
return known_entities[entity]
# 尝试从表名推断
for domain_key, domain_name in TABLE_DOMAIN_MAP.items():
if domain_key in table_name and domain_key in entity:
return f"{domain_name}ID"
return f"{entity}ID（关联{entity}表）"
def check_uncommented_fields():
print("=" * 60)
print(" 无注释字段检查")
print("=" * 60)
data_dict = load_json(DATA_DICT_FILE)
# 筛选无注释字段
uncommented = []
for col in data_dict:
comment = col.get("column_comment", "").strip()
if not comment:
uncommented.append(col)
print(f"\n 数据字典总字段数: {len(data_dict)}")
print(f" 无注释字段数: {len(uncommented)}")
print(f" 无注释比例: {len(uncommented) / len(data_dict) * 100:.1f}%")
# 推断注释
inferred = 0
cannot_infer = 0
results = []
for col in uncommented:
tname = col["table_name"]
cname = col["column_name"]
dtype = col.get("data_type", "")
col_type = col.get("column_type", "")
inferred_comment, confidence = infer_comment(cname, tname, dtype, col_type)
record = {
"table_name": tname,
"column_name": cname,
"column_type": col_type,
"data_type": dtype,
"char_max_length": col.get("char_max_length"),
"is_nullable": col.get("is_nullable"),
"column_default": col.get("column_default"),
"table_comment": col.get("table_comment", ""),
"inferred_comment": inferred_comment,
"confidence": confidence,
}
if inferred_comment:
inferred += 1
else:
cannot_infer += 1
results.append(record)
# 按表+字段排序
results.sort(key=lambda x: (x["table_name"], x["column_name"]))
# 统计
# 按表统计
from collections import Counter, defaultdict
table_stats = Counter()
table_details = defaultdict(list)
for r in results:
table_stats[r["table_name"]] += 1
table_details[r["table_name"]].append(r)
# 按置信度统计
confidence_stats = Counter()
for r in results:
if r["confidence"]:
confidence_stats[r["confidence"]] += 1
else:
confidence_stats["cannot_infer"] += 1
summary = {
"meta": {
"script": "check_uncommented_fields.py",
"description": "无注释字段检查及推测注释",
"total_fields": len(data_dict),
"uncommented_fields": len(uncommented),
"uncommented_ratio": f"{len(uncommented) / len(data_dict) * 100:.1f}%",
"inferred": inferred,
"cannot_infer": cannot_infer,
"inferred_ratio": f"{inferred / len(uncommented) * 100:.1f}%" if uncommented else "N/A",
},
"confidence_distribution": {
"high": confidence_stats.get("high", 0),
"medium": confidence_stats.get("medium", 0),
"low": confidence_stats.get("low", 0),
"cannot_infer": confidence_stats.get("cannot_infer", 0),
},
"tables_affected": len(table_stats),
"by_table": [
{
"table_name": tname,
"table_comment": table_details[tname][0]["table_comment"],
"uncommented_count": cnt,
}
for tname, cnt in table_stats.most_common()
],
"results": results,
}
return summary
def main():
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
result = check_uncommented_fields()
with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
json.dump(result, f, ensure_ascii=False, indent=2)
print(f"\n✓ 结果已保存至: {OUTPUT_FILE}")
# ── 打印摘要 ──
meta = result["meta"]
print(f"\n{'─' * 60}")
print(" 摘要")
print(f"{'─' * 60}")
print(f" 无注释字段总数: {meta['uncommented_fields']}")
print(f" 可推断: {meta['inferred']} ({meta['inferred_ratio']})")
print(f" 无法推断: {meta['cannot_infer']}")
print(f" 置信度分布: 高={result['confidence_distribution']['high']}, "
f"中={result['confidence_distribution']['medium']}, "
f"低={result['confidence_distribution']['low']}")
print(f"\n{'─' * 60}")
print(" 各表无注释字段 Top 10")
print(f"{'─' * 60}")
for item in result["by_table"][:10]:
print(f" {item['table_name']}: {item['uncommented_count']}个 "
f"({item['table_comment']})")
# ── 无法推断的字段 ──
cannot = [r for r in result["results"] if not r["inferred_comment"]]
if cannot:
print(f"\n{'─' * 60}")
print(f" 无法推断的字段 ({len(cannot)}个)")
print(f"{'─' * 60}")
for r in cannot:
print(f" [{r['table_name']}] {r['column_name']} ({r['column_type']})")
if __name__ == "__main__":
main()
data_dictionary/fetch_data_dictionary.py 0 → 100644
View file @ 31911f51
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据字典获取脚本
功能：连接 MySQL 数据库，读取 smart-build 库中所有表的字段信息，
并以 JSON 格式保存到本地文件。
输出字段：
- 表名 (TABLE_NAME)
- 字段名 (COLUMN_NAME)
- 字段注释 (COLUMN_COMMENT)
- 数据类型 (COLUMN_TYPE / DATA_TYPE)
- 字符最大长度 (CHARACTER_MAXIMUM_LENGTH)
- 是否可为空 (IS_NULLABLE)
- 默认值 (COLUMN_DEFAULT)
- 字段序号 (ORDINAL_POSITION)
- 表注释 (TABLE_COMMENT)
"""
import json
import os
import sys
from datetime import datetime
from getpass import getpass
try:
import pymysql
except ImportError:
print("错误：缺少 pymysql 库，请执行: pip install pymysql")
sys.exit(1)
# ── 配置 ──────────────────────────────────────────────
DB_CONFIG = {
"host": "192.168.20.10",
"port": 3306,
"user": "root",
"database": "smart-build",
"charset": "utf8mb4",
}
OUTPUT_FILE = "data_dictionary.json"
OUTPUT_FILE_TABLES = "table_summary.json"
# ── 获取密码 ──────────────────────────────────────────
def get_db_password():
"""以安全方式获取数据库密码（不回显）"""
print("=" * 50)
print(f" 连接目标: {DB_CONFIG['user']}@{DB_CONFIG['host']}:{DB_CONFIG['port']}")
print(f" 数据库: {DB_CONFIG['database']}")
print("=" * 50)
password = getpass("请输入数据库密码（输入不显示）: ")
if not password:
print("错误：密码不能为空")
sys.exit(1)
return password
# ── 连接数据库 ────────────────────────────────────────
def connect_db(password):
"""建立数据库连接"""
try:
conn = pymysql.connect(
host=DB_CONFIG["host"],
port=DB_CONFIG["port"],
user=DB_CONFIG["user"],
password=password,
database=DB_CONFIG["database"],
charset=DB_CONFIG["charset"],
connect_timeout=10,
)
print("✓ 数据库连接成功\n")
return conn
except pymysql.err.OperationalError as e:
print(f"错误：无法连接数据库 — {e}")
sys.exit(1)
# ── 查询字段详情 ──────────────────────────────────────
def fetch_column_details(cursor):
"""
从 INFORMATION_SCHEMA.COLUMNS 获取当前库所有字段信息。
返回: list[dict]
"""
sql = """
SELECT
c.TABLE_NAME AS table_name,
c.COLUMN_NAME AS column_name,
c.ORDINAL_POSITION AS ordinal_position,
c.COLUMN_TYPE AS column_type,
c.DATA_TYPE AS data_type,
c.CHARACTER_MAXIMUM_LENGTH AS char_max_length,
c.NUMERIC_PRECISION AS numeric_precision,
c.NUMERIC_SCALE AS numeric_scale,
c.IS_NULLABLE AS is_nullable,
c.COLUMN_DEFAULT AS column_default,
c.COLUMN_COMMENT AS column_comment,
c.EXTRA AS extra,
t.TABLE_COMMENT AS table_comment
FROM
INFORMATION_SCHEMA.COLUMNS AS c
LEFT JOIN
INFORMATION_SCHEMA.TABLES AS t
ON c.TABLE_SCHEMA = t.TABLE_SCHEMA
AND c.TABLE_NAME = t.TABLE_NAME
WHERE
c.TABLE_SCHEMA = %s
ORDER BY
c.TABLE_NAME,
c.ORDINAL_POSITION
"""
cursor.execute(sql, (DB_CONFIG["database"],))
rows = cursor.fetchall()
columns_meta = [desc[0] for desc in cursor.description]
result = []
for row in rows:
record = dict(zip(columns_meta, row))
# 类型转换，确保 JSON 可序列化
for key in ("char_max_length", "numeric_precision", "numeric_scale"):
if record.get(key) is not None:
record[key] = int(record[key])
if record.get("ordinal_position") is not None:
record["ordinal_position"] = int(record["ordinal_position"])
result.append(record)
return result
# ── 查询表汇总信息 ────────────────────────────────────
def fetch_table_summary(cursor):
"""
从 INFORMATION_SCHEMA.TABLES 获取表级别汇总信息。
返回: list[dict]
"""
sql = """
SELECT
TABLE_NAME,
TABLE_TYPE,
ENGINE,
TABLE_ROWS,
DATA_LENGTH,
INDEX_LENGTH,
TABLE_COMMENT,
CREATE_TIME,
UPDATE_TIME
FROM
INFORMATION_SCHEMA.TABLES
WHERE
TABLE_SCHEMA = %s
ORDER BY
TABLE_NAME
"""
cursor.execute(sql, (DB_CONFIG["database"],))
rows = cursor.fetchall()
columns_meta = [desc[0] for desc in cursor.description]
result = []
for row in rows:
record = dict(zip(columns_meta, row))
# 时间字段转字符串
for t_field in ("CREATE_TIME", "UPDATE_TIME"):
if record.get(t_field) is not None:
record[t_field] = record[t_field].strftime("%Y-%m-%d %H:%M:%S")
# 数值转 int
for n_field in ("TABLE_ROWS", "DATA_LENGTH", "INDEX_LENGTH"):
if record.get(n_field) is not None:
record[n_field] = int(record[n_field])
result.append(record)
return result
# ── 保存 JSON ─────────────────────────────────────────
def save_json(data, filepath, label):
script_dir = os.path.dirname(os.path.abspath(__file__))
full_path = os.path.join(script_dir, filepath)
with open(full_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"✓ {label}已保存至: {full_path} ({len(data)} 条记录)")
# ── 统计输出 ──────────────────────────────────────────
def print_stats(column_details, table_summary):
"""输出简要统计信息到控制台"""
tables_in_columns = set(r["table_name"] for r in column_details)
print(f"\n{'─' * 40}")
print(f" 表总数（数据字典）: {len(tables_in_columns)}")
print(f" 字段总数: {len(column_details)}")
print(f" 表总数（汇总）: {len(table_summary)}")
print(f"{'─' * 40}\n")
# ── 主流程 ────────────────────────────────────────────
def main():
password = get_db_password()
conn = connect_db(password)
try:
with conn.cursor() as cursor:
# 1. 获取字段详情
print("正在获取字段详情...")
column_details = fetch_column_details(cursor)
save_json(column_details, OUTPUT_FILE, "字段详情")
# 2. 获取表汇总
print("正在获取表汇总信息...")
table_summary = fetch_table_summary(cursor)
save_json(table_summary, OUTPUT_FILE_TABLES, "表汇总")
# 3. 统计数据
print_stats(column_details, table_summary)
finally:
conn.close()
print("数据库连接已关闭。")
if __name__ == "__main__":
main()
data_dictionary/generate_report.py 0 → 100644
View file @ 31911f51
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据治理 - Word 报告生成脚本
汇总所有分析结果，生成最终的 Word 格式数据治理报告。
输出文件: 数据治理报告_smart-build.docx
"""
import json
import os
import sys
from datetime import datetime
from docx import Document
from docx.shared import Inches, Pt, Cm, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.enum.table import WD_TABLE_ALIGNMENT
from docx.oxml.ns import qn
from docx.oxml import OxmlElement
# ── 配置 ──────────────────────────────────────────────
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = SCRIPT_DIR
OUTPUT_FILE = os.path.join(SCRIPT_DIR, "..", "数据治理报告_smart-build.docx")
def load_json(filename):
path = os.path.join(DATA_DIR, filename)
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
# ── 样式辅助 ──────────────────────────────────────────
def set_cell_shading(cell, color):
tc = cell._tc
tcPr = tc.get_or_add_tcPr()
shading = OxmlElement('w:shd')
shading.set(qn('w:fill'), color)
shading.set(qn('w:val'), 'clear')
tcPr.append(shading)
def add_styled_heading(doc, text, level=1):
return doc.add_heading(text, level=level)
def add_para(doc, text, bold=False, font_size=11, alignment=None, space_after=6):
p = doc.add_paragraph()
run = p.add_run(text)
run.font.size = Pt(font_size)
run.bold = bold
run.font.name = '宋体'
run._element.rPr.rFonts.set(qn('w:eastAsia'), '宋体')
if alignment is not None:
p.alignment = alignment
p.paragraph_format.space_after = Pt(space_after)
return p
def add_table(doc, headers, rows, col_widths=None, header_color="2F5496", font_size=8.5):
table = doc.add_table(rows=1 + len(rows), cols=len(headers))
table.style = 'Table Grid'
table.alignment = WD_TABLE_ALIGNMENT.CENTER
for i, header in enumerate(headers):
cell = table.rows[0].cells[i]
cell.text = header
for paragraph in cell.paragraphs:
paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER
for run in paragraph.runs:
run.bold = True
run.font.size = Pt(8)
run.font.color.rgb = RGBColor(0xFF, 0xFF, 0xFF)
run.font.name = '宋体'
run._element.rPr.rFonts.set(qn('w:eastAsia'), '宋体')
set_cell_shading(cell, header_color)
for r, row in enumerate(rows):
for c, val in enumerate(row):
cell = table.rows[r + 1].cells[c]
cell.text = str(val) if val is not None else ""
for paragraph in cell.paragraphs:
for run in paragraph.runs:
run.font.size = Pt(font_size)
run.font.name = '宋体'
run._element.rPr.rFonts.set(qn('w:eastAsia'), '宋体')
if r % 2 == 1:
set_cell_shading(cell, "F2F2F2")
if col_widths:
for i, width in enumerate(col_widths):
for row in table.rows:
row.cells[i].width = Cm(width)
return table
# ── 报告生成 ──────────────────────────────────────────
def generate_report():
print("=" * 60)
print(" 开始生成数据治理报告...")
print("=" * 60)
# 加载数据
table_summary = load_json("table_summary.json")
merge_findings = load_json("findings_verified.json")
empty_fields = load_json("empty_fields_report.json")
uncommented = load_json("uncommented_fields.json")
length_violations = load_json("field_length_violations.json")
standard_violations = load_json("validate_standard_fields.json")
data_dict = load_json("data_dictionary.json")
# 每个表的字段数
column_count_map = {}
for col in data_dict:
tn = col["table_name"]
column_count_map[tn] = column_count_map.get(tn, 0) + 1
doc = Document()
style = doc.styles['Normal']
font = style.font
font.name = '宋体'
font.size = Pt(10.5)
style.element.rPr.rFonts.set(qn('w:eastAsia'), '宋体')
# ==================== 封面 ====================
for _ in range(6):
doc.add_paragraph()
t = doc.add_paragraph(); t.alignment = WD_ALIGN_PARAGRAPH.CENTER
run = t.add_run("数据库数据治理分析报告"); run.font.size = Pt(28)
run.bold = True; run.font.color.rgb = RGBColor(0x2F, 0x54, 0x96)
run.font.name = '黑体'; run._element.rPr.rFonts.set(qn('w:eastAsia'), '黑体')
doc.add_paragraph()
t = doc.add_paragraph(); t.alignment = WD_ALIGN_PARAGRAPH.CENTER
run = t.add_run("smart-build 数据库"); run.font.size = Pt(18)
run.font.color.rgb = RGBColor(0x59, 0x56, 0x59)
run.font.name = '微软雅黑'; run._element.rPr.rFonts.set(qn('w:eastAsia'), '微软雅黑')
doc.add_paragraph(); doc.add_paragraph()
t = doc.add_paragraph(); t.alignment = WD_ALIGN_PARAGRAPH.CENTER
run = t.add_run(f"报告生成日期：{datetime.now().strftime('%Y年%m月%d日')}")
run.font.size = Pt(12); run.font.color.rgb = RGBColor(0x80, 0x80, 0x80)
doc.add_page_break()
# ==================== 目录 ====================
add_styled_heading(doc, "目 录", level=1)
for item in [
"一、项目概述", "二、数据库概况", "三、表结构分析 — 表合并与冗余",
"四、表结构分析 — 大范围空字段", "五、表结构分析 — 无注释字段",
"六、字段规范检查 — 存储长度超标", "七、字段规范检查 — 数据不符合标准",
"八、综合问题清单", "九、改进建议汇总", "十、附录",
]:
add_para(doc, item, font_size=12, space_after=4)
doc.add_page_break()
# ==================== 一、项目概述 ====================
add_styled_heading(doc, "一、项目概述", level=1)
add_styled_heading(doc, "1.1 项目背景", level=2)
add_para(doc, "本项目对 smart-build 数据库进行全面数据治理分析，旨在发现数据库设计、数据存储和数据质量方面的问题，为数据库优化提供参考依据。所有分析均为只读操作，不涉及任何数据修改。")
add_styled_heading(doc, "1.2 分析目标", level=2)
for i, g in enumerate([
"检查可合并的数据库表，识别冗余字段",
"检查大范围空值（NULL/空字符串）字段",
"检查缺少注释的字段，并尝试推测注释内容",
"检查字段存储长度是否超出标准所需值（如身份证号18位却用VARCHAR(50)存储）",
"检查数据是否符合国家/国际标准（身份证、手机号、统一社会信用代码、行政区划代码等）",
], 1):
add_para(doc, f"{i}. {g}")
add_styled_heading(doc, "1.3 分析范围", level=2)
for s in [
f"数据库：smart-build (MySQL)",
f"服务器：192.168.20.10",
f"分析日期：2026年8月3日",
f"数据字典字段总数：{uncommented['meta']['total_fields']} 个",
f"数据表总数：{len(table_summary)} 张",
]:
add_para(doc, s)
add_styled_heading(doc, "1.4 参考标准", level=2)
for s in [
"GB 11643-1999 — 公民身份号码",
"GB 32100-2015 — 法人和其他组织统一社会信用代码编码规则",
"GB/T 2260 — 中华人民共和国行政区划代码",
"GB/T 2261.1 — 人的性别代码",
"GB/T 3304 — 中国各民族名称的罗马字母拼写法和代码",
"GB/T 4658 — 学历代码",
"GB/T 4762 — 政治面貌代码",
"GB/T 6565 — 职业分类与代码",
"GB/T 4754 — 国民经济行业分类",
"ISO 3166-1 — 国家代码", "ISO 4217 — 货币代码",
"ISO 639-1 — 语言代码",
"ISO 6709 — 地理点位置的纬度、经度和高程的标准表示法",
"工信部 — 电信网编号计划（手机号码段）",
]:
add_para(doc, f"• {s}", font_size=10, space_after=2)
doc.add_page_break()
# ==================== 二、数据库概况 ====================
add_styled_heading(doc, "二、数据库概况", level=1)
total_tables = len(table_summary)
with_data = sum(1 for t in table_summary if t.get("TABLE_ROWS", 0) > 0)
empty_tables = total_tables - with_data
total_rows = sum(t.get("TABLE_ROWS", 0) for t in table_summary)
for s in [
f"数据表总数：{total_tables} 张",
f"有数据的表：{with_data} 张",
f"空表（估算行数=0）：{empty_tables} 张",
f"估算总数据行数：{total_rows:,} 行",
f"数据字典字段总数：{uncommented['meta']['total_fields']} 个",
]:
add_para(doc, f"• {s}")
# 全部空表（数据行数=0）
add_styled_heading(doc, f"2.1 空表列表（共 {empty_tables} 张）", level=2)
add_para(doc, "以下表在 INFORMATION_SCHEMA 中估算行数为0，建议与业务方确认是否可以归档删除：", font_size=10)
empty_list = sorted([t["TABLE_NAME"] for t in table_summary if t.get("TABLE_ROWS", 0) == 0])
add_para(doc, "、".join(empty_list), font_size=9, space_after=4)
add_para(doc, "", space_after=2)
# 全部表按数据量排列（有数据的表）
add_styled_heading(doc, f"2.2 数据表规模一览（有数据的表，共 {with_data} 张）", level=2)
sorted_tables = sorted(
[t for t in table_summary if t.get("TABLE_ROWS", 0) > 0],
key=lambda x: x.get("TABLE_ROWS", 0), reverse=True
)
headers = ["排名", "表名", "表注释", "估算行数", "字段数"]
rows = []
for i, t in enumerate(sorted_tables, 1):
tn = t.get("TABLE_NAME", "")
rows.append([i, tn, t.get("TABLE_COMMENT", ""),
f"{t.get('TABLE_ROWS', 0):,}", column_count_map.get(tn, 0)])
add_table(doc, headers, rows, col_widths=[1, 3.5, 5, 2, 1.5], font_size=7.5)
doc.add_page_break()
# ==================== 三、表合并与冗余 ====================
add_styled_heading(doc, "三、表结构分析 — 表合并与冗余", level=1)
s = merge_findings["summary"]
add_styled_heading(doc, "3.1 检查概况", level=2)
for line in [
f"分析数据表总数：{s['total_tables']} 张",
f"分析字段总数：{s['total_fields']} 个",
f"发现可合并候选：{s['confirmed_merge_candidates']} 组",
f"两套命名规范的表（t_ vs t_talent_）：{s['confirmed_decommission_strong']} 组",
f"零行表（需确认是否在用）：{s['confirmed_decommission_weak']} 张",
f"发现冗余字段模式：{s['redundancy_fields_confirmed']} 类",
f"数据质量问题：{s['data_quality_issues']} 项",
]:
add_para(doc, f"• {line}")
# 合并候选
add_styled_heading(doc, "3.2 建议合并的表", level=2)
for mc in merge_findings["confirmed_merge_candidates"]:
add_styled_heading(doc, f"{mc['id']}: 优先级{mc['priority']}", level=3)
add_para(doc, f"涉及表：{'、'.join(mc['tables'])}", bold=True)
add_para(doc, f"公共字段数：{mc.get('common_columns', 'N/A')} 个")
add_para(doc, f"分析发现：{mc['data_evidence']}")
add_para(doc, f"建议方案：{mc['suggestion']}")
add_para(doc, f"实施难度评估：{mc.get('effort', 'N/A')}")
if mc.get('risk'):
add_para(doc, f"注意事项：{mc['risk']}")
# 可考虑归档的表
add_styled_heading(doc, "3.3 建议确认后归档的表", level=2)
strong = merge_findings["confirmed_decommission"]["strong"]
add_styled_heading(doc, "3.3.1 t_ 前缀人才相关表（共19张，全部0行数据）", level=3)
add_para(doc, "数据库中存在两套命名规范的人才相关表：t_ 前缀的19张表全部0行数据，t_talent_ 前缀的13张对应表各有1-12行数据。两套表的功能存在对应关系，建议与业务方确认两套表的定位和使用情况。")
add_para(doc, f"t_ 前缀表（{len(strong['tables'])}张，数据行数均为0）：{'、'.join(strong['tables'])}")
add_para(doc, "t_talent_ 前缀对应表（有实际数据）：")
for tn, nrows in strong["new_system_status"].items():
add_para(doc, f" • {tn}: {nrows} 行", font_size=10, space_after=1)
weak = merge_findings["confirmed_decommission"]["weak"]
add_styled_heading(doc, "3.3.2 零行表（共11张，建议与业务方确认是否在用）", level=3)
add_para(doc, weak["description"])
add_para(doc, f"涉及表（{len(weak['tables'])}张）：{'、'.join(weak['tables'])}")
if weak.get("exception"):
add_para(doc, f"说明：{weak['exception']}")
# 冗余字段
add_styled_heading(doc, "3.4 发现的冗余字段", level=2)
for rf in merge_findings["confirmed_redundancy_fields"]:
add_styled_heading(doc, f"{rf['field']}（影响程度：{rf['risk']}）", level=3)
add_para(doc, f"出现表数：{rf['table_count']} 张")
add_para(doc, f"数据证据：{rf['evidence']}")
add_para(doc, f"建议方案：{rf['suggestion']}")
# 数据质量问题
add_styled_heading(doc, "3.5 数据质量问题", level=2)
for dq in merge_findings["data_quality_issues"]:
add_para(doc,
f"{dq['id']}: {dq['type']}（严重程度：{dq['severity']}）\n"
f" 描述：{dq['description']}\n 建议：{dq['suggestion']}",
font_size=10, space_after=8)
doc.add_page_break()
# ==================== 四、大范围空字段 ====================
add_styled_heading(doc, "四、表结构分析 — 大范围空字段", level=1)
es = empty_fields["summary"]
add_styled_heading(doc, "4.1 检查概况", level=2)
for line in [
f"扫描数据表：{es['total_tables_scanned']} 张",
f"检查字段总数：{es['total_fields_checked']} 个",
f"高空字段（空值率 ≥80%）：{es['high_empty_fields_count']} 个",
f"中空字段（空值率 ≥50% 且 <80%）：{es['mid_empty_fields_count']} 个",
f"正常字段：{es['normal_fields_count']} 个",
f"有问题表数：{es['tables_with_issues_count']} 张",
f"跳过表（数据行数 < 5，不纳入统计）：{es['skipped_tables_count']} 张",
]:
add_para(doc, f"• {line}")
# 全部高空字段分布
add_styled_heading(doc, "4.2 各表高空字段分布（空值率 ≥80%）", level=2)
top_list = es.get("top_tables_by_high_empty_fields", [])
if top_list:
headers = ["排名", "表名", "高空字段数"]
rows = []
for i, t in enumerate(top_list, 1):
rows.append([i, t["table_name"], t["high_empty_field_count"]])
add_table(doc, headers, rows, col_widths=[1, 5, 3])
# 中空字段分布
add_styled_heading(doc, "4.3 各表中空字段分布（空值率 50%~80%）", level=2)
mid_list = es.get("top_tables_by_mid_empty_fields", [])
if mid_list:
headers = ["排名", "表名", "中空字段数"]
rows = []
for i, t in enumerate(mid_list, 1):
rows.append([i, t["table_name"], t["mid_empty_field_count"]])
add_table(doc, headers, rows, col_widths=[1, 5, 3])
else:
# fallback: load from empty_field_summary_by_table
efs = empty_fields.get("empty_field_summary_by_table", [])
mid_entries = [(t["table_name"], t.get("mid_empty_count", 0)) for t in efs
if t.get("mid_empty_count", 0) > 0]
if mid_entries:
mid_entries.sort(key=lambda x: x[1], reverse=True)
headers = ["排名", "表名", "中空字段数"]
rows = []
for i, (tn, cnt) in enumerate(mid_entries, 1):
rows.append([i, tn, cnt])
add_table(doc, headers, rows, col_widths=[1, 5, 3])
add_styled_heading(doc, "4.4 关键发现", level=2)
for f in [
"t_ai_construction_plan 表有23个高空字段，为所有表中最多，说明该表大部分列未被实际使用",
"c_bri_unit 表有21个高空字段、c_bri_project 表有19个高空字段，一带一路相关表字段利用率较低",
"mall_product（10个）、mall_shop（9个）、retailers_server_user（8个）— 商城模块多表存在大量空字段",
"t_talent_archive 表有 37,155 行数据但存在13个高空字段，影响范围较大",
]:
add_para(doc, f"• {f}")
doc.add_page_break()
# ==================== 五、无注释字段 ====================
add_styled_heading(doc, "五、表结构分析 — 无注释字段", level=1)
um = uncommented["meta"]
add_styled_heading(doc, "5.1 检查概况", level=2)
for line in [
f"数据字典总字段数：{um['total_fields']} 个",
f"无注释字段数：{um['uncommented_fields']} 个（{um['uncommented_ratio']}）",
f"可推测注释：{um['inferred']} 个（{um['inferred_ratio']}）",
f"涉及表数：{uncommented['tables_affected']} 张",
]:
add_para(doc, f"• {line}")
add_styled_heading(doc, "5.2 各表无注释字段分布（全部）", level=2)
headers = ["排名", "表名", "表注释", "无注释字段数"]
rows = []
for i, t in enumerate(uncommented["by_table"], 1):
rows.append([i, t["table_name"], t["table_comment"], t["uncommented_count"]])
add_table(doc, headers, rows, col_widths=[1, 4, 5.5, 2], font_size=7.5)
add_styled_heading(doc, "5.3 推测说明", level=2)
medium_conf = [r for r in uncommented["results"] if r["confidence"] == "medium"]
add_para(doc,
f"无注释字段主要集中为以下模式：\n\n"
f"1. 审计字段遗漏（最常见）：大量表缺少 create_time（创建时间）、update_time（更新时间）、"
f"create_by（创建人）、update_by（更新人）、remark（备注）的注释。涉及 t_enterprise_credit* 系列（5张）、"
f"t_ai_* 系列（4张）、t_enterprise_performance、t_enterprise_profile、t_enterprise_risk 等表。\n\n"
f"2. 地理信息表完全无注释：c_bri_geo_boundary 表全部20个字段无注释，但字段名规范、可根据名称推测含义"
f"（如 geojson→GeoJSON地理数据、centroid_lng→中心点经度 等）。\n\n"
f"3. ID 外键字段缺少注释：部分 enterprise_id、credit_id、conversation_id 等外键字段无注释。\n\n"
f"推测置信度分布：高置信度（可直接参考采用）{um.get('inferred', 0) - len(medium_conf)} 个，"
f"中等置信度（建议人工复核）{len(medium_conf)} 个。所有推测注释详见附录C。")
doc.add_page_break()
# ==================== 六、字段存储长度超标 ====================
add_styled_heading(doc, "六、字段规范检查 — 存储长度超标", level=1)
lv_meta = length_violations["meta"]
add_styled_heading(doc, "6.1 检查概况", level=2)
for line in [
f"长度超标字段：{lv_meta['total_violations']} 个",
f"类型不匹配（经纬度等应为数值型却用字符串）：{lv_meta['total_type_mismatches']} 个",
f"日期字段使用非日期类型：{lv_meta['total_date_type_issues']} 个",
f"ID 字段使用字符串类型：{lv_meta['total_id_type_issues']} 个",
f"涉及表数：{lv_meta['tables_affected']} 张",
]:
add_para(doc, f"• {line}")
add_styled_heading(doc, "6.2 按标准分类的长度超标统计", level=2)
headers = ["标准", "超标数量"]
rows = [[s["standard"], s["violation_count"]] for s in length_violations["summary_by_standard"]]
add_table(doc, headers, rows, col_widths=[8, 3])
# 全部长度超标
add_styled_heading(doc, f"6.3 长度超标字段详情（全部 {lv_meta['total_violations']} 个）", level=2)
headers = ["表名", "字段名", "当前类型", "实际长度", "标准长度", "超出位数", "参考标准"]
rows = []
for v in length_violations["length_violations"]:
rows.append([v["table_name"], v["column_name"], v["column_type"],
v["actual_length"], v["required_length"], v["excess"], v["standard"]])
add_table(doc, headers, rows, col_widths=[3, 2.5, 2, 1.2, 1.2, 1.2, 3], font_size=7)
# 类型不匹配
add_styled_heading(doc, "6.4 类型不匹配详情", level=2)
add_styled_heading(doc, "6.4.1 经纬度字段使用字符串类型", level=3)
if length_violations.get("type_mismatches"):
headers = ["表名", "字段名", "当前类型", "建议类型", "参考标准"]
rows = []
for m in length_violations["type_mismatches"]:
rows.append([m["table_name"], m["column_name"], m["column_type"],
m["required_type"].upper(), m["standard"]])
add_table(doc, headers, rows, col_widths=[3, 2.5, 2.5, 2, 3.5], font_size=7)
else:
add_para(doc, "未发现此类型问题。")
add_styled_heading(doc, f"6.4.2 日期字段使用非日期类型（共 {lv_meta['total_date_type_issues']} 个）", level=3)
if length_violations.get("date_type_issues"):
headers = ["表名", "字段名", "当前类型", "建议"]
rows = []
for d in length_violations["date_type_issues"]:
rows.append([d["table_name"], d["column_name"], d["column_type"], "改为 DATE 或 DATETIME"])
add_table(doc, headers, rows, col_widths=[3, 2.5, 2.5, 5], font_size=7)
add_styled_heading(doc, f"6.4.3 ID 字段使用字符串类型（共 {lv_meta['total_id_type_issues']} 个）", level=3)
if length_violations.get("id_type_issues"):
headers = ["表名", "字段名", "当前类型", "建议"]
rows = []
for d in length_violations["id_type_issues"]:
rows.append([d["table_name"], d["column_name"], d["column_type"], "改为 BIGINT（如业务允许）"])
add_table(doc, headers, rows, col_widths=[3, 2.5, 2.5, 5], font_size=7)
doc.add_page_break()
# ==================== 七、数据不符合标准 ====================
add_styled_heading(doc, "七、字段规范检查 — 数据不符合标准", level=1)
sv_summary = standard_violations["summary"]
add_styled_heading(doc, "7.1 检查概况", level=2)
for line in [
f"检查字段总数：{sv_summary['total_fields_checked']} 个",
f"存在违规的字段：{sv_summary['fields_with_violations']} 个",
f"全部合规的字段：{sv_summary['fields_clean']} 个",
f"每字段抽样上限：{standard_violations['meta']['sample_limit_per_table']} 条",
]:
add_para(doc, f"• {line}")
add_styled_heading(doc, "7.2 按校验类型统计", level=2)
headers = ["校验类型", "涉及字段数", "违规记录总数"]
rows = []
for rule_type, info in sorted(sv_summary["by_rule_type"].items(),
key=lambda x: x[1]["total_violations"], reverse=True):
rows.append([rule_type, info["count"], info["total_violations"]])
add_table(doc, headers, rows, col_widths=[4, 3, 3])
# 违规字段详情 - 全部
violation_fields = standard_violations.get("violation_fields", [])
if not violation_fields:
violation_fields = [f for f in standard_violations.get("fields_checked", [])
if f.get("violation_count", 0) > 0]
violation_fields.sort(key=lambda x: x.get("violation_count", 0), reverse=True)
add_styled_heading(doc, f"7.3 违规字段详情（全部 {len(violation_fields)} 个字段）", level=2)
if violation_fields:
headers = ["表名", "字段名", "校验类型", "违规数", "抽样数", "违规率"]
rows = []
for vf in violation_fields:
total = vf.get("total_checked", vf.get("sample_count", 0))
vcount = vf.get("violation_count", 0)
rate = f"{vcount/total*100:.1f}%" if total > 0 else "N/A"
rows.append([vf.get("table_name", ""), vf.get("column_name", ""),
vf.get("rule_type", ""), vcount, total, rate])
add_table(doc, headers, rows, col_widths=[3, 2.5, 2.5, 1.5, 1.5, 1.5], font_size=7)
doc.add_page_break()
# ==================== 八、综合问题清单 ====================
add_styled_heading(doc, "八、综合问题清单", level=1)
add_para(doc, "以下汇总本次数据治理分析发现的所有问题，供参考和决策：", space_after=12)
headers = ["编号", "类别", "问题描述", "影响程度", "建议措施"]
rows = []
issues = [
("P-001", "表合并", "em_gas（8行）与 em_water_supply（7行）字段结构 100% 一致，建议合并", "高",
"合并为 em_infrastructure 表，新增 infrastructure_type 字段区分燃气/供水"),
("P-002", "表合并", "t_project_*_record 共5张表字段结构 100% 一致，合计仅7行数据", "高",
"合并为 t_project_business_record 表，新增 record_type 字段区分业务类型"),
("P-003", "表合并", "biz_history_field_label（38行）与 biz_history_focus_field（13行）功能高度重叠", "中",
"评估合并为 biz_history_field_config，新增 field_category 字段区分"),
("P-004", "两套表", "t_ 前缀人才相关表（t_course、t_exam_paper 等）共19张全部0行；同时存在 t_talent_ 前缀的13张对应表有实际数据", "高",
"建议与业务方确认两套表的定位和关系，确定哪套在用后清理不再使用的表"),
("P-005", "零行表", "11张表估算行数为0（mall_address、project_decision 等）", "中",
"建议与业务方确认实际使用情况后归档"),
("P-006", "冗余字段", "project_name 在30张表中冗余存储", "高",
"各表建议仅保留 project_id，通过关联主表获取 project_name"),
("P-007", "冗余字段", "enterprise_name 在13张表中冗余存储，t_talent_archive 表填充率仅1.7%", "高",
"各表建议仅保留 enterprise_id，通过关联主表获取 enterprise_name"),
("P-008", "冗余字段", "site_code/site_name 存在数据不一致：相同 code 对应不同 name", "高",
"建议建立 site 主表统一管理，修复已有不一致数据"),
("P-009", "冗余字段", "xzqhbm（行政区划编码）在20张表中冗余，且存在字典表之外的编码", "中",
"统一字符集排序规则后清理不在字典表中的编码，或补充字典表"),
("P-010", "两套ID", "user_id / talent_id 两套ID体系并存（14张表用 user_id，15张表用 talent_id）", "中",
"建议与业务方确认两套ID的关系，统一为一种ID体系；排查 t_talent_evaluation、t_talent_offer 的 talent_id=0 问题"),
("P-011", "空字段", f"241个高空字段（≥80%）、92个中空字段（≥50%），涉及50张表", "中",
"逐表评估是否可删除不再使用的列"),
("P-012", "空字段", "t_ai_construction_plan 有23个高空字段，字段利用率极低", "高",
"检查表设计是否过度冗余，收缩或删除不需要的列"),
("P-013", "无注释", "119个字段无注释，c_bri_geo_boundary 整表20个字段全无注释", "高",
"参考推测注释批量补充；高置信度推测可直接采用，中等置信度需人工复核"),
("P-014", "长度超标", "114个字段存储长度超出标准所需", "中",
"修改列类型为合适长度（如 VARCHAR(50)→CHAR(18)）"),
("P-015", "长度超标", "51个日期字段使用 VARCHAR 类型存储", "中",
"建议改为 DATE 或 DATETIME 类型"),
("P-016", "长度超标", "19个 ID 字段使用 VARCHAR 类型存储", "中",
"如业务允许，建议改为 BIGINT 类型以提升查询效率"),
("P-017", "数据违规", "手机号码：5个字段存在55条格式/长度违规", "高",
"前端增加格式校验，历史数据批量清洗"),
("P-018", "数据违规", "行政区划代码：10个字段存在205条违规", "高",
"校验数据格式，参照 GB/T 2260 清理非法编码"),
("P-019", "数据违规", "学历代码：2个字段存在101条违规", "中",
"参照 GB/T 4658 规范学历代码数据"),
("P-020", "数据违规", "货币代码：1个字段存在100条违规", "中",
"参照 ISO 4217 规范货币代码数据"),
("P-021", "数据质量", "em_* 和 ex_* 系列表字符集不一致（utf8mb4_unicode_ci vs utf8mb4_general_ci），导致 JOIN 报错", "中",
"统一所有表使用同一种字符集排序规则"),
("P-022", "数据质量", "t_talent_evaluation 和 t_talent_offer 的 talent_id 填充率为0%", "中",
"排查数据写入逻辑，确认是否为数据遗漏"),
]
for issue in issues:
rows.append(list(issue))
add_table(doc, headers, rows, col_widths=[1.2, 1.5, 5.5, 1, 4.5], font_size=7)
doc.add_page_break()
# ==================== 九、改进建议汇总 ====================
add_styled_heading(doc, "九、改进建议汇总", level=1)
add_styled_heading(doc, "9.1 紧急修复（建议1周内处理）", level=2)
for i, item in enumerate([
"修复 site_code/site_name 数据不一致（P-008）",
"补充 c_bri_geo_boundary 整表字段注释（P-013）",
"排查 t_ai_construction_plan 表23个高空字段（P-012）",
"手机号码数据清洗（P-017）",
"行政区划代码数据清洗（P-018）",
], 1):
add_para(doc, f"{i}. {item}")
add_styled_heading(doc, "9.2 短期优化（建议1个月内处理）", level=2)
for i, item in enumerate([
"合并 em_gas 与 em_water_supply 表（P-001）",
"合并 t_project_*_record 共5张表（P-002）",
"与业务方确认 t_ / t_talent_ 两套人才相关表的定位，确定哪套在用后清理不再使用的表（P-004）；同时确认11张零行表是否可以归档（P-005）",
"消除 project_name、enterprise_name 冗余字段（P-006、P-007）",
"修复 em_* 和 ex_* 系列表字符集不一致问题（P-021）",
"修复114个长度超标字段，缩小 VARCHAR 长度到合理范围（P-014）",
"51个日期字段改为 DATE/DATETIME 类型（P-015）",
], 1):
add_para(doc, f"{i}. {item}")
add_styled_heading(doc, "9.3 长期规范（建议纳入日常开发规范）", level=2)
for i, item in enumerate([
"字段注释强制规范：所有新建表及字段必须有中文注释",
"字段长度规范：对照国家标准设置 VARCHAR/CHAR 长度，避免不加区分地使用 VARCHAR(255) 或更大的默认值",
"数据格式校验规范：前端和后端对身份证、手机号、统一社会信用代码等标准字段进行双重格式校验",
"数据字典维护机制：定期检查无注释字段、空字段率、数据类型合理性",
"新表审批机制：由 DBA 或数据架构师审核新表设计，避免字段冗余和类型不当",
"固定数据治理工作流程：定期按照 '获取字典→分析冗余→检查空字段→检查注释→检查长度→校验数据→生成报告' 流程执行检查",
], 1):
add_para(doc, f"{i}. {item}")
doc.add_page_break()
# ==================== 十、附录 ====================
add_styled_heading(doc, "十、附录", level=1)
add_styled_heading(doc, "附录A：数据治理工作流程", level=2)
add_para(doc, "本次数据治理形成了以下可复制的工作流程，后续可按此步骤定期执行：")
add_para(doc, "")
headers = ["步骤", "任务", "操作说明"]
rows = [
["Step 1", "获取数据字典",
"连接数据库，通过 INFORMATION_SCHEMA 查询所有表和字段的元数据信息（字段名、类型、长度、注释、默认值等），保存为结构化 JSON 文件"],
["Step 2", "表合并与冗余分析",
"对比各表字段结构相似度，结合实际数据交叉验证，发现可合并的表和冗余存储的字段"],
["Step 3", "大范围空字段检查",
"逐表批量查询每个字段的空值（NULL 及空字符串）占比，识别高空字段（≥80%）和中空字段（≥50%）"],
["Step 4", "无注释字段检查",
"筛选数据字典中缺少注释的字段，基于字段名和数据类型推测可能的注释内容"],
["Step 5", "字段长度超标检查",
"将 VARCHAR/CHAR 字段长度与国家标准中规定的固定长度进行对比，发现存储长度超出标准所需的字段"],
["Step 6", "标准数据校验",
"对识别到的身份证号、手机号、统一社会信用代码、行政区划代码、学历代码、货币代码等标准字段，抽样实际数据并进行格式校验"],
["Step 7", "生成报告",
"汇总以上所有分析结果，生成 Word 格式的数据治理报告，包含问题清单和改进建议"],
]
add_table(doc, headers, rows, col_widths=[1.5, 3, 9], font_size=8)
add_para(doc, "")
add_para(doc, "所有分析程序均位于 data_dictionary/ 目录下，使用交互式密码输入，不存储数据库凭证。每次执行将结果保存为 JSON 文件，供报告生成步骤汇总使用。")
add_styled_heading(doc, "附录B：分析产出文件索引", level=2)
headers = ["分析任务", "产出结果文件"]
file_rows = [
["获取数据字典", "data_dictionary.json（全量字段信息）、table_summary.json（表级汇总）"],
["表合并与冗余分析", "findings_verified.json（分析结论）、verify_results.json（验证明细）"],
["大范围空字段检查", "empty_fields_report.json"],
["无注释字段检查", "uncommented_fields.json（含推测注释）"],
["字段长度超标检查", "field_length_violations.json"],
["标准字段数据校验", "validate_standard_fields.json（完整报告）、validate_standard_fields_flat.json（违规记录扁平列表）"],
["国家标准数据提取", "extracted_standards.json（7项标准共6,661条记录）、field_standards_reference.json（字段与标准对照表）"],
["生成Word报告", "数据治理报告_smart-build.docx（本文件）"],
]
add_table(doc, headers, file_rows, col_widths=[4.5, 9], font_size=8)
add_styled_heading(doc, "附录C：所有推测注释的字段（119个）", level=2)
add_para(doc, "高置信度 → 建议直接参考采用；中等置信度 → 建议人工复核确认", font_size=9)
headers = ["表名", "字段名", "数据类型", "推测注释", "置信度"]
rows = []
for r in uncommented["results"]:
rows.append([r["table_name"], r["column_name"], r["column_type"],
r["inferred_comment"], r["confidence"]])
add_table(doc, headers, rows, col_widths=[3, 2.5, 2, 5, 1], font_size=7)
# 保存
output_path = os.path.abspath(OUTPUT_FILE)
doc.save(output_path)
print(f"\n 报告已生成: {output_path}")
print(f" 文件大小: {os.path.getsize(output_path) / 1024:.1f} KB")
return output_path
def main():
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
output_path = generate_report()
print(f"\n{'=' * 60}")
print(f" 数据治理报告生成完成！")
print(f" 文件: {output_path}")
print(f"{'=' * 60}")
if __name__ == "__main__":
main()
data_dictionary/validate_standard_fields.py 0 → 100644
View file @ 31911f51
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据治理 - 标准字段格式校验脚本
功能：对数据库中的标准字段（身份证、手机号、统一社会信用代码等）
进行格式合规性校验，找出不合规的记录。
输出文件: validate_standard_fields.json
"""
import json
import os
import re
import sys
from datetime import datetime, date
from getpass import getpass
try:
import pymysql
except ImportError:
print("错误：缺少 pymysql 库，请执行: pip install pymysql")
sys.exit(1)
# ── 配置 ──────────────────────────────────────────────
DB_CONFIG = {
"host": "192.168.20.10",
"port": 3306,
"user": "root",
"database": "smart-build",
"charset": "utf8mb4",
}
OUTPUT_FILE = "validate_standard_fields.json"
SAMPLE_LIMIT = 100 # 每张表最多扫描行数（避免全表扫描大表）
# ── 校验规则 ──────────────────────────────────────────
# 每个规则: 字段匹配模式列表 → 校验函数
def validate_id_card(value):
"""GB 11643-1999: 身份证号码18位 + 校验码"""
if value is None or str(value).strip() == "":
return None # NULL不算违规，由空字段检查负责
v = str(value).strip().upper()
if len(v) != 18:
return f"长度错误：{len(v)}位（应为18位）"
if not re.match(r'^[1-9]\d{5}(19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[\dX]$', v):
return f"格式不符：{v}（前17位需为数字，末位数字或X，出生日期需有效）"
# 校验码
weights = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
check_map = "10X98765432"
total = sum(int(v[i]) * weights[i] for i in range(17))
expected = check_map[total % 11]
if v[17] != expected:
return f"校验码错误：期望{expected}，实际{v[17]}"
return None
def validate_mobile(value):
"""手机号码：1开头 + 11位数字"""
if value is None or str(value).strip() == "":
return None
v = str(value).strip().replace(" ", "").replace("-", "")
if len(v) != 11:
return f"长度错误：{len(v)}位（应为11位）"
if not re.match(r'^1[3-9]\d{9}$', v):
return f"格式不符：{v}（应为1开头的11位数字）"
return None
def validate_uscc(value):
"""GB 32100-2015: 统一社会信用代码18位"""
if value is None or str(value).strip() == "":
return None
v = str(value).strip().upper()
if len(v) != 18:
return f"长度错误：{len(v)}位（应为18位）"
# 不应包含 I O Z S V
invalid_chars = set("IOZSV")
found = [c for c in v if c in invalid_chars]
if found:
return f"含禁用字符：{found}（统一社会信用代码不含I/O/Z/S/V）"
if not re.match(r'^[0-9A-HJ-NP-RT-Y]{18}$', v):
return f"格式不符：{v}"
return None
def validate_xzqhbm(value):
"""GB/T 2260: 行政区划代码6位或12位数字"""
if value is None or str(value).strip() == "":
return None
v = str(value).strip()
if not re.match(r'^\d+$', v):
return f"含非数字字符：{v}"
if len(v) not in (6, 12):
return f"长度错误：{len(v)}位（应为6位或12位）"
return None
def validate_gender(value):
"""性别：应为 0/1/2 或 男/女 或 M/F"""
if value is None or str(value).strip() == "":
return None
v = str(value).strip()
valid_codes = {"0", "1", "2", "9", "男", "女", "M", "F", "m", "f", "未知", "未说明"}
if v not in valid_codes and not re.match(r'^[0129]$', v):
return f"非标准性别值：'{v}'（标准值为0/1/2/9或男/女）"
return None
def validate_nation(value):
"""民族代码：2位数字 01-56"""
if value is None or str(value).strip() == "":
return None
v = str(value).strip()
if not re.match(r'^\d{2}$', v):
# 可能存的是中文名 如"汉族"
if v in ("汉族", "蒙古族", "回族", "藏族", "维吾尔族", "苗族", "彝族", "壮族",
"布依族", "朝鲜族", "满族", "侗族", "瑶族", "白族", "土家族",
"哈尼族", "哈萨克族", "傣族", "黎族", "傈僳族", "佤族", "畲族",
"高山族", "拉祜族", "水族", "东乡族", "纳西族", "景颇族",
"柯尔克孜族", "土族", "达斡尔族", "仫佬族", "羌族", "布朗族",
"撒拉族", "毛南族", "仡佬族", "锡伯族", "阿昌族", "普米族",
"塔吉克族", "怒族", "乌孜别克族", "俄罗斯族", "鄂温克族",
"德昂族", "保安族", "裕固族", "京族", "塔塔尔族", "独龙族",
"鄂伦春族", "赫哲族", "门巴族", "珞巴族", "基诺族"):
return "存储为中文名而非数字代码（非严格违规，建议统一为代码）"
return f"非标准民族代码：'{v}'"
code = int(v)
if code < 1 or code > 56:
return f"代码超出范围：{v}（应为01-56）"
return None
def validate_education(value):
"""学历代码：2位数字"""
if value is None or str(value).strip() == "":
return None
v = str(value).strip()
# 常见中文存储
cn_edu = {"博士", "硕士", "本科", "大专", "中专", "高中", "初中", "小学",
"研究生", "博士研究生", "硕士研究生", "大学本科", "大学专科"}
if v in cn_edu:
return "存储为中文名而非数字代码（建议统一为GB/T 4658代码）"
if not re.match(r'^\d{2}$', v):
return f"非标准学历代码：'{v}'"
return None
def validate_country_iso3(value):
"""ISO 3166-1 alpha-3: 3位大写字母"""
if value is None or str(value).strip() == "":
return None
v = str(value).strip().upper()
if len(v) != 3:
return f"长度错误：{len(v)}位（应为3位）"
if not re.match(r'^[A-Z]{3}$', v):
return f"格式不符：{v}（应为3位大写字母）"
return None
def validate_currency(value):
"""ISO 4217: 3位大写字母"""
if value is None or str(value).strip() == "":
return None
v = str(value).strip().upper()
if len(v) != 3:
return f"长度错误：{len(v)}位（应为3位）"
if not re.match(r'^[A-Z]{3}$', v):
return f"格式不符：{v}（应为3位大写字母）"
return None
def validate_date_format(value):
"""日期格式：YYYY-MM-DD 或 YYYY-MM-DD HH:MM:SS"""
if value is None or str(value).strip() == "":
return None
v = str(value).strip()
# 尝试多种格式
for fmt in ["%Y-%m-%d", "%Y-%m-%d %H:%M:%S", "%Y/%m/%d", "%Y%m%d"]:
try:
datetime.strptime(v, fmt)
return None
except ValueError:
continue
return f"无效日期格式：'{v}'"
def validate_lng(value):
"""经度：-180 ~ 180"""
if value is None or str(value).strip() == "":
return None
try:
v = float(value)
except (ValueError, TypeError):
return f"非数值：'{value}'"
if v < -180 or v > 180:
return f"超出范围：{v}（应为-180~180）"
return None
def validate_lat(value):
"""纬度：-90 ~ 90"""
if value is None or str(value).strip() == "":
return None
try:
v = float(value)
except (ValueError, TypeError):
return f"非数值：'{value}'"
if v < -90 or v > 90:
return f"超出范围：{v}（应为-90~90）"
return None
def validate_email(value):
"""邮箱基本格式"""
if value is None or str(value).strip() == "":
return None
v = str(value).strip()
if not re.match(r'^[^@\s]+@[^@\s]+\.[^@\s]+$', v):
return f"邮箱格式不符：'{v}'"
return None
def validate_year(value):
"""年份：4位数字"""
if value is None or str(value).strip() == "":
return None
v = str(value).strip()
if not re.match(r'^\d{4}$', v):
return f"年份格式不符：'{v}'"
return None
# ── 字段名 → 校验规则 映射 ─────────────────────────────
FIELD_RULES = {
# 精确匹配
"id_card": ("身份证号码", validate_id_card),
"id_card_no": ("身份证号码", validate_id_card),
"mobile": ("手机号码", validate_mobile),
"contact_phone": ("手机号码", validate_mobile),
"customer_mobile": ("手机号码", validate_mobile),
"legal_person_mobile": ("手机号码", validate_mobile),
"notify_phone": ("手机号码", validate_mobile),
"maintainer_phone":("手机号码", validate_mobile),
"manager_phone": ("手机号码", validate_mobile),
"receiver_phone": ("手机号码", validate_mobile),
"phone": ("手机号码", validate_mobile),
"uscc": ("统一社会信用代码", validate_uscc),
"credit_code": ("统一社会信用代码", validate_uscc),
"xzqhbm": ("行政区划代码", validate_xzqhbm),
"xzqhjb": ("行政区划代码", validate_xzqhbm),
"sjxzqhbm": ("行政区划代码", validate_xzqhbm),
"adcode": ("行政区划代码", validate_xzqhbm),
"district_code": ("行政区划代码", validate_xzqhbm),
"region_code": ("行政区划代码", validate_xzqhbm),
"gender": ("性别", validate_gender),
"sex": ("性别", validate_gender),
"nation": ("民族", validate_nation),
"ethnicity": ("民族", validate_nation),
"education": ("学历", validate_education),
"education_level": ("学历", validate_education),
"educate_level": ("学历", validate_education),
"country_iso3": ("国家代码ISO3", validate_country_iso3),
"currency": ("货币代码", validate_currency),
"email": ("邮箱", validate_email),
"lng": ("经度", validate_lng),
"longitude": ("经度", validate_lng),
"lat": ("纬度", validate_lat),
"latitude": ("纬度", validate_lat),
"centroid_lng": ("经度", validate_lng),
"centroid_lat": ("纬度", validate_lat),
}
# 日期字段名后缀匹配（VARCHAR类型的日期字段）
DATE_PATTERNS = [
r".*_date$", r".*_time$", "birth_date", "date_of_birth",
"established_date", "create_time", "update_time",
"issue_date", "expire_date", "start_date", "end_date",
"start_time", "end_time",
]
# year 字段
YEAR_FIELDS = ["year", "build_year", "stat_year", "work_year", "work_years"]
# ── 数据库操作 ────────────────────────────────────────
def get_db_password():
print("=" * 50)
print(f" 连接目标: {DB_CONFIG['user']}@{DB_CONFIG['host']}:{DB_CONFIG['port']}")
print(f" 数据库: {DB_CONFIG['database']}")
print("=" * 50)
password = getpass("请输入数据库密码（输入不显示）: ")
if not password:
print("错误：密码不能为空")
sys.exit(1)
return password
def connect_db(password):
try:
conn = pymysql.connect(
host=DB_CONFIG["host"], port=DB_CONFIG["port"],
user=DB_CONFIG["user"], password=password,
database=DB_CONFIG["database"], charset=DB_CONFIG["charset"],
connect_timeout=10,
)
print("✓ 数据库连接成功\n")
return conn
except pymysql.err.OperationalError as e:
print(f"错误：无法连接数据库 — {e}")
sys.exit(1)
def query_all(cursor, sql, params=None):
cursor.execute(sql, params)
cols = [desc[0] for desc in cursor.description]
return [dict(zip(cols, row)) for row in cursor.fetchall()]
def default_serializer(obj):
if hasattr(obj, "isoformat"):
return obj.isoformat()
if isinstance(obj, bytes):
return obj.decode("utf-8", errors="replace")
return str(obj)
# ── 获取候选字段 ──────────────────────────────────────
def find_candidate_fields(cursor):
"""从数据字典查找所有需要校验的字段（仅 VARCHAR/CHAR/TEXT 类型且有数据的表）"""
candidates = []
# 1. 固定规则字段
field_names = list(FIELD_RULES.keys())
placeholders = ",".join(["%s"] * len(field_names))
rows = query_all(
cursor,
f"""
SELECT c.TABLE_NAME, c.COLUMN_NAME, c.DATA_TYPE, c.COLUMN_TYPE,
c.CHARACTER_MAXIMUM_LENGTH, c.COLUMN_COMMENT, t.TABLE_ROWS
FROM INFORMATION_SCHEMA.COLUMNS c
JOIN INFORMATION_SCHEMA.TABLES t
ON c.TABLE_SCHEMA = t.TABLE_SCHEMA AND c.TABLE_NAME = t.TABLE_NAME
WHERE c.TABLE_SCHEMA = %s
AND c.COLUMN_NAME IN ({placeholders})
AND t.TABLE_ROWS > 0
ORDER BY c.TABLE_NAME, c.COLUMN_NAME
""",
(DB_CONFIG["database"], *field_names),
)
for r in rows:
r["rule_type"] = FIELD_RULES[r["COLUMN_NAME"]][0]
r["validator"] = FIELD_RULES[r["COLUMN_NAME"]][1]
candidates.append(r)
# 2. 日期字段（VARCHAR类型）
date_rows = query_all(
cursor,
"""
SELECT c.TABLE_NAME, c.COLUMN_NAME, c.DATA_TYPE, c.COLUMN_TYPE,
c.CHARACTER_MAXIMUM_LENGTH, c.COLUMN_COMMENT, t.TABLE_ROWS
FROM INFORMATION_SCHEMA.COLUMNS c
JOIN INFORMATION_SCHEMA.TABLES t
ON c.TABLE_SCHEMA = t.TABLE_SCHEMA AND c.TABLE_NAME = t.TABLE_NAME
WHERE c.TABLE_SCHEMA = %s
AND c.DATA_TYPE IN ('varchar', 'char', 'text', 'tinytext', 'mediumtext', 'longtext')
AND t.TABLE_ROWS > 0
ORDER BY c.TABLE_NAME, c.COLUMN_NAME
""",
(DB_CONFIG["database"],),
)
for r in date_rows:
cname = r["COLUMN_NAME"].lower()
for pat in DATE_PATTERNS:
if re.match(pat, cname, re.IGNORECASE):
r["rule_type"] = "日期格式"
r["validator"] = validate_date_format
candidates.append(r)
break
# year 字段
if cname in YEAR_FIELDS:
r["rule_type"] = "年份"
r["validator"] = validate_year
candidates.append(r)
return candidates
# ── 执行校验 ──────────────────────────────────────────
def validate_field(cursor, table_name, column_name, validator, rule_type):
"""对单个字段采样并校验"""
# 先确认表存在、有数据
try:
count_row = query_all(cursor, f"SELECT COUNT(*) AS cnt FROM `{table_name}`")
total = count_row[0]["cnt"] if count_row else 0
except Exception:
return {"error": "表不存在或无权限", "total_rows": 0, "violations": []}
if total == 0:
return {"total_rows": 0, "violations": [], "note": "表无数据"}
# 采样：取非空值，限制行数
try:
sample = query_all(
cursor,
f"SELECT `{column_name}` AS val FROM `{table_name}` "
f"WHERE `{column_name}` IS NOT NULL AND `{column_name}` != '' "
f"LIMIT {SAMPLE_LIMIT}",
)
except Exception as e:
return {"error": str(e), "total_rows": total, "violations": []}
violations = []
for row in sample:
val = row["val"]
error = validator(val)
if error:
violations.append({
"value": str(val)[:100], # 截断过长值
"error": error,
})
# 统计 NULL 率
try:
null_row = query_all(
cursor,
f"SELECT COUNT(*) AS total, "
f"SUM(CASE WHEN `{column_name}` IS NULL OR `{column_name}` = '' THEN 1 ELSE 0 END) AS empty "
f"FROM `{table_name}`",
)
total_rows = null_row[0]["total"] if null_row else total
empty_rows = null_row[0]["empty"] if null_row else 0
except Exception:
total_rows = total
empty_rows = 0
return {
"total_rows": total_rows,
"empty_rows": empty_rows,
"empty_rate": round(empty_rows / total_rows, 4) if total_rows > 0 else 0,
"sampled": len(sample),
"violation_count": len(violations),
"violation_rate": round(len(violations) / len(sample), 4) if sample else 0,
"violations": violations, # 全部违规记录
}
# ── 主流程 ────────────────────────────────────────────
def main():
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
password = get_db_password()
conn = connect_db(password)
try:
with conn.cursor() as cursor:
print("正在查找候选校验字段...")
candidates = find_candidate_fields(cursor)
# 按表+字段名去重
seen = set()
unique = []
for c in candidates:
key = (c["TABLE_NAME"], c["COLUMN_NAME"], c["rule_type"])
if key not in seen:
seen.add(key)
unique.append(c)
print(f"共 {len(unique)} 个字段待校验\n")
results = []
has_violations = []
no_issues = []
for i, cand in enumerate(unique):
tname = cand["TABLE_NAME"]
cname = cand["COLUMN_NAME"]
rtype = cand["rule_type"]
if i % 20 == 0:
print(f" 进度: {i + 1}/{len(unique)}")
result = validate_field(
cursor, tname, cname,
cand["validator"], rtype,
)
result["table_name"] = tname
result["column_name"] = cname
result["rule_type"] = rtype
result["column_type"] = cand.get("COLUMN_TYPE", "")
result["column_comment"] = cand.get("COLUMN_COMMENT", "")
results.append(result)
if result.get("error"):
continue
if result.get("violation_count", 0) > 0:
has_violations.append(result)
elif result.get("total_rows", 0) > 0 and result.get("sampled", 0) > 0:
no_issues.append(result)
# ── 汇总 ──
summary = {
"total_fields_checked": len(results),
"fields_with_violations": len(has_violations),
"fields_clean": len(no_issues),
"by_rule_type": {},
}
for r in has_violations:
rt = r["rule_type"]
if rt not in summary["by_rule_type"]:
summary["by_rule_type"][rt] = {"count": 0, "total_violations": 0}
summary["by_rule_type"][rt]["count"] += 1
summary["by_rule_type"][rt]["total_violations"] += r.get("violation_count", 0)
output = {
"meta": {
"script": "validate_standard_fields.py",
"database": DB_CONFIG["database"],
"executed_at": datetime.now().isoformat(),
"sample_limit_per_table": SAMPLE_LIMIT,
},
"summary": summary,
"violations": has_violations,
"clean_fields": [
{"table_name": r["table_name"], "column_name": r["column_name"],
"rule_type": r["rule_type"], "total_rows": r.get("total_rows", 0)}
for r in no_issues
],
"all_results": results,
}
# 保存完整报告
script_dir = os.path.dirname(os.path.abspath(__file__))
out_path = os.path.join(script_dir, OUTPUT_FILE)
with open(out_path, "w", encoding="utf-8") as f:
json.dump(output, f, ensure_ascii=False, indent=2, default=default_serializer)
print(f"\n✓ 完整报告已保存至: {out_path}")
# ── 单独提取所有违规记录存为扁平文件 ──
flat_violations = []
for v in has_violations:
for item in v.get("violations", []):
flat_violations.append({
"table_name": v["table_name"],
"column_name": v["column_name"],
"rule_type": v["rule_type"],
"column_type": v.get("column_type", ""),
"value": item["value"],
"error": item["error"],
})
flat_path = os.path.join(script_dir, "validate_standard_fields_flat.json")
with open(flat_path, "w", encoding="utf-8") as f:
json.dump(flat_violations, f, ensure_ascii=False, indent=2, default=default_serializer)
print(f"✓ 违规记录已单独保存至: {flat_path} ({len(flat_violations)} 条)")
# 控制台输出
print(f"\n{'=' * 50}")
print(f" 格式校验完成")
print(f"{'=' * 50}")
print(f" 检查字段数: {summary['total_fields_checked']}")
print(f" 有违规字段: {summary['fields_with_violations']}")
print(f" 全部合规字段: {summary['fields_clean']}")
print(f"\n 按类型违规明细:")
for rt, info in sorted(summary["by_rule_type"].items(),
key=lambda x: x[1]["total_violations"], reverse=True):
print(f" {rt}: {info['count']}个字段, {info['total_violations']}条违规记录")
print(f"{'=' * 50}\n")
finally:
conn.close()
print("数据库连接已关闭。")
if __name__ == "__main__":
main()
data_dictionary/verify_merge_redundancy.py 0 → 100644
View file @ 31911f51
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据治理 - 表合并与冗余字段验证脚本
功能：执行一系列数据验证查询，将结果以 JSON 格式存储，
用于验证数据字典分析中发现的合并候选和冗余问题。
输出文件: verify_results.json
"""
import json
import os
import sys
from datetime import datetime
from getpass import getpass
try:
import pymysql
except ImportError:
print("错误：缺少 pymysql 库，请执行: pip install pymysql")
sys.exit(1)
# ── 配置 ──────────────────────────────────────────────
DB_CONFIG = {
"host": "192.168.20.10",
"port": 3306,
"user": "root",
"database": "smart-build",
"charset": "utf8mb4",
}
OUTPUT_FILE = "verify_results.json"
# ── 获取密码 ──────────────────────────────────────────
def get_db_password():
print("=" * 50)
print(f" 连接目标: {DB_CONFIG['user']}@{DB_CONFIG['host']}:{DB_CONFIG['port']}")
print(f" 数据库: {DB_CONFIG['database']}")
print("=" * 50)
password = getpass("请输入数据库密码（输入不显示）: ")
if not password:
print("错误：密码不能为空")
sys.exit(1)
return password
# ── 连接数据库 ────────────────────────────────────────
def connect_db(password):
try:
conn = pymysql.connect(
host=DB_CONFIG["host"],
port=DB_CONFIG["port"],
user=DB_CONFIG["user"],
password=password,
database=DB_CONFIG["database"],
charset=DB_CONFIG["charset"],
connect_timeout=10,
)
print("✓ 数据库连接成功\n")
return conn
except pymysql.err.OperationalError as e:
print(f"错误：无法连接数据库 — {e}")
sys.exit(1)
# ── 辅助函数 ──────────────────────────────────────────
def query_all(cursor, sql, params=None):
"""执行查询并返回所有行（字典格式）"""
cursor.execute(sql, params)
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
return [dict(zip(columns, row)) for row in rows]
def query_single(cursor, sql, params=None):
"""执行查询并返回单个值"""
cursor.execute(sql, params)
row = cursor.fetchone()
return row[0] if row else None
def default_serializer(obj):
"""处理 datetime 等 JSON 不可序列化的类型"""
if hasattr(obj, "isoformat"):
return obj.isoformat()
if isinstance(obj, bytes):
return obj.decode("utf-8", errors="replace")
return str(obj)
# ── 一、结构完全相同表 - 数据验证 ─────────────────────
def verify_merge_001_gas_water(cursor):
"""MERGE-001: em_gas vs em_water_supply 数据对比"""
print(" 1.1 em_gas vs em_water_supply 数据量对比...")
rows = query_all(
cursor,
"SELECT 'em_gas' AS table_name, COUNT(*) AS row_count FROM em_gas "
"UNION ALL "
"SELECT 'em_water_supply' AS table_name, COUNT(*) AS row_count FROM em_water_supply",
)
samples_em_gas = query_all(cursor, "SELECT * FROM em_gas LIMIT 3")
samples_em_water_supply = query_all(cursor, "SELECT * FROM em_water_supply LIMIT 3")
return {
"row_counts": rows,
"sample_em_gas": samples_em_gas,
"sample_em_water_supply": samples_em_water_supply,
}
def verify_merge_002_project_records(cursor):
"""MERGE-002: 5张 t_project_*_record 表"""
print(" 1.2 t_project_*_record ×5 数据量对比...")
tables = [
"t_project_acceptance_record",
"t_project_construction_record",
"t_project_design_record",
"t_project_general_record",
"t_project_operation_record",
]
counts = []
for t in tables:
cnt = query_single(cursor, f"SELECT COUNT(*) FROM `{t}`")
counts.append({"table_name": t, "row_count": cnt})
# 每张表取3条样本
samples = {}
for t in tables:
try:
samples[t] = query_all(cursor, f"SELECT * FROM `{t}` LIMIT 3")
except Exception as e:
samples[t] = {"error": str(e)}
return {"row_counts": counts, "sample_data": samples}
def verify_merge_003_biz_history(cursor):
"""MERGE-003: biz_history_field_label vs biz_history_focus_field"""
print(" 1.3 biz_history_field_label vs biz_history_focus_field...")
counts = query_all(
cursor,
"SELECT 'biz_history_field_label' AS table_name, COUNT(*) AS row_count FROM biz_history_field_label "
"UNION ALL "
"SELECT 'biz_history_focus_field', COUNT(*) FROM biz_history_focus_field",
)
sample_label = query_all(cursor, "SELECT * FROM biz_history_field_label LIMIT 5")
sample_focus = query_all(cursor, "SELECT * FROM biz_history_focus_field LIMIT 5")
return {
"row_counts": counts,
"sample_biz_history_field_label": sample_label,
"sample_biz_history_focus_field": sample_focus,
}
# ── 二、旧版系统 vs 新版系统 ──────────────────────────
def verify_old_vs_new_talent(cursor):
"""对比旧版 t_* 表和新版 t_talent_* 表的数据量"""
print(" 2.1 旧版系统表数据量...")
old_tables = [
"t_course", "t_course_chapter", "t_course_review",
"t_exam_paper", "t_exam_question", "t_exam_record",
"t_interview", "t_job_application", "t_job_post",
"t_learning_progress", "t_offer_record", "t_resume",
"t_title_application", "t_title_material", "t_title_policy",
"t_training_certificate", "t_certificate_template",
"t_cert_verify_cache", "t_cert_verify_record",
]
old_counts = []
for t in old_tables:
cnt = query_single(cursor, f"SELECT COUNT(*) FROM `{t}`")
old_counts.append({"table_name": t, "row_count": cnt})
print(" 2.2 新版人才系统表数据量...")
new_tables = [
"t_talent_course", "t_talent_course_chapter",
"t_talent_exam", "t_talent_exam_record",
"t_talent_interview", "t_talent_application",
"t_talent_job", "t_talent_offer",
"t_talent_resume", "t_talent_title_application",
"t_talent_title_policy", "t_talent_e_certificate",
"t_talent_cert_verify_log",
]
new_counts = []
for t in new_tables:
cnt = query_single(cursor, f"SELECT COUNT(*) FROM `{t}`")
new_counts.append({"table_name": t, "row_count": cnt})
return {"old_system_counts": old_counts, "new_system_counts": new_counts}
# ── 三、弱废弃候选表 ──────────────────────────────────
def verify_weak_decommission(cursor):
"""ROWS=0 且无替代表的确认"""
print(" 3. 弱废弃候选表数据量确认...")
tables = [
"mall_address", "mall_favorite",
"mall_freight_template", "mall_freight_template_rule",
"mall_inventory_log", "project_decision",
"project_decision_attach", "project_decision_field",
"project_design", "t_ai_llm_provider",
"t_ai_plan_optimize", "c_gp_catalog",
]
results = []
for t in tables:
cnt = query_single(cursor, f"SELECT COUNT(*) FROM `{t}`")
results.append({"table_name": t, "row_count": cnt})
return results
# ── 四、project_name 冗余验证 ─────────────────────────
def verify_project_name_redundancy(cursor):
"""验证 project_name 在不同表中是否一致（动态探测）"""
db = DB_CONFIG["database"]
# 1. 查找所有含 project_name 字段且有数据的表
print(" 4.1 查找所有含 project_name 的表...")
tables_with_project_name = query_all(
cursor,
"""
SELECT c.TABLE_NAME, t.TABLE_ROWS
FROM INFORMATION_SCHEMA.COLUMNS c
JOIN INFORMATION_SCHEMA.TABLES t
ON c.TABLE_SCHEMA = t.TABLE_SCHEMA AND c.TABLE_NAME = t.TABLE_NAME
WHERE c.TABLE_SCHEMA = %s AND c.COLUMN_NAME = 'project_name'
ORDER BY t.TABLE_ROWS DESC
""",
(db,),
)
# 2. 动态查找同时含 project_id 和 project_name 且有数据的表，两两交叉验证
print(" 4.2 动态交叉验证 project_name 一致性...")
candidates = query_all(
cursor,
"""
SELECT c.TABLE_NAME, t.TABLE_ROWS
FROM INFORMATION_SCHEMA.COLUMNS c
JOIN INFORMATION_SCHEMA.TABLES t
ON c.TABLE_SCHEMA = t.TABLE_SCHEMA AND c.TABLE_NAME = t.TABLE_NAME
WHERE c.TABLE_SCHEMA = %s
AND c.COLUMN_NAME IN ('project_id', 'id')
AND t.TABLE_ROWS > 0
ORDER BY t.TABLE_ROWS DESC
""",
(db,),
)
# 取前两个有 project_id+project_name 且有数据的表做交叉对比
consistency = []
tables_with_both = query_all(
cursor,
"""
SELECT a.TABLE_NAME AS table_a, b.TABLE_NAME AS table_b
FROM (
SELECT TABLE_NAME FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = %s AND COLUMN_NAME = 'project_id'
) a
INNER JOIN (
SELECT TABLE_NAME FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = %s AND COLUMN_NAME = 'project_name'
) b ON a.TABLE_NAME = b.TABLE_NAME
""",
(db, db),
)
if len(tables_with_both) >= 1:
# 取第一张表做自我一致性检查：project_name 是否有多种写法对应同一个 project_id
t0 = tables_with_both[0]["table_a"]
try:
consistency = query_all(
cursor,
f"""
SELECT project_id, project_name, COUNT(DISTINCT project_name) AS name_variants
FROM `{t0}`
WHERE project_id IS NOT NULL
GROUP BY project_id
HAVING COUNT(DISTINCT project_name) > 1
LIMIT 20
""",
)
except Exception:
pass
return {
"tables_with_project_name": tables_with_project_name,
"tables_with_both_id_and_name": tables_with_both,
"consistency_check": consistency,
"inconsistent_count": len(consistency) if consistency else 0,
}
# ── 五、enterprise_name 冗余验证 ──────────────────────
def verify_enterprise_name_redundancy(cursor):
"""验证 enterprise_name 在各表中的填充率"""
print(" 5. enterprise_name 冗余验证...")
# 查找所有含 enterprise_name 字段的表
tables = query_all(
cursor,
"""
SELECT
c.TABLE_NAME,
t.TABLE_ROWS
FROM INFORMATION_SCHEMA.COLUMNS c
JOIN INFORMATION_SCHEMA.TABLES t
ON c.TABLE_SCHEMA = t.TABLE_SCHEMA
AND c.TABLE_NAME = t.TABLE_NAME
WHERE c.TABLE_SCHEMA = %s
AND c.COLUMN_NAME = 'enterprise_name'
ORDER BY c.TABLE_NAME
""",
(DB_CONFIG["database"],),
)
fill_rate_results = []
for row in tables:
tname = row["TABLE_NAME"]
try:
stats = query_all(
cursor,
f"""
SELECT
COUNT(*) AS total_rows,
SUM(CASE WHEN enterprise_name IS NOT NULL AND enterprise_name != '' THEN 1 ELSE 0 END) AS filled_rows
FROM `{tname}`
""",
)
if stats and stats[0]["total_rows"] > 0:
stats[0]["table_name"] = tname
stats[0]["fill_rate_pct"] = round(
stats[0]["filled_rows"] / stats[0]["total_rows"] * 100, 1
)
else:
stats = [{"table_name": tname, "total_rows": 0, "filled_rows": 0, "fill_rate_pct": 0}]
fill_rate_results.append(stats[0])
except Exception as e:
fill_rate_results.append({"table_name": tname, "error": str(e)})
return fill_rate_results
# ── 六、xzqhbm 冗余验证 ───────────────────────────────
def verify_xzqhbm_redundancy(cursor):
"""验证 xzqhbm 是否存在字典表中不存在的值"""
print(" 6. xzqhbm 行政区划编码验证...")
# 字典表行数
dict_count = query_single(cursor, "SELECT COUNT(*) FROM c_bri_xzqh")
dict_sample = query_all(cursor, "SELECT * FROM c_bri_xzqh LIMIT 10")
# 查找所有含 xzqhbm 字段且有数据的表
tables_with_xzqhbm = query_all(
cursor,
"""
SELECT
c.TABLE_NAME,
t.TABLE_ROWS
FROM INFORMATION_SCHEMA.COLUMNS c
JOIN INFORMATION_SCHEMA.TABLES t
ON c.TABLE_SCHEMA = t.TABLE_SCHEMA
AND c.TABLE_NAME = t.TABLE_NAME
WHERE c.TABLE_SCHEMA = %s
AND c.COLUMN_NAME = 'xzqhbm'
ORDER BY t.TABLE_ROWS DESC
""",
(DB_CONFIG["database"],),
)
# 每张含 xzqhbm 的表，检查有多少值不在字典表中
orphan_check = []
for row in tables_with_xzqhbm:
tname = row["TABLE_NAME"]
if row["TABLE_ROWS"] == 0:
orphan_check.append({
"table_name": tname,
"row_count": 0,
"orphan_count": 0,
"note": "表无数据，跳过检查",
})
continue
try:
orphan = query_all(
cursor,
f"""
SELECT g.xzqhbm
FROM `{tname}` g
LEFT JOIN c_bri_xzqh x ON g.xzqhbm = x.xzqhbm
WHERE g.xzqhbm IS NOT NULL
AND g.xzqhbm != ''
AND x.xzqhbm IS NULL
LIMIT 20
""",
)
orphan_check.append({
"table_name": tname,
"row_count": row["TABLE_ROWS"],
"orphan_count": len(orphan),
"orphan_samples": orphan[:10],
"note": "存在字典表外的编码" if orphan else "全部编码都在字典表中",
})
except Exception as e:
orphan_check.append({"table_name": tname, "error": str(e)})
return {
"dict_table_row_count": dict_count,
"dict_sample": dict_sample,
"tables_with_xzqhbm": tables_with_xzqhbm,
"orphan_check": orphan_check,
}
# ── 七、site_code / site_name 冗余验证 ────────────────
def verify_site_code_redundancy(cursor):
"""验证 site_code/site_name 冗余（动态探测）"""
db = DB_CONFIG["database"]
# 1. 查找所有含 site_code 字段的表
print(" 7.1 查找所有含 site_code 的表...")
site_tables = query_all(
cursor,
"""
SELECT c.TABLE_NAME, t.TABLE_ROWS
FROM INFORMATION_SCHEMA.COLUMNS c
JOIN INFORMATION_SCHEMA.TABLES t
ON c.TABLE_SCHEMA = t.TABLE_SCHEMA AND c.TABLE_NAME = t.TABLE_NAME
WHERE c.TABLE_SCHEMA = %s AND c.COLUMN_NAME = 'site_code'
ORDER BY t.TABLE_ROWS DESC
""",
(db,),
)
site_counts = []
for row in site_tables:
tname = row["TABLE_NAME"]
cnt = query_single(cursor, f"SELECT COUNT(*) FROM `{tname}`")
site_counts.append({"table_name": tname, "row_count": cnt})
# 2. 动态交叉验证：取有数据的前两张表，检查 site_code 对应的 site_name 是否一致
print(" 7.2 site_code 跨表 site_name 一致性验证...")
inconsistency = []
tables_with_data = [r for r in site_tables if r["TABLE_ROWS"] > 0]
if len(tables_with_data) >= 2:
a, b = tables_with_data[0]["TABLE_NAME"], tables_with_data[1]["TABLE_NAME"]
try:
inconsistency = query_all(
cursor,
f"""
SELECT a.site_code, a.site_name AS name_in_a, b.site_name AS name_in_b
FROM `{a}` a
INNER JOIN `{b}` b ON a.site_code = b.site_code
WHERE a.site_name != b.site_name
LIMIT 20
""",
)
except Exception as e:
inconsistency = [{"error": str(e), "tables_checked": [a, b]}]
return {
"site_table_counts": site_counts,
"site_name_inconsistency": inconsistency,
"inconsistency_count": len(inconsistency) if inconsistency else 0,
}
# ── 八、无注释表检查 ──────────────────────────────────
def verify_empty_comment_tables(cursor):
"""检查 c_bri_geo_boundary 等无注释或注释缺失的表"""
print(" 8. 无注释表及字段检查...")
# 所有表注释为空的表
empty_comment_tables = query_all(
cursor,
"""
SELECT TABLE_NAME, TABLE_TYPE, ENGINE, TABLE_ROWS, TABLE_COMMENT
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = %s
AND (TABLE_COMMENT IS NULL OR TABLE_COMMENT = '')
ORDER BY TABLE_NAME
""",
(DB_CONFIG["database"],),
)
# 对每张无注释的表，列出字段
table_field_details = {}
for t in empty_comment_tables:
tname = t["TABLE_NAME"]
fields = query_all(
cursor,
"""
SELECT
COLUMN_NAME, COLUMN_TYPE, COLUMN_COMMENT,
IS_NULLABLE, COLUMN_DEFAULT, ORDINAL_POSITION
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s
ORDER BY ORDINAL_POSITION
""",
(DB_CONFIG["database"], tname),
)
table_field_details[tname] = fields
# 统计所有表中字段注释为空的情况
empty_field_comment_count = query_all(
cursor,
"""
SELECT
TABLE_NAME,
COUNT(*) AS empty_comment_fields
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = %s
AND (COLUMN_COMMENT IS NULL OR COLUMN_COMMENT = '')
GROUP BY TABLE_NAME
ORDER BY empty_comment_fields DESC
""",
(DB_CONFIG["database"],),
)
return {
"empty_table_comment": empty_comment_tables,
"empty_table_field_details": table_field_details,
"empty_field_comment_by_table": empty_field_comment_count,
}
# ── 九、user_id vs talent_id 两套ID体系 ───────────────
def verify_dual_id_system(cursor):
"""验证 user_id 和 talent_id 并存"""
print(" 9. user_id vs talent_id 双ID体系验证...")
# 查找所有含 user_id 字段且有数据的表
user_id_tables = query_all(
cursor,
"""
SELECT
c.TABLE_NAME,
t.TABLE_ROWS
FROM INFORMATION_SCHEMA.COLUMNS c
JOIN INFORMATION_SCHEMA.TABLES t
ON c.TABLE_SCHEMA = t.TABLE_SCHEMA
AND c.TABLE_NAME = t.TABLE_NAME
WHERE c.TABLE_SCHEMA = %s
AND c.COLUMN_NAME = 'user_id'
ORDER BY t.TABLE_ROWS DESC
""",
(DB_CONFIG["database"],),
)
# 查找所有含 talent_id 字段且有数据的表
talent_id_tables = query_all(
cursor,
"""
SELECT
c.TABLE_NAME,
t.TABLE_ROWS
FROM INFORMATION_SCHEMA.COLUMNS c
JOIN INFORMATION_SCHEMA.TABLES t
ON c.TABLE_SCHEMA = t.TABLE_SCHEMA
AND c.TABLE_NAME = t.TABLE_NAME
WHERE c.TABLE_SCHEMA = %s
AND c.COLUMN_NAME = 'talent_id'
ORDER BY t.TABLE_ROWS DESC
""",
(DB_CONFIG["database"],),
)
# 对含 user_id 的表的填充率
user_id_fill = []
for row in user_id_tables:
tname = row["TABLE_NAME"]
if row["TABLE_ROWS"] == 0:
user_id_fill.append({"table_name": tname, "total_rows": 0, "has_value": 0, "fill_rate_pct": 0})
continue
try:
stats = query_all(
cursor,
f"""
SELECT
COUNT(*) AS total_rows,
SUM(CASE WHEN user_id IS NOT NULL THEN 1 ELSE 0 END) AS has_value
FROM `{tname}`
""",
)
if stats:
stats[0]["table_name"] = tname
stats[0]["fill_rate_pct"] = round(
stats[0]["has_value"] / stats[0]["total_rows"] * 100, 1
) if stats[0]["total_rows"] > 0 else 0
user_id_fill.append(stats[0])
except Exception as e:
user_id_fill.append({"table_name": tname, "error": str(e)})
# 对含 talent_id 的表的填充率
talent_id_fill = []
for row in talent_id_tables:
tname = row["TABLE_NAME"]
if row["TABLE_ROWS"] == 0:
talent_id_fill.append({"table_name": tname, "total_rows": 0, "has_value": 0, "fill_rate_pct": 0})
continue
try:
stats = query_all(
cursor,
f"""
SELECT
COUNT(*) AS total_rows,
SUM(CASE WHEN talent_id IS NOT NULL THEN 1 ELSE 0 END) AS has_value
FROM `{tname}`
""",
)
if stats:
stats[0]["table_name"] = tname
stats[0]["fill_rate_pct"] = round(
stats[0]["has_value"] / stats[0]["total_rows"] * 100, 1
) if stats[0]["total_rows"] > 0 else 0
talent_id_fill.append(stats[0])
except Exception as e:
talent_id_fill.append({"table_name": tname, "error": str(e)})
return {
"tables_with_user_id": user_id_tables,
"tables_with_talent_id": talent_id_tables,
"user_id_fill_rate": user_id_fill,
"talent_id_fill_rate": talent_id_fill,
}
# ── 保存结果 ──────────────────────────────────────────
def save_json(data, filepath):
script_dir = os.path.dirname(os.path.abspath(__file__))
full_path = os.path.join(script_dir, filepath)
with open(full_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2, default=default_serializer)
print(f"\n✓ 结果已保存至: {full_path}")
# ── 主流程 ────────────────────────────────────────────
def safe_verify(section_name, func, cursor, results):
"""安全执行一个验证步骤，失败时记录错误但继续执行"""
try:
results[section_name] = func(cursor)
print(f" ✓ {section_name} 完成")
except Exception as e:
print(f" ✗ {section_name} 失败: {e}")
results[section_name] = {"error": str(e), "status": "failed"}
def main():
password = get_db_password()
conn = connect_db(password)
results = {
"meta": {
"script": "verify_merge_redundancy.py",
"database": DB_CONFIG["database"],
"host": DB_CONFIG["host"],
"executed_at": datetime.now().isoformat(),
},
}
# 定义所有验证步骤：(section_name, 标题, 函数)
steps = [
("merge_001_em_gas_water", "一、结构完全相同表 - em_gas vs em_water_supply", verify_merge_001_gas_water),
("merge_002_project_records", "二、结构完全相同表 - t_project_*_record ×5", verify_merge_002_project_records),
("merge_003_biz_history", "三、结构高度相似表 - biz_history_*", verify_merge_003_biz_history),
("old_vs_new_talent", "四、旧版系统表 vs 新版人才系统表", verify_old_vs_new_talent),
("weak_decommission", "五、弱废弃候选表（ROWS=0）", verify_weak_decommission),
("project_name_redundancy", "六、project_name 冗余验证", verify_project_name_redundancy),
("enterprise_name_redundancy", "七、enterprise_name 冗余验证", verify_enterprise_name_redundancy),
("xzqhbm_redundancy", "八、xzqhbm 行政区划编码验证", verify_xzqhbm_redundancy),
("site_code_redundancy", "九、site_code / site_name 冗余验证", verify_site_code_redundancy),
("empty_comment_check", "十、无注释表及字段检查", verify_empty_comment_tables),
("dual_id_system", "十一、user_id vs talent_id 双ID体系", verify_dual_id_system),
]
try:
with conn.cursor() as cursor:
for section_name, title, func in steps:
print(f"\n{'─' * 50}")
print(title)
print("─" * 50)
safe_verify(section_name, func, cursor, results)
# ── 保存 ──
save_json(results, OUTPUT_FILE)
# ── 统计输出 ──
succeeded = sum(1 for k, v in results.items()
if not k.startswith("meta") and not isinstance(v, dict) or
(isinstance(v, dict) and v.get("status") != "failed"))
failed = sum(1 for k, v in results.items()
if isinstance(v, dict) and v.get("status") == "failed")
print(f"\n{'─' * 40}")
print(f" 验证完成: {succeeded} 成功, {failed} 失败")
print(f"{'─' * 40}\n")
finally:
conn.close()
print("数据库连接已关闭。")
if __name__ == "__main__":
main()
data_dictionary/verify_merge_redundancy.sql 0 → 100644
View file @ 31911f51
-- ============================================================
-- 数据治理 - 表合并与冗余字段验证 SQL
-- 数据库: smart-build
-- 生成日期: 2026-08-03
-- 用途: 验证数据字典分析中发现的合并候选和冗余问题
-- ============================================================
-- ============================================================
-- 一、结构完全相同表 - 数据验证
-- ============================================================
-- 1.1 MERGE-001: em_gas vs em_water_supply 数据对比
-- 查看两张表的数据量和字段内容，判断是否可合并
SELECT 'em_gas' AS table_name, COUNT(*) AS row_count FROM em_gas
UNION ALL
SELECT 'em_water_supply' AS table_name, COUNT(*) AS row_count FROM em_water_supply;
-- 1.2 MERGE-001: 两张表各取3条样本数据对比结构
SELECT * FROM em_gas LIMIT 3;
SELECT * FROM em_water_supply LIMIT 3;
-- 1.3 MERGE-002: 5张 t_project_*_record 表的数据量
SELECT 't_project_acceptance_record' AS table_name, COUNT(*) AS row_count FROM t_project_acceptance_record
UNION ALL
SELECT 't_project_construction_record', COUNT(*) FROM t_project_construction_record
UNION ALL
SELECT 't_project_design_record', COUNT(*) FROM t_project_design_record
UNION ALL
SELECT 't_project_general_record', COUNT(*) FROM t_project_general_record
UNION ALL
SELECT 't_project_operation_record', COUNT(*) FROM t_project_operation_record;
-- 1.4 MERGE-002: 查看每张表的 sample 数据
SELECT 'acceptance' AS type, * FROM t_project_acceptance_record LIMIT 3;
SELECT 'construction' AS type, * FROM t_project_construction_record LIMIT 3;
SELECT 'design' AS type, * FROM t_project_design_record LIMIT 3;
SELECT 'general' AS type, * FROM t_project_general_record LIMIT 3;
SELECT 'operation' AS type, * FROM t_project_operation_record LIMIT 3;
-- 1.5 MERGE-003: biz_history_field_label vs biz_history_focus_field
SELECT 'biz_history_field_label' AS table_name, COUNT(*) AS row_count FROM biz_history_field_label
UNION ALL
SELECT 'biz_history_focus_field', COUNT(*) FROM biz_history_focus_field;
SELECT * FROM biz_history_field_label LIMIT 5;
SELECT * FROM biz_history_focus_field LIMIT 5;
-- ============================================================
-- 二、旧版系统表（t_*）vs 新版系统表（t_talent_*）数据量对比
-- ============================================================
-- 2.1 确认旧版表是否真的全为空
SELECT 't_course' AS tbl, COUNT(*) AS cnt FROM t_course
UNION ALL SELECT 't_course_chapter', COUNT(*) FROM t_course_chapter
UNION ALL SELECT 't_course_review', COUNT(*) FROM t_course_review
UNION ALL SELECT 't_exam_paper', COUNT(*) FROM t_exam_paper
UNION ALL SELECT 't_exam_question', COUNT(*) FROM t_exam_question
UNION ALL SELECT 't_exam_record', COUNT(*) FROM t_exam_record
UNION ALL SELECT 't_interview', COUNT(*) FROM t_interview
UNION ALL SELECT 't_job_application', COUNT(*) FROM t_job_application
UNION ALL SELECT 't_job_post', COUNT(*) FROM t_job_post
UNION ALL SELECT 't_learning_progress', COUNT(*) FROM t_learning_progress
UNION ALL SELECT 't_offer_record', COUNT(*) FROM t_offer_record
UNION ALL SELECT 't_resume', COUNT(*) FROM t_resume
UNION ALL SELECT 't_title_application', COUNT(*) FROM t_title_application
UNION ALL SELECT 't_title_material', COUNT(*) FROM t_title_material
UNION ALL SELECT 't_title_policy', COUNT(*) FROM t_title_policy
UNION ALL SELECT 't_training_certificate', COUNT(*) FROM t_training_certificate
UNION ALL SELECT 't_certificate_template', COUNT(*) FROM t_certificate_template
UNION ALL SELECT 't_cert_verify_cache', COUNT(*) FROM t_cert_verify_cache
UNION ALL SELECT 't_cert_verify_record', COUNT(*) FROM t_cert_verify_record;
-- 2.2 确认新版表是否有数据
SELECT 't_talent_course' AS tbl, COUNT(*) AS cnt FROM t_talent_course
UNION ALL SELECT 't_talent_course_chapter', COUNT(*) FROM t_talent_course_chapter
UNION ALL SELECT 't_talent_exam', COUNT(*) FROM t_talent_exam
UNION ALL SELECT 't_talent_exam_record', COUNT(*) FROM t_talent_exam_record
UNION ALL SELECT 't_talent_interview', COUNT(*) FROM t_talent_interview
UNION ALL SELECT 't_talent_application', COUNT(*) FROM t_talent_application
UNION ALL SELECT 't_talent_job', COUNT(*) FROM t_talent_job
UNION ALL SELECT 't_talent_offer', COUNT(*) FROM t_talent_offer
UNION ALL SELECT 't_talent_resume', COUNT(*) FROM t_talent_resume
UNION ALL SELECT 't_talent_title_application', COUNT(*) FROM t_talent_title_application
UNION ALL SELECT 't_talent_title_policy', COUNT(*) FROM t_talent_title_policy
UNION ALL SELECT 't_talent_e_certificate', COUNT(*) FROM t_talent_e_certificate
UNION ALL SELECT 't_talent_cert_verify_log', COUNT(*) FROM t_talent_cert_verify_log;
-- ============================================================
-- 三、弱废弃候选表（ROWS=0 且无替代表）
-- ============================================================
SELECT 'mall_address' AS tbl, COUNT(*) AS cnt FROM mall_address
UNION ALL SELECT 'mall_favorite', COUNT(*) FROM mall_favorite
UNION ALL SELECT 'mall_freight_template', COUNT(*) FROM mall_freight_template
UNION ALL SELECT 'mall_freight_template_rule', COUNT(*) FROM mall_freight_template_rule
UNION ALL SELECT 'mall_inventory_log', COUNT(*) FROM mall_inventory_log
UNION ALL SELECT 'project_decision', COUNT(*) FROM project_decision
UNION ALL SELECT 'project_decision_attach', COUNT(*) FROM project_decision_attach
UNION ALL SELECT 'project_decision_field', COUNT(*) FROM project_decision_field
UNION ALL SELECT 'project_design', COUNT(*) FROM project_design
UNION ALL SELECT 't_ai_llm_provider', COUNT(*) FROM t_ai_llm_provider
UNION ALL SELECT 't_ai_plan_optimize', COUNT(*) FROM t_ai_plan_optimize
UNION ALL SELECT 'c_gp_catalog', COUNT(*) FROM c_gp_catalog;
-- ============================================================
-- 四、冗余字段验证 - project_name 在30张表中冗余
-- ============================================================
-- 4.1 验证：同一 project_id 在不同表中 project_name 是否一致
-- 示例：t_project vs t_project_pre_project
SELECT
a.id AS project_id,
a.name AS name_in_project_table,
b.project_name AS name_in_pre_project_table,
(a.name = b.project_name) AS is_consistent
FROM t_project a
INNER JOIN t_project_pre_project b ON a.id = b.project_id
LIMIT 20;
-- 4.2 统计 project_name 在各表中出现不一致的情况（多表对比）
SELECT 't_project_pre_project' AS source_table, project_id, project_name FROM t_project_pre_project WHERE project_id IS NOT NULL LIMIT 0;
-- ============================================================
-- 五、冗余字段验证 - enterprise_name 在13张表中冗余
-- ============================================================
-- 5.1 统计 enterprise_name 在各表中的填充率
SELECT
't_enterprise_credit_record' AS table_name,
COUNT(*) AS total_rows,
SUM(CASE WHEN enterprise_name IS NOT NULL AND enterprise_name != '' THEN 1 ELSE 0 END) AS filled_rows,
ROUND(SUM(CASE WHEN enterprise_name IS NOT NULL AND enterprise_name != '' THEN 1 ELSE 0 END) / COUNT(*) * 100, 1) AS fill_rate
FROM t_enterprise_credit_record
UNION ALL
SELECT 't_enterprise_performance', COUNT(*),
SUM(CASE WHEN enterprise_name IS NOT NULL AND enterprise_name != '' THEN 1 ELSE 0 END),
ROUND(SUM(CASE WHEN enterprise_name IS NOT NULL AND enterprise_name != '' THEN 1 ELSE 0 END) / COUNT(*) * 100, 1)
FROM t_enterprise_performance;
-- ============================================================
-- 六、冗余字段验证 - xzqhbm（行政区划编码）在20张表中冗余
-- ============================================================
-- 6.1 c_bri_xzqh 字典表的内容
SELECT COUNT(*) AS dict_count FROM c_bri_xzqh;
SELECT * FROM c_bri_xzqh LIMIT 10;
-- 6.2 检查使用 xzqhbm 的表是否都可以关联到字典表
-- 以 em_gas 为例
SELECT
g.xzqhbm,
x.xzqhmc AS dict_name
FROM em_gas g
LEFT JOIN c_bri_xzqh x ON g.xzqhbm = x.xzqhbm
WHERE g.xzqhbm IS NOT NULL AND x.xzqhbm IS NULL
LIMIT 20;
-- 如果有结果，说明存在字典表中不存在的编码
-- ============================================================
-- 七、site_code / site_name 冗余验证
-- ============================================================
-- 7.1 统计 site 相关表的数据量
SELECT 't_site_attendance' AS tbl, COUNT(*) AS cnt FROM t_site_attendance
UNION ALL SELECT 't_site_check_apply', COUNT(*) FROM t_site_check_apply
UNION ALL SELECT 't_site_check_item', COUNT(*) FROM t_site_check_item
UNION ALL SELECT 't_site_check_record', COUNT(*) FROM t_site_check_record
UNION ALL SELECT 't_site_equipment', COUNT(*) FROM t_site_equipment
UNION ALL SELECT 't_site_lift', COUNT(*) FROM t_site_lift
UNION ALL SELECT 't_site_quality', COUNT(*) FROM t_site_quality
UNION ALL SELECT 't_site_safety', COUNT(*) FROM t_site_safety
UNION ALL SELECT 't_site_tower', COUNT(*) FROM t_site_tower
UNION ALL SELECT 't_site_video', COUNT(*) FROM t_site_video;
-- 7.2 检查同一 site_code 在不同表中的 site_name 是否一致
SELECT
a.site_code,
a.site_name AS name_in_safety,
b.site_name AS name_in_quality
FROM t_site_safety a
INNER JOIN t_site_quality b ON a.site_code = b.site_code
WHERE a.site_name != b.site_name
LIMIT 20;
-- ============================================================
-- 八、无注释表检查 - c_bri_geo_boundary
-- ============================================================
-- 8.1 查看该表结构和数据样本
SELECT COUNT(*) FROM c_bri_geo_boundary;
SELECT * FROM c_bri_geo_boundary LIMIT 5;
-- 8.2 查看该表字段注释情况
SELECT
COLUMN_NAME,
COLUMN_TYPE,
COLUMN_COMMENT,
IS_NULLABLE,
COLUMN_DEFAULT
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = 'smart-build'
AND TABLE_NAME = 'c_bri_geo_boundary'
ORDER BY ORDINAL_POSITION;
-- ============================================================
-- 九、user_id vs talent_id 两套ID体系
-- ============================================================
-- 9.1 统计 user_id 和 talent_id 分别在哪些表中有数据
SELECT 'talent_id in t_talent_course' AS location, COUNT(*) AS total,
SUM(CASE WHEN talent_id IS NOT NULL THEN 1 ELSE 0 END) AS has_value
FROM t_talent_course
UNION ALL
SELECT 'user_id in t_course', COUNT(*),
SUM(CASE WHEN user_id IS NOT NULL THEN 1 ELSE 0 END)
FROM t_course;
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment