本文引用的文件
dm(data model)是连接原始数据与业务查询的中间层。它负责:
对外暴露的 HTTP 入口集中在 /dm 下,包括文件预处理、上传解析、原始预览、执行清洗、取消清洗、清理批次、获取文件列表和删除文件等能力。
dm 后端位于 ai-server/src/main/java/com/zsjz/ai/module/dm,按职责分为控制器、服务、预处理、AI 画像、清洗策略、映射器及通用模型;元数据和建库脚本位于 sql/case_table_1.sql、sql/external_tables.sql 和 sql/external_table_field.sql。
graph TB
FE["前端页面<br/>案例数据导入、画像页"] --> DC["DmController<br/>/dm"]
DC --> DS["DmService<br/>预处理、清洗调度"]
DS --> PT["PreTask<br/>Excel/CSV 解析"]
PT --> PDL["PreDataListener<br/>流式写入 raw_{sheetId}"]
DS --> FAPS["FileAiProfileService<br/>非表格 AI 档案"]
DS --> CF["CleanFactory<br/>策略工厂"]
CF --> DCI["DataCleaner 实现<br/>通话/转账/外部表清洗器"]
DS --> DB[("案件数据库<br/>file_info / file_ai_profile<br/>clean_error_log / edge_conf")]
DS --> EXTDB[("外部表源<br/>cert_info / entry_rec / vehicle_info …")]
图表来源
章节来源
| 组件 | 职责 | 关键输出 |
|---|---|---|
| DmController | 暴露 /dm REST 接口 |
文件树、清洗任务状态、文件列表 |
| DmService | 汇总文件路径、并发预处理、上传落盘、调用清洗、清理批次 | FileInfo 树、批量清洗结果 |
| PreTask | 检测文件类型、编码、密码、分隔符 | 初始化后的 FileInfo |
| PreDataListener | 监听每行、找表头、写 DuckDB 动态表 | raw_{sheetId} 全量原始数据 |
| FileAiProfileService | 为非表格文件建档并提交异步识别 | file_ai_profile 档案 |
| CleanFactory | 根据 StrategyType 创建 DataCleaner/DataLoader | 具体业务清洗器 |
| EdgeConf | 描述实体边关系配置 | edge_conf 关系定义 |
| CleanErrorLogService | 访问清洗错误日志表 | clean_error_log |
章节来源
dm 的工作流可分为四个阶段:文件接入、结构识别、清洗入库、产物消费。
sequenceDiagram
participant Client as "前端"
participant Controller as "DmController"
participant Service as "DmService"
participant Task as "PreTask"
participant Listener as "PreDataListener"
participant DuckDB as "DuckDB 动态表"
participant Profile as "FileAiProfileService"
participant Cleaner as "CleanFactory + DataCleaner"
participant Store as "案件数据库"
Client->>Controller: POST /dm/preFileUpload
Controller->>Service: preProcessUploadedFile(file, batchId, reClean)
Service->>Store: 登记 file_info(含 FILE_UNSUPPORTED_FAIL)
alt CSV/XLS/XLSX
Service->>Task: 解析文件类型/编码/分隔符
Task->>Listener: 逐行回调
Listener->>DuckDB: CREATE TABLE raw_{sheetId}
Listener->>DuckDB: 追加 row_no + 全 VARCHAR 行
Service->>Profile: register(非表格文件)
else 不支持后缀
Service-->>Client: 返回失败节点
end
Client->>Controller: POST /dm/doClean
Controller->>Service: doClean(dto)
Service->>Cleaner: createDataCleaner(strategyType)
Cleaner->>Store: 写入业务表/外部表
Cleaner->>Store: 更新 file_info 成功/失败数
图表来源
FileInfo 是建模结果的核心载体,既对应数据库 file_info 表,又作为前端树形结构的节点对象。关键字段包括:
展示字段:children、tableRuleList、tableNameEn、classify、cleanMode、cleanNote 等不在数据库中的临时属性。
classDiagram
class FileInfo {
+Long id
+Long pid
+String fileName
+String sheetName
+String tableName
+Integer templateId
+String filePath
+String fileType
+Integer dataNum
+Long fileSize
+FileSheetStateEnum fileStatus
+Integer lineNo
+Long batchId
+Integer successNum
+Integer failNum
+LocalDateTime createDateTime
+String password
+String charSet
+DelimiterEnum delimiter
+String md5
+List~FileInfo~ children
+List~TableRuleDTO~ tableRuleList
+String cleanMode
+String cleanNote
}
图表来源
章节来源
PreTask 负责打开文件流、判断 Excel/CSV、推断 CSV 编码、设置密码和分隔符,并启动 PreDataListener。若遇到不支持格式或加密异常,会把 fileStatus 标记为相应失败状态。
PreDataListener 在每行回调中:
raw_{sheetId},列全部为 VARCHAR。所有行完成后保留原文件物理行号,供 AI 清洗定位真实行号。
flowchart TD
Start(["开始"]) --> Detect["检测文件类型与编码"]
Detect --> TypeOk{"是否受支持的表格格式?"}
TypeOk -->|否| MarkUnsupported["标记 FILE_UNSUPPORTED_FAIL"]
TypeOk -->|是| OpenReader["创建 Excel/CSV 阅读器"]
OpenReader --> InitListener["初始化 PreDataListener"]
InitListener --> ReadRow["读取下一行"]
ReadRow --> HeadCheck{"前 20 行且未匹配表头?"}
HeadCheck -->|是| FindHead["尝试识别表头"]
HeadCheck -->|否| AppendRow["追加到 raw_{sheetId}"]
FindHead --> AppendRow
AppendRow --> RowWidth{"列宽超过现有列数?"}
RowWidth -->|是| GrowCols["ALTER 增加列"]
RowWidth -->|否| NextRow["继续下一行"]
GrowCols --> NextRow
NextRow --> Done{"是否读完?"}
Done -->|否| ReadRow
Done --> Close["释放 DuckDB 资源"]
Close --> End(["结束"])
图表来源
章节来源
当上传文件不是 CSV/XLS/XLSX 时,DmService 不会直接拒绝用户,而是:
file_info 登记一条 FILE_UNSUPPORTED_FAIL 记录。FileAiProfileService 的职责是“建档 + 查询”,真正的识别由后台服务异步完成。建档过程:
file_ai_profile,状态为 PENDING 或 SKIP。查询时限制最大 LIMIT,防止数据分析工具拉取过多数据。
sequenceDiagram
participant Upload as "上传接口"
participant DM as "DmService"
participant Profile as "FileAiProfileService"
participant DB as "file_ai_profile"
participant Recognize as "异步识别服务"
Upload->>DM: 上传非表格文件
DM->>DB: 写入 file_info(FILE_UNSUPPORTED_FAIL)
DM->>Profile: register(file, fileInfo)
Profile->>Profile: 抽取样本文本
Profile->>DB: 插入档案(PENDING/SKIP)
Profile->>Recognize: submit(profile, caseId, userId)
Recognize-->>DB: 回写 category/summary/keywords
图表来源
章节来源
边配置用于描述不同实体之间的关联规则,主要结构如下:
| 字段 | 含义 |
|---|---|
| clazz | 关系类型,如通话关系、转账关系或其他自定义边类型 |
| tableName | 需要识别关系的业务表名 |
| node1 | 起始节点字段 |
| node2 | 终止节点字段 |
| mc | 关系中文描述 |
| sj | 时间字段 |
这些配置持久化到 edge_conf,配合 rel_edge、rel_node、rel_node_conf 等关系表和节点配置表,供 graph 模块生成图结构。
erDiagram
EDGE_CONF {
bigint id PK
varchar clazz
varchar tableName
varchar node1
varchar node2
varchar mc
varchar sj
}
REL_EDGE {
bigint id PK
varchar node1
varchar node2
varchar node1cn
varchar node2cn
varchar mc
varchar clazz
varchar num
timestamp sj
integer fx
integer lx
integer cs
}
REL_NODE {
varchar node_id PK
varchar node_type
}
REL_NODE_CONF {
bigint id PK
varchar clazz
varchar tableName
varchar node
varchar fd
}
EDGE_CONF ||--o{ REL_EDGE : "定义关系类型"
REL_NODE ||--o{ REL_EDGE : "作为端点"
图表来源
章节来源
dm 的清洗层采用策略模式:CleanFactory 维护 StrategyType 到 DataCleaner/DataLoader 的映射,支持通话、转账、出行、外部公安数据等多种场景。
DataCleaner 定义了标准生命周期:
外部表(如 cert_info、entry_rec、vehicle_info、lic_info、veh_violation、household、co_resident、co_veh_viol、criminal 等)通过独立 SQL 文件定义表结构和字段映射,清洗器将原始 sheet 数据映射到这些表,再由 graph、agent、aiclean 等业务模块消费。
flowchart LR
Sheet["原始 sheet<br/>raw_{sheetId}"] --> Factory["CleanFactory"]
Factory --> Cleaner["DataCleaner 实现"]
Cleaner --> External["外部表<br/>cert_info / entry_rec / vehicle_info …"]
Cleaner --> Business["业务表"]
Cleaner --> Log["clean_error_log"]
图表来源
章节来源
清洗错误日志存储在 clean_error_log,关键字段包括:
CleanErrorLogService 继承 MyBatis-Plus 的 ServiceImpl,提供基础 CRUD 能力,供上层服务或控制器查询清洗失败明细。
章节来源
dm 模块内部耦合度较低,主要依赖以下方向:
所有持久化操作通过 Mapper 访问案件数据库或外部数据源。
graph LR
DmController --> DmService
DmService --> PreTask
DmService --> PreDataListener
DmService --> FileAiProfileService
DmService --> CleanFactory
CleanFactory --> DataCleaner
DmService --> FileInfo
DmService --> EdgeConf
图表来源
章节来源
建议:
章节来源
建议:
章节来源
外部表定义集中在 external_tables.sql,字段映射在 external_table_field.sql。dm 清洗器将这些外部数据标准化后写入对应表,再由其他模块进行关联分析。
协作要点:
章节来源
/dm/preFileUpload,传入单个文件或同批文件的 batchId。raw_{sheetId} 并写入原始数据。预期结果:
章节来源
注意事项:
章节来源
章节来源
/dm/getFiles 查询已导入文件列表。raw_{sheetId} 查询;如需导出清洗结果,从业务表或 external 表查询。章节来源
| 现象 | 可能原因 | 排查位置 | 处理建议 |
|---|---|---|---|
| 上传后返回失败节点 | 后缀不在允许列表中 | DmController.preFileUpload | 确认文件后缀是否为 CSV/XLS/XLSX,或接受 FILE_UNSUPPORTED_FAIL |
| Excel 无法读取 | 加密文件或不支持格式 | PreTask.call | 检查 EncryptedDocumentException 或 UnsupportedFileFormatException |
| CSV 乱码 | 编码推断错误 | PreTask、PreDataListener | 手动指定 charSet,或使用 UTF-8 保存 CSV |
| 列数不一致 | 稀疏行导致列宽变化 | PreDataListener.appendRawRow | 检查 ALTER 扩列逻辑,必要时固定列顺序 |
| AI 画像一直 Pending | 异步识别未完成或样本为空 | FileAiProfileService | 查看 sampleText、parseStatus,确认识别服务是否正常 |
| 清洗后行数不对 | 表头行、空行、过滤规则影响 | RawPreviewQuery、DataCleaner.clean | 核对 lineNo、row_no、模板匹配结果 |
| 边关系不生效 | edge_conf 配置错误或字段不存在 | EdgeConf、rel_edge | 检查 clazz、tableName、node1/node2 是否与清洗结果一致 |
| 清洗失败明细为空 | clean_error_log 未写入或查询条件错误 | CleanErrorLogService | 检查 fileId/batchId/sheetId 是否正确 |
章节来源
dm 模块以文件识别为入口,通过流式读取、DuckDB 动态表和策略化清洗,把原始 CSV/Excel 和非表格文件转化为可被 graph、agent、aiclean 消费的建模产物。其核心价值在于:
后续改进可围绕大文件并发调优、字段类型智能推断、外部表 schema 变更管理和边配置可视化编辑展开。