简介：「数据质量管理平台需求文档.pdf」面向数据治理、数据中台建设及信息化项目规划人员，聚焦企业数据资产准确性、完整性与合规性难题，提供一份可直接参照的平台建设需求蓝本。文档以需求规格形式展开，依次覆盖项目介绍、业务方案与功能设计三大板块：前者界定数据质量含义与准确性、完整性、一致性、时效性、有效性、可追溯性六要素，并说明项目背景与目标；中段给出模块化系统架构与整体要求；后段细化模板管理中的内置模板、创建、查询、修改与审批流程，并延伸至规则管理、任务管理等模块。资源包共1个文件，为1.81MB的PDF文档，目录层级分明、条目编号完整，便于按章节检索与摘录成招标或开发需求清单。目前已有443人学习，适合作为需求梳理、方案评审与开发落地的参考底稿。
1. 从一份 PDF 需求文档拆出可落地的数据质量管理平台
拿到《数据质量管理平台需求文档》这类 PDF，多数人的第一反应是翻到功能清单数模块数量，再照着目录排期。真正拖垮进度的从来不是功能多少，而是文档里没落到字段和状态机上的那几句话——“模板审批通过后可升级为内置模板”，审批状态存哪张表、升级后引用旧模板的实例怎么处理，文档不会回答。这份需求把平台划成模板管理、规则管理、任务管理、检查结果分析、问题处理、知识库、系统管理 7 大模块、66 个功能点，节点定在 11 月中旬上线。两类人看它收益最大：要出技术方案的开发负责人，和准备把数据质量六要素写成检核 SQL 的数据开发。下面按元数据建模、规则与调度、问题闭环、验收与安全基线的顺序拆开。
2. 数据质量六要素到模板元数据的建模路径
2.1 六要素怎么落成一条条可执行的检核口径
文档给的数据质量六要素是完整性、唯一性、一致性、精确度、合法性、及时性。这六个词写在需求里很漂亮，落到代码里必须换成具体口径，否则每条规则都是一次自由发挥。我的做法是先做一张要素到口径的映射表，评审时逐条确认，避免开发按自己理解写 SQL、业务按自己理解提缺陷。
| 要素 | 文档中的定义 | 可落地的检核口径 | 误报高发点 |
|---|---|---|---|
| 完整性 | 实体、属性、记录、字段四类缺失 | 主键非空、必填字段空值率、主表与从表左连接计数为 0 | 宽表按天分区，上游停跑被误判为整表缺失 |
| 唯一性 | 主键唯一、候选键唯一 | COUNT(*)与COUNT(DISTINCT key)比对 | 大小写、首尾空格、全半角造成假重复 |
| 一致性 | 统一来源、冗余存储、统一口径 | 同一指标跨表比对、与上下游对账差值 | 两表统计时点不一致，差值是时间差不是数据差 |
| 精确度 | 计量误差、度量单位 | 数值区间校验、精度位数校验、单位换算后比对 | 元与万元混用、汇率日期未对齐 |
| 合法性 | 格式、类型、域值、业务规则 | 正则匹配、枚举字典校验、日期区间校验 | 枚举字典没做版本，历史数据全部报警 |
| 及时性 | 刷新、修改、提取的及时性 | 数据到达时间减业务发生时间，与 SLA 比对 | 上游批处理窗口漂移，阈值写死导致周期性误报 |
这张表的用法不是贴在文档里，而是直接变成规则模板的字段来源：模板体里每增加一个“检核类型”，就必须能在这张表里找到对应的口径描述和误报说明。
2.2 模板头与模板体：两段式元数据
文档讲得很明确，一个模板由模板头和模板体两部分组成，模板头用于模板管理，模板体才是真正功能，用到模板的地方其实是模板体的实例化。这句话决定了表结构不能拍平成一张大表。模板头强制字段有模板编号、模板名称、模板类别、创建人员、创建时间、模板状态、模板描述；模板体至少要有对象编号、对象名称（可选）、对象创建时间、对象状态、对象创建人员、对象类别。
-- 模板头：只回答“这个模板是谁建的、能不能用” CREATE TABLE dq_template_head ( id BIGINT NOT NULL AUTO_INCREMENT, template_code VARCHAR(64) NOT NULL COMMENT '模板编号，业务唯一', template_name VARCHAR(128) NOT NULL COMMENT '模板名称', category VARCHAR(32) NOT NULL COMMENT 'RULE/TASK/REPORT/PLAN/RESULT/TICKET/QUALITY_REPORT/VIZ/LOG', status TINYINT NOT NULL DEFAULT 0 COMMENT '0草稿 1待审批 2已通过 3已否决 4已内置 5已停用', is_builtin TINYINT NOT NULL DEFAULT 0 COMMENT '是否内置模板', creator VARCHAR(64) NOT NULL COMMENT '创建人员', create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, description VARCHAR(512) DEFAULT NULL COMMENT '模板描述', PRIMARY KEY (id), UNIQUE KEY uk_template_code (template_code), KEY idx_category_status (category, status) ) COMMENT='模板头'; -- 模板体：模板的实际功能，实例化时被复制到业务表 CREATE TABLE dq_template_body ( id BIGINT NOT NULL AUTO_INCREMENT, template_code VARCHAR(64) NOT NULL COMMENT '归属模板头', object_code VARCHAR(64) NOT NULL COMMENT '对象编号，如规则码、字段名、报表项', object_name VARCHAR(128) DEFAULT NULL COMMENT '对象名称，可选', object_type VARCHAR(32) NOT NULL COMMENT 'FIELD/SQL/CHART/ACTION', object_status TINYINT NOT NULL DEFAULT 1 COMMENT '1启用 0停用', object_creator VARCHAR(64) NOT NULL COMMENT '对象创建人员', object_create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, body_schema JSON DEFAULT NULL COMMENT '校验类型、阈值、渲染配置等扩展定义', PRIMARY KEY (id), UNIQUE KEY uk_tpl_object (template_code, object_code) ) COMMENT='模板体';参数上要留意两点。category用短码而不是中文名，是为了后面按类别做权限和数据隔离时不用做字符串匹配；body_schema用 JSON 而不是继续拆宽表，是因为规则模板要存 SQL 片段、任务模板要存调度表达式、可视化模板要存图表配置，字段差异大到无法共用一张明细表，硬拉宽的结果就是一半列常年为 NULL。另外object_status、object_creator这些字段在模板体里是冗余的，实例化时可以直接带过去，少一次 join。
注意：文档目录写“6 类内置模板”，但 3.1.1 小节里实际列了规则、任务、报表、方案、结果、工单、质量报告、可视化、日志共 9 类。这类自相矛盾必须在设计评审上定死，我一般把它做成
category字典表，字典项可扩展，代码里不做枚举硬编码。
2.3 模板状态机与“升级为内置模板”的判定
文档说新创建的模板根据使用频率和重要性，在通过审批之后可以升级为内置模板。“使用频率”和“重要性”是不可计算的词，得换成可查的指标：引用该模板的实例数量、近 90 天被执行的任务数、是否被标记为核心。引用量可以直接聚合出来。
-- 统计模板被规则引用的次数，作为升级为内置模板的依据之一 SELECT t.template_code, t.template_name, COUNT(r.id) AS ref_cnt FROM dq_template_head t LEFT JOIN dq_rule r ON r.template_code = t.template_code AND r.status <> 'DELETED' WHERE t.category = 'RULE' AND t.status = 2 GROUP BY t.template_code, t.template_name HAVING COUNT(r.id) >= 20 ORDER BY ref_cnt DESC;审批流不要为模板和规则各写一套，用一张流水表更省事，业务类型加业务编号就能复用。
CREATE TABLE dq_approval_flow ( id BIGINT AUTO_INCREMENT, biz_type VARCHAR(32) NOT NULL COMMENT 'TEMPLATE/RULE', biz_code VARCHAR(64) NOT NULL COMMENT '业务编号', approve_status TINYINT NOT NULL COMMENT '1通过 2否决', approve_comment VARCHAR(512) DEFAULT NULL COMMENT '审批结果文字说明', approver VARCHAR(64) NOT NULL COMMENT '自动写入当前审批人', approve_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '自动写入', PRIMARY KEY (id), KEY idx_biz (biz_type, biz_code) ) COMMENT='审批流水';文档要求审批界面有“保存”和“取消”，取消要放弃界面操作内容，也就是说取消动作不能落库，连草稿记录都不该产生。实现上别用“先插入再回滚”的写法，直接让前端不提交即可。
2.4 从 PDF 需求文档里抽字段表，减少手抄错
需求文档里的字段表是横排加合并单元格，手抄进 DDL 出错率很高。常见做法是用 pdfplumber 把字段表抽成 CSV，再和 DDL 逐字段比对，把差异当成评审议题。
import csv, re import pdfplumber # 只关心模板头/模板体里出现的字段名，避免把正文段落也抓进来 FIELD_RE = re.compile(r"^(模板|对象)(编号|名称|类别|状态|描述|创建人|创建时间)") with pdfplumber.open("数据质量管理平台需求文档.pdf") as pdf, \ open("template_fields.csv", "w", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow(["page", "field", "optional", "raw_row"]) for pno, page in enumerate(pdf.pages, 1): for table in page.extract_tables(): for row in table: cells = [(c or "").replace("\n", "").strip() for c in row] for cell in cells: if FIELD_RE.match(cell): writer.writerow([pno, cell, "可选" in "".join(cells), "|".join(cells)])extract_tables()遇到合并单元格会返回 None，c or ""的兜底不能省，否则整张表抽取会中断；page是页码，用来回查原文，页码偏移因为前置目录的存在需要人工确认一次。抽出来的结果只当校验参照，不要当唯一事实来源，最终以评审确认的字段清单为准。
3. 规则管理与检核任务调度的实现
3.1 规则创建与 Excel 批量导入
规则创建的硬约束是必须先指定规则模板，也就是内置规则模板的模板体部分，创建时状态自动置为“开发”。批量导入是最大的坑位：业务给的 Excel 列名不固定、枚举值大小写不一致、同一个规则编号重复出现在两行。我的做法是整表先校验、全通过再入库，按规则编号做幂等 upsert。
import openpyxl, pymysql REQUIRED = ["规则编号", "规则名称", "规则模板编号", "对象编号", "检核类型", "检核SQL", "阈值", "数据源", "责任部门"] VALID_CHECK = {"完整性", "唯一性", "一致性", "精确度", "合法性", "及时性"} def load_rules(path): wb = openpyxl.load_workbook(path, read_only=True) ws = wb.active header = [str(c.value).strip() if c.value else "" for c in next(ws.rows)] idx = {name: header.index(name) for name in REQUIRED} rows, errors = [], [] for rno, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2): item = {k: (row[i] if row[i] is not None else "") for k, i in idx.items()} item = {k: str(v).strip() for k, v in item.items()} if not item["规则编号"]: errors.append(f"第{rno}行：规则编号为空") continue if item["检核类型"] not in VALID_CHECK: errors.append(f"第{rno}行：检核类型 {item['检核类型']} 不在六要素范围内") continue item["状态"] = "DEVELOP" # 文档要求：创建后自动为“开发” rows.append(item) return rows, errorsread_only=True避免大文件把内存打满；values_only=True拿到的是值而不是单元格对象，省掉.value调用。错误信息带上行号，业务能自己定位。入库时用INSERT ... ON DUPLICATE KEY UPDATE，键是规则编号，重复导入不会产生重复规则。
规则状态的回退规则也要一并定清楚，否则审批通过后的规则被改一次，历史任务跑的是哪一版就说不清了。
| 操作 | 修改前状态 | 修改后状态 | 附带动作 |
|---|---|---|---|
| 界面录入或导入 | 无 | 开发 | 生成规则编号，绑定模板编号 |
| 编辑规则体或检核 SQL | 已发布 | 开发 | 给关联任务打待复核标记并通知负责人 |
| 审批通过 | 开发 / 待审批 | 已发布 | 写入审批流水，通知任务负责人 |
| 审批否决 | 待审批 | 开发 | 记录审批意见，允许再次修改后重新提交 |
3.2 任务状态机与规则执行引擎的交互
文档把任务状态分成未创建完成、未执行、在执行、已完成、异常中止五类，查询页要能按这些状态分开展示任务，还要显示成功次数、失败次数。这里有个容易做错的地方：提交成功后任务自动从“未执行任务列表”中消失。如果列表过滤条件和状态字段是两套逻辑，线上一定会出现“列表里看不见、状态还是未执行”的脏数据。正确做法是列表完全由task_status驱动，不做额外过滤。
-- 任务提交：乐观锁防止调度中心重复拉起同一个任务 UPDATE dq_task SET task_status = 'RUNNING', run_id = ?, last_start_time = NOW(), version = version + 1 WHERE task_code = ? AND task_status IN ('READY', 'RETRY') AND version = ?; -- 应用层判断 affected_rows，等于 0 说明已被其他节点抢占，直接放弃本次调度version是乐观锁版本号，调度节点多实例部署时必须带。run_id是一轮执行的唯一标识，后面问题数据、任务日志全部挂在它上面，排错时只用一个 ID 就能串起整条链路。规则执行的开始结束时间、每条规则的耗时、问题数据量，都由执行引擎在执行过程中回写，不要等任务结束再批量补，否则任务被 kill 后什么痕迹都没有。
3.3 任务日志的分级与日志模板落地
文档对任务日志列了七类内容，本质上对应三个级别：开始结束时间、执行状态、问题数量、规则耗时属于提示信息；数据库脚本属于调试信息；错误和警告单独成类。把这些字段硬塞进一张宽表会很别扭，拆成“执行实例表 + 明细日志表”更合理。
CREATE TABLE dq_task_log ( id BIGINT NOT NULL AUTO_INCREMENT, run_id VARCHAR(64) NOT NULL COMMENT '执行实例，与任务表对应', task_code VARCHAR(64) NOT NULL, rule_code VARCHAR(64) DEFAULT NULL COMMENT '规则级日志才填', log_level VARCHAR(8) NOT NULL COMMENT 'INFO/DEBUG/WARN/ERROR', content VARCHAR(1024) DEFAULT NULL COMMENT '提示、警告或错误信息', script_sql TEXT DEFAULT NULL COMMENT '仅 DEBUG 级别落库，避免日志表膨胀', cost_ms INT DEFAULT 0 COMMENT '规则执行耗时', problem_cnt INT DEFAULT 0 COMMENT '该规则产生的问题数据量', log_time DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), PRIMARY KEY (id), KEY idx_run_time (run_id, log_time), KEY idx_rule (rule_code, log_time) ) COMMENT='任务日志';| 文档要求的日志内容 | 日志级别 | 落地位置 |
|---|---|---|
| 任务开始/结束时间 | INFO | 执行实例表的 start_time、end_time |
| 成功或失败状态、问题数据数量 | INFO | 执行实例表的 status、problem_cnt |
| 每个规则的开始/结束时间 | INFO | 明细日志的 log_time 与 cost_ms |
| 规则耗时与问题数据量 | INFO | 明细日志的 cost_ms、problem_cnt |
| 规则执行的主要数据库脚本 | DEBUG | 明细日志的 script_sql |
| 错误信息 | ERROR | 明细日志的 content |
| 警告信息 | WARN | 明细日志的 content |
生产默认开 INFO，SQL 脚本不落库；出问题后按run_id单独把某条规则重跑并打开 DEBUG，比全局开 DEBUG 划算得多。log_time用毫秒精度，是因为同一条规则内的多条日志在同一秒内产生，秒级精度排不出顺序。
3.4 规则变更对在执行任务的联动影响
文档明确写了，修改规则状态时，要自动对与该规则相关联的在执行任务、问题、方案做修改。字面实现是级联 UPDATE，但那样会把历史执行记录改成“当时不存在的样子”，问题数据再也复算不出来。我一般不走级联改数据，改成两件事：规则版本号自增，任务引用版本；历史执行把规则体内联成快照，保证任何一次历史结果都能复现。
ALTER TABLE dq_rule ADD COLUMN version INT NOT NULL DEFAULT 1 COMMENT '规则版本号'; -- 规则体变更时版本自增，并把关联任务标记为待复核 UPDATE dq_rule SET version = version + 1, rule_body = ?, status = 'DEVELOP' WHERE rule_code = ? AND version = ?; INSERT INTO dq_task_review (task_code, rule_code, reason, create_time) VALUES (?, ?, '规则已变更，需确认是否继续调度', NOW());“关联的方案”指的是已经按老规则制定的整改方案，处理方式不是改方案内容，而是在方案上挂一条“依据规则已变更”的提示，让方案负责人决定是否重新生成。这样审计时能回答清楚：这次任务跑的是第几版规则，问题数据是按哪一版口径产生的。
4. 问题数据、工单与知识库的闭环
4.1 结果模板决定问题数据怎么存
结果模板规定了检查任务结果的数据项，问题数据表必须严格按这个规范来，因为下游的分析、工单、报告全部读它。表结构里最需要斟酌的是“问题记录怎么标识”。
CREATE TABLE dq_problem_data ( id BIGINT AUTO_INCREMENT PRIMARY KEY, run_id VARCHAR(64) NOT NULL COMMENT '来自哪次执行', task_code VARCHAR(64) NOT NULL, rule_code VARCHAR(64) NOT NULL, rule_version INT NOT NULL COMMENT '产生该问题时规则的版本', table_name VARCHAR(128) NOT NULL COMMENT '问题所在表', pk_value VARCHAR(512) NOT NULL COMMENT '问题记录主键值，复合键用 || 拼接', problem_desc VARCHAR(512) DEFAULT NULL COMMENT '按结果模板渲染的描述', severity TINYINT NOT NULL DEFAULT 2 COMMENT '1高 2中 3低', occur_time DATETIME NOT NULL, handle_status TINYINT NOT NULL DEFAULT 0 COMMENT '0待处理 1工单中 2已整改 3已豁免', KEY idx_rule_time (rule_code, occur_time), KEY idx_run (run_id), KEY idx_table (table_name) ) COMMENT='问题数据';只存主键值不存整行，是最关键的取舍。一条规则在千万级表上命中十万行时，把整行复制过来会让质量库直接爆掉；而只存主键，整改人员用一次IN查询就能回原表取数。pk_value拼接时选一个不会出现在主键里的分隔符，拼完检查长度，超过 512 的规则直接在设计阶段否掉。误报不能删数据，用handle_status = 3加豁免原因，否则下一轮检核又会全部报出来。
4.2 分析管理：聚合口径要先固定分母
按规则统计问题数据量是内置报表模板里的第一张表，写起来简单，坑全在分母上。
SELECT r.rule_code, r.rule_name, COUNT(*) AS problem_cnt, SUM(CASE WHEN p.severity = 1 THEN 1 ELSE 0 END) AS high_cnt, ROUND(COUNT(*) / NULLIF(t.total_cnt, 0) * 100, 4) AS problem_rate FROM dq_problem_data p JOIN dq_rule r ON r.rule_code = p.rule_code LEFT JOIN dq_table_row_count t ON t.table_name = p.table_name AND t.stat_date = ? -- 与检核批次同一天的快照行数 WHERE p.occur_time >= ? AND p.occur_time < ? GROUP BY r.rule_code, r.rule_name ORDER BY problem_cnt DESC;dq_table_row_count必须是检核当时记录的行数快照，不能用SELECT COUNT(*)现场算。任务重跑一次，现场行数变了，问题率就对不上报表。NULLIF防止空表除零。报表模板不要存 SQL 字符串，存数据集 ID 加参数名，SQL 由后端统一维护，这样换个库名、加个过滤条件不用改模板。
{ "template_code": "RPT_RULE_PROBLEM", "category": "REPORT", "dataset": { "sql_id": "dq.rule_problem_stat", "params": ["stat_date", "begin_time", "end_time"] }, "columns": [ { "field": "rule_code", "title": "规则编号", "width": 140 }, { "field": "problem_cnt", "title": "问题数据量", "width": 100, "agg": "sum" }, { "field": "problem_rate","title": "问题率", "format": "0.00%" } ] }4.3 工单流转与问题通知规则的收敛
工单模板规定了工单长什么样，通知规则模板规定了什么情况下发通知。工单的核心是状态机和责任人，状态迁移必须有闭环，否则问题数据会永远挂在“处理中”。
CREATE TABLE dq_ticket ( id BIGINT AUTO_INCREMENT PRIMARY KEY, ticket_code VARCHAR(64) NOT NULL, rule_code VARCHAR(64) NOT NULL, problem_cnt INT NOT NULL, owner VARCHAR(64) NOT NULL COMMENT '由规则的责任部门推导', deadline DATETIME DEFAULT NULL COMMENT '按方案模板的时限推算', ticket_status VARCHAR(16) NOT NULL DEFAULT 'CREATED' COMMENT 'CREATED/PROCESSING/REVIEWING/CLOSED/REJECTED', create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_ticket_code (ticket_code), KEY idx_owner_status (owner, ticket_status) ) COMMENT='质量问题工单';| 状态迁移 | 触发动作 | 校验条件 |
|---|---|---|
| CREATED → PROCESSING | 处理人接单 | 处理人必须是规则责任部门成员 |
| PROCESSING → REVIEWING | 提交整改结果 | 必须填写整改说明 |
| REVIEWING → CLOSED | 数据负责人复核 | 下一轮同规则检核问题数下降或为零 |
| REVIEWING → REJECTED | 复核不通过 | 退回处理人，保留退回次数 |
通知最容易踩的坑是告警风暴：一条规则在凌晨批量跑出几万条问题，直接按问题条数发消息，处理人的手机能响一晚上。收敛做法是按(rule_code, owner)做时间窗口聚合，高严重度问题立即通知、其余按日报汇总，同一窗口内只发一条。
4.4 质量报告智能文字与知识库检索
文档要求质量报告模板以文字为主、辅助报表图形，并且能根据具体数据情况产生相适应的文字描述。这里的“智能分析”在多数项目里的落地方式是文字模板加占位符渲染，而不是让模型自由生成，原因是口径要可解释、可回归测试、责任能落到人。
TEXT_TPL = ( "本期共执行 {task_cnt} 个检核任务，覆盖 {rule_cnt} 条规则，" "发现问题数据 {problem_cnt} 条，其中高严重度 {high_cnt} 条。" "{top_rule} 规则问题最集中，占本期问题总量的 {top_ratio}，" "较上期{trend}。" ) def render(ctx: dict) -> str: ctx.setdefault("trend", "基本持平") # 缺少上期数据时给出中性描述，避免出现 None return TEXT_TPL.format(**ctx)占位符值全部来自 4.2 的聚合查询，模板里只允许出现有数据来源的字段，避免报告里出现编出来的结论。知识库的检索要提前准备中文分词，MySQL 默认分词器按单个汉字切，搜“数据完整性”会命中一堆无关条目。
ALTER TABLE dq_knowledge ADD FULLTEXT INDEX ft_knowledge (title, content) WITH PARSER ngram;知识库条目建议和规则编号建立弱关联，问题处理完成后把结论写回知识库并关联rule_code，下次同类问题触发时能直接命中历史方案。
5. 上线前的验收口径与数据源安全基线
5.1 把 66 个功能点拆成可测项
功能数量多的时候，逐条写测试用例会失控。我一般先挑那些“描述含糊但影响面大”的功能点，把验收口径和证据形式先定下来，开发照着做，测试照着验。
| 功能点 | 验收口径 | 证据形式 |
|---|---|---|
| 模板审批 | 点“取消”不产生任何数据库记录 | 审批流水表前后行数比对 |
| 模板查询 | 支持模板名称分词查询，分类列表带数量 | 查询接口返回结构快照 |
| 规则导出 | 支持全部导出与按查询条件导出，属性可选 | 导出 Excel 的列与选择项一致 |
| 任务删除 | 任意状态任务可删，删除前有确认 | 任务表记录与审计日志 |
| 任务日志 | 按 run_id 可查到七类日志内容 | 一次真实执行的日志导出 |
| 规则变更 | 修改后关联任务进入待复核，历史结果可复算 | 规则版本号与执行快照 |
关键是把“有确认”“可复算”这种形容词换成可观测的动作，否则验收会上谁也说服不了谁。
5.2 数据源凭据与权限的安全基线
系统管理里的数据源管理、用户管理、权限管理是最容易被赶工糊弄的部分，而数据源连接串一旦明文落库，整个平台的问题数据就等于裸奔。基线只有三条：凭据加密存储、权限按数据源收敛、访问留痕。
import os from cryptography.fernet import Fernet _key = os.environ["DQ_SECRET_KEY"] # 密钥不进代码库、不进配置文件 _fernet = Fernet(_key) def save_datasource(password_plain: str) -> bytes: return _fernet.encrypt(password_plain.encode("utf-8")) def load_datasource(cipher: bytes) -> str: return _fernet.decrypt(cipher).decode("utf-8")密钥托管在环境变量或密钥管理服务里，代码库和配置文件里只出现密钥名，不出现密钥值；轮换密钥时保留旧密钥做一次性解密重写，不要直接覆盖。数据源账号一律用只读账号，按库表白名单授权，检核 SQL 上线前做一次语法白名单校验，禁止出现UPDATE、DELETE、DROP。查看问题数据明细的操作单独记审计日志，记录操作人、时间、表名和run_id，这样任何一条问题记录的访问路径都能回溯。把run_id和rule_version写进问题数据表之后，任何一条问题记录都能倒查回具体某次执行和那一版规则，复盘时不用再去翻聊天记录。
本文还有配套的精品资源，点击获取
