feat: 升级FLOAT/DOUBLE审核规则为ERROR级别-覆盖CREATE+ALTER-MODIFY-CHANGE-防止生产精度丢失

This commit is contained in:
logan committed 2026-05-25 01:56:26 +08:00
1 parent 8f962a988f
commit ddf4856a85
1 file changed
+27 -11
+27 -11
View File
@@ -392,25 +392,41 @@ def check_mysql_ddl_004(sql_clean: str, sql_upper: str, report: AuditReportV2, r
code="MYSQL-DDL-005",
name="DDLCheckFloatDouble",
category=RuleCategory.DDL,
severity=Severity.WARNING,
severity=Severity.ERROR,
db_type="mysql",
description="不建议使用 FLOAT/DOUBLE 类型,建议使用 DECIMAL 替代",
description="禁止使用 FLOAT/DOUBLE 类型存储金额/数值数据,必须使用 DECIMAL 保证精度",
config={"enabled": True},
message_template="表 '{table}' 的字段 '{column}' 使用了 {ctype} 类型,建议改为 DECIMAL",
message_template="表 '{table}' 的字段 '{column}' 使用了 {ctype} 类型,存在精度丢失风险,必须使用 DECIMAL 替代(如 DECIMAL(16,4))",
))
def check_mysql_ddl_005(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
"""强制float/double类型变更为decimal类型。"""
if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE):
return
table_name = _extract_table_name(sql_upper, sql_clean) or "unknown"
columns = _extract_column_definitions(sql_upper)
for col in columns:
if col["type"].upper() in ("FLOAT", "DOUBLE"):
"""禁止 FLOAT/DOUBLE 类型,强制使用 DECIMAL 保证精度。"""
# 检查 CREATE TABLE
if re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE):
table_name = _extract_table_name(sql_upper, sql_clean) or "unknown"
columns = _extract_column_definitions(sql_upper)
for col in columns:
if col["type"].upper() in ("FLOAT", "DOUBLE", "REAL"):
report.findings.append(AuditFindingV2(
rule_code=rule.code,
rule_name=rule.name,
level=rule.severity,
message=rule.message_template.format(table=table_name, column=col["name"], ctype=col["type"]),
))
# 检查 ALTER TABLE ADD/MODIFY/CHANGE COLUMN
alter_match = re.search(r"\bALTER\s+TABLE\s+[`\"']?(\w+)", sql_upper, re.IGNORECASE)
if alter_match:
table_name = alter_match.group(1)
# 匹配 ADD/MODIFY/CHANGE COLUMN 中出现的 FLOAT/DOUBLE 类型
float_pattern = r"(?:ADD|MODIFY|CHANGE)\s+(?:COLUMN\s+)?[`\"']?(\w+)[`\"']?\s+(?:FLOAT|DOUBLE|REAL)\b"
for m in re.finditer(float_pattern, sql_upper, re.IGNORECASE):
col_name = m.group(1)
ctype = "FLOAT" if "FLOAT" in m.group(0).upper() else ("DOUBLE" if "DOUBLE" in m.group(0).upper() else "REAL")
report.findings.append(AuditFindingV2(
rule_code=rule.code,
rule_name=rule.name,
level=rule.severity,
message=rule.message_template.format(table=table_name, column=col["name"], ctype=col["type"]),
message=rule.message_template.format(table=table_name, column=col_name, ctype=ctype),
))