数据建模模块 dm

本文引用的文件

DmController.java
DmService.java
PreTask.java
PreDataListener.java
FileAiProfileService.java
FileInfo.java
EdgeConf.java
CleanFactory.java
DataCleaner.java
CleanErrorLogService.java
case_table_1.sql
external_tables.sql
external_table_field.sql

简介

dm（data model）是连接原始数据与业务查询的中间层。它负责：

文件识别：读取 Excel、CSV，推断表头、列数、编码和分隔符。
自动建模：为每个 sheet 创建 DuckDB 动态表，匹配模板或新建目标表。
AI 画像：对无法解析为表格的非表格文件抽取样本文本，提交大模型分类、摘要、关键词。
边配置：定义实体间关系，把清洗后的数据映射成图节点和边。
错误日志：记录清洗失败行，供前端查看“导入失败”明细。

对外暴露的 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.java:1-179
DmService.java:1-200
PreTask.java:1-187
PreDataListener.java:1-200
FileAiProfileService.java:1-200
CleanFactory.java:1-166

章节来源

DmController.java:1-179
case_table_1.sql:1-80

核心组件

组件	职责	关键输出
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`

章节来源

DmController.java:1-179
DmService.java:1-200
PreTask.java:1-187
PreDataListener.java:1-200
FileAiProfileService.java:1-200
CleanFactory.java:1-166
EdgeConf.java:1-52
CleanErrorLogService.java:1-10

架构总览

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 成功/失败数

图表来源

DmController.java:60-120
DmService.java:120-200
PreTask.java:70-170
PreDataListener.java:120-200
FileAiProfileService.java:60-140
CleanFactory.java:1-166

详细组件分析

文件信息模型 FileInfo

FileInfo 是建模结果的核心载体，既对应数据库 file_info 表，又作为前端树形结构的节点对象。关键字段包括：

文件标识：id、pid、batchId、md5、filePath、fileName。
结构标识：tableName、templateId、lineNo、header、delimiter、charSet、password。
统计字段：fileSize、dataNum、successNum、failNum、fileStatus。

展示字段：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
}

图表来源

FileInfo.java:1-203

章节来源

FileInfo.java:1-203
case_table_1.sql:1-38

文件识别流程

PreTask 负责打开文件流、判断 Excel/CSV、推断 CSV 编码、设置密码和分隔符，并启动 PreDataListener。若遇到不支持格式或加密异常，会把 fileStatus 标记为相应失败状态。

PreDataListener 在每行回调中：

构建当前 sheet 的 FileInfo。
前 20 行尝试匹配表头。
首行懒初始化 DuckDB 动态表 raw_{sheetId}，列全部为 VARCHAR。
对稀疏行按最大列索引扩列，追加 fileId、sheetId、row_no 和业务列。

所有行完成后保留原文件物理行号，供 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(["结束"])

图表来源

PreTask.java:70-170
PreDataListener.java:120-200

章节来源

PreTask.java:1-187
PreDataListener.java:1-200

AI 画像：非表格文件识别

当上传文件不是 CSV/XLS/XLSX 时，DmService 不会直接拒绝用户，而是：

不落盘解析表格。
在 file_info 登记一条 FILE_UNSUPPORTED_FAIL 记录。
调用 FileAiProfileService.register 为非表格文件建档。

FileAiProfileService 的职责是“建档 + 查询”，真正的识别由后台服务异步完成。建档过程：

检查是否为可识别文本类文件。
从 MultipartFile 抽取样本文本。
插入 file_ai_profile，状态为 PENDING 或 SKIP。
有样本则提交异步识别；无样本直接标 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

图表来源

DmService.java:120-200
FileAiProfileService.java:1-200
case_table_1.sql:39-80

章节来源

FileAiProfileService.java:1-200
case_table_1.sql:39-80

边配置与实体关系

边配置用于描述不同实体之间的关联规则，主要结构如下：

字段	含义
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 : "作为端点"

图表来源

EdgeConf.java:1-52
case_table_1.sql:120-190

章节来源

EdgeConf.java:1-52
case_table_1.sql:120-190

数据清洗与外部表协作

dm 的清洗层采用策略模式：CleanFactory 维护 StrategyType 到 DataCleaner/DataLoader 的映射，支持通话、转账、出行、外部公安数据等多种场景。

DataCleaner 定义了标准生命周期：

init：根据 funcRegx 和 FileInfo 初始化规则。
prepare：批量预处理，例如加载字典。
clean：单行清洗，返回 CleanResult。
after：成功后后置处理。
finish：批量结束后释放资源。

外部表（如 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"]

图表来源

CleanFactory.java:1-166
DataCleaner.java:1-59
external_tables.sql:1-200
external_table_field.sql:1-80

章节来源

CleanFactory.java:1-166
DataCleaner.java:1-59
external_tables.sql:1-200
external_table_field.sql:1-80

错误日志格式

清洗错误日志存储在 clean_error_log，关键字段包括：

fileId：来源文件 ID。
batchId：清洗批次。
sheetId：来源 sheet ID。
fileName：文件名。
sheetName：工作表名。
content：错误内容或失败行信息。
createDateTime：记录时间。

CleanErrorLogService 继承 MyBatis-Plus 的 ServiceImpl，提供基础 CRUD 能力，供上层服务或控制器查询清洗失败明细。

章节来源

CleanErrorLogService.java:1-10
case_table_1.sql:230-260

依赖关系分析

dm 模块内部耦合度较低，主要依赖以下方向：

控制器依赖服务：DmController → DmService。
服务依赖预处理、AI 画像、清洗工厂和外部服务：DmService → PreTask、PreDataListener、FileAiProfileService、CleanFactory。
清洗层依赖策略枚举和具体 cleaner/loader：CleanFactory → DataCleaner/DataLoader。

所有持久化操作通过 Mapper 访问案件数据库或外部数据源。

graph LR
DmController --> DmService
DmService --> PreTask
DmService --> PreDataListener
DmService --> FileAiProfileService
DmService --> CleanFactory
CleanFactory --> DataCleaner
DmService --> FileInfo
DmService --> EdgeConf

图表来源

DmController.java:1-179
DmService.java:1-200
CleanFactory.java:1-166
FileInfo.java:1-203
EdgeConf.java:1-52

章节来源

DmController.java:1-179
CleanFactory.java:1-166

性能与扩展性

大文件识别优化

并发控制：DmService 使用虚拟线程池和信号量限制预览并发，默认最大并发为 20，带超时保护。
流式读取：PreTask 使用 FesodSheet 流式读取，避免一次性加载整个 Excel。
动态表写入：PreDataListener 用 DuckDBAppender 追加行，首行懒建表，列宽不足时 ALTER 扩列。
内存保护：Raw 表列类型为 VARCHAR，避免复杂类型转换带来的额外开销。

建议：

对超大目录分批上传，避免单次 batchId 过大。
优先选择 CSV 而非多 sheet 大 Excel，减少工作表遍历成本。
对重复文件利用 md5 去重，避免重复解析。

章节来源

DmService.java:1-200
PreTask.java:70-170
PreDataListener.java:120-200

字段类型推断准确率提升

当前 raw 表统一使用 VARCHAR，降低类型推断风险。
模板匹配采用部分匹配比例和最小列数下限，避免小模板抢占匹配。
对于结构化外部表，字段映射以 external_table_field.sql 为准，人工维护更可靠。

建议：

对金额、时间、身份证号等关键字段在清洗阶段显式转换类型。
对不确定字段保留原始字符串列，并在查询侧做类型转换。
定期校验模板匹配率，补充缺失模板或调整匹配阈值。

章节来源

PreDataListener.java:120-200
external_table_field.sql:1-80

与外部表 external 的协作方式

外部表定义集中在 external_tables.sql，字段映射在 external_table_field.sql。dm 清洗器将这些外部数据标准化后写入对应表，再由其他模块进行关联分析。

协作要点：

外部表结构与字段顺序以 SQL 为准，新增字段需同步更新 SQL 和清洗器。
清洗失败时，clean_error_log 应记录可回溯的行内容和错误原因。
如果外部数据来源发生变化，应先评估是否需要新增 StrategyType 和 cleaner。

章节来源

external_tables.sql:1-200
external_table_field.sql:1-80
CleanFactory.java:1-166

使用示例

上传 CSV/Excel 进行自动建模

前端调用 /dm/preFileUpload，传入单个文件或同批文件的 batchId。
服务端校验后缀：CSV/XLS/XLSX 落盘并解析，其他后缀登记 FILE_UNSUPPORTED_FAIL。
PreTask 检测类型和编码，PreDataListener 创建 raw_{sheetId} 并写入原始数据。
返回树形 FileInfo：根节点为文件，children 为各 sheet。

预期结果：

file_status 显示文件读取完成或失败。
每个 sheet 包含 tableName、templateId、lineNo、header 等建模信息。
非表格文件可在 file_ai_profile 中查看 AI 识别结果。

章节来源

DmController.java:60-120
DmService.java:120-200
PreTask.java:70-170

配置实体边

在 edge_conf 中新增关系类型 clazz、目标表 tableName、node1/node2、mc、sj。
在 rel_node_conf 中配置节点字段映射。
graph 模块根据 edge_conf 和 rel_edge 渲染实体关系图。

注意事项：

clazz 应与清洗器的 StrategyType 保持一致语义。
node1/node2 必须存在于对应表的清洗结果中。
若关系涉及时间过滤，确保 sj 字段正确映射。

章节来源

EdgeConf.java:1-52
case_table_1.sql:120-190

查看 AI 画像

上传非表格文件后，检查 file_info 中是否存在 FILE_UNSUPPORTED_FAIL。
在 file_ai_profile 中查看该文件的 parseStatus、category、summary、keywords。
若 parseStatus 为 PENDING，等待异步识别完成；若为 SKIP，说明未抽取到文本样本。

章节来源

FileAiProfileService.java:60-140
case_table_1.sql:39-80

导出建模结果

使用 /dm/getFiles 查询已导入文件列表。
对已完成清洗的文件，查看 successNum/failNum。
如需导出原始结构，可从 DuckDB 的 raw_{sheetId} 查询；如需导出清洗结果，从业务表或 external 表查询。
对失败明细，通过 clean_error_log 查看 content 并修正数据或清洗规则。

章节来源

DmController.java:120-179
case_table_1.sql:230-260

故障排查指南

现象	可能原因	排查位置	处理建议
上传后返回失败节点	后缀不在允许列表中	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 是否正确

章节来源

PreTask.java:70-170
PreDataListener.java:120-200
FileAiProfileService.java:60-140
CleanErrorLogService.java:1-10

结论

dm 模块以文件识别为入口，通过流式读取、DuckDB 动态表和策略化清洗，把原始 CSV/Excel 和非表格文件转化为可被 graph、agent、aiclean 消费的建模产物。其核心价值在于：

把“未知结构”的文件变成“可查询”的 raw 表和业务表。
把“无法解析”的非表格文件变成“可理解”的 AI 画像。
把“分散数据”通过边配置整合成“可分析”的实体关系。
把“清洗失败”从静默丢弃变成“可追溯”的错误日志。

后续改进可围绕大文件并发调优、字段类型智能推断、外部表 schema 变更管理和边配置可视化编辑展开。

数据建模模块 dm.md 26 KB История Исходник

数据建模模块 dm

目录

简介

项目结构

核心组件

架构总览

详细组件分析

文件信息模型 FileInfo

文件识别流程

AI 画像：非表格文件识别

边配置与实体关系

数据清洗与外部表协作

错误日志格式

依赖关系分析

性能与扩展性

大文件识别优化

字段类型推断准确率提升

与外部表 external 的协作方式

使用示例

上传 CSV/Excel 进行自动建模

配置实体边

查看 AI 画像

导出建模结果

故障排查指南

结论

数据建模模块 dm.md 26 KB

История Исходник