涉及自然语言处理的一些知识

CLIP(Contrastive Language–Image Pre-training):把文本和图片映射到同一个向量空间里，然后比较它们的语义是否接近。

CLIP 特别适合做这些事情：

• 图文匹配：一张图和一句话是否描述同一内容
• 图片检索：输入“穿红衣服的人”，找相关图片
• 零样本分类：不用专门训练分类器，也可以判断图片属于“猫 / 狗 / 汽车”
• 多模态聚类：把语义相似的文字和图片聚到一起

Milvus：是一个向量数据库，向量相似度查询，例如“找最像这个语义的内容”

AlignmentPipeline：它把已经预处理好的资源进一步转成“主题语义表示”，再把资源组织成事件，并通过即时 + 全局校验尽量修正错误事件绑定。

局部即时校验：每绑定一条马上检查

全局校验：所有资源绑定完成后再处理

wash.json表示的是“当前这一轮数据预处理产生的完整数据”，而不是永久积累所有历史数据

pipeline_data_preprocessing.py:

配置给你、数据库连接给你，然后我把真正干活的SourceIngestionPipeline叫过来，让它处理当前这一轮数据。它处理完后，把结果告诉我，我再把结果往上层返回。

T_Source:

pipline_source_ingestion.py:

可以把这个文件夹理解成原始数据入库前的总负责人。

它做的事情不是某一个单独动作，而是把下面这些动作串起来：原始数据先被外部采集程序放进resource_receive。这个文件会去扫描里面有哪些 Twitter、Wiki、Website、Facebook、News 或 PDF 数据。找到以后，它不会马上处理，而是先判断这一整个批次是不是已经“写完了”。

为什么要等稳定？因为外部程序可能还在持续往目录里写文件。如果现在就去读，可能只能读到半份 BCP 数据。所以程序会连续比较目录的状态，包括文件数量、总大小、最新修改时间；连续若干轮都没变化，才认为这一批数据已经稳定，可以正式处理。默认是连续 2 轮稳定。

批次稳定以后，程序先把整个批次原样复制一份到resource_target。这相当于先做一个备份和存档，防止以后原始数据丢失，也方便后续追溯。

接下来开始真正处理数据。如果是 Twitter、Wiki、Website、Facebook、News 这种信源，它会找到该目录里的唯一.bcp文件，然后逐行读取里面的 JSON 数据。每一条原始 JSON 数据都会交给SourceRecordMapper，把原始字段转换成T_Source所需要的统一字段。

映射完成以后，如果当前不是dry_run模式，就调用：repository.upsert_source(source_data)把数据正式写进 MySQL 的T_Source。同时，这些映射完成的数据也会暂时保存到wash_items里面，准备后面生成wash.json。

PDF 的处理稍微不一样。PDF 不需要找 BCP，而是直接读取 PDF 文件，一个 PDF 最终对应一条T_Source记录。如果一个文件夹里有多个 PDF，就逐个处理，每个 PDF 产生一条记录。等当前整个批次里的所有信源都处理完以后，程序会把这些标准化后的数据交给：BatchWashJsonBuilder统一生成“大 wash.json”。

如果这一轮同时有多个稳定批次，第一个批次会覆盖旧的wash.json，后面的批次继续追加，最后这一轮所有稳定批次的数据会合并到同一个大wash.json里面。

最后，这个批次会被标记成ingested，以后再次扫描到它时，就不会重复处理，从而避免重复入库。

实现的功能：（ 6 个核心功能）

1. 发现数据：从resource_receive里找到需要处理的批次和信源目录。
2. 判断数据是否完整：通过目录快照判断批次是否已经稳定，避免读取还没写完的数据。
3. 备份原始数据：稳定以后复制到resource_target保存原始版本。
4. 把不同来源的数据转成统一格式：BCP、PDF 最终都转换成T_Source的标准字段。
5. 写数据库：调用MySQLRepository.upsert_source()写入T_Source。

解决了什么问题

1.解决“数据还没写完就被读取”的问题：外部程序还在写 BCP，你的程序已经开始读，读到一半数据，数据库数据不完整。通过目录快照，连续两轮不变，才开始处理。

2.解决“不同数据来源格式不统一”的问题：数据来源很多，每种原始结构可能都不一样，需要把他们统一变成T_Source 标准字段，后面的模型就不用再管了

3.解决“重复处理同一个批次”的问题：程序会记录folder_status.json，如果某个批次已经是：ingested，下次再运行的时候会直接跳过。

一句话总结：它负责把resource_receive里的原始批次，等到数据稳定后，先备份，再解析、映射成统一的T_Source数据，写进 MySQL，同时生成给后续模型使用的大wash.json。

source_record_mapper.py:

这个文件不负责读取 BCP，也不负责真正操作数据库。它负责把“读出来的一条原始数据”加工成“一条可以直接写入 T_Source 的标准数据”。

一句话总结：把 Twitter、Wiki、Website、Facebook、News、PDF 等不同来源的数据统一转换成 T_Source 标准格式。

词语解释：

tweet_topic= 不同信源最终统一整理出来的“核心文本内容”。所有有意义的文本都尽量进入tweet_topic，供后续主题抽取和事件匹配使用。

SourceRecordMapper：信源记录映射器（原始数据格式转换器）把不同来源的一条原始记录，转换成统一格式。

解决了什么问题

如果没有这个文件，后面的数据库可能面对Twitter一种格式、Wiki一种格式，后面代码可能就得写Twitter怎么处理、Wiki怎么处理，每一种格式都要分别处理，非常乱。

它解决的核心问题是：把多源异构数据统一成一种标准格式，让后面的数据库、topic 提取、事件对齐模块不用关心数据最开始是从哪里来的。

我提出的问题：bcp信源交给 source_record_mapper和pdf信源交给 pdf_source_processor都是为了进行字段映射吗 ，字段映射之后再统一字段吗？

准确描述：BCP 数据本身已经是结构化记录，可以直接交给 source_record_mapper做字段映射；PDF是非结构化文件，需要先交给pdf_source_processor 提取正文和图片，然后再调用source_record_mapper ，映射成统一的 T_source 字段。

• “解析”是把文件内容读取出来；
• “清洗”是处理空值、时间、列表和异常格式；
• “字段映射”是把不同名称、不同结构转换成统一字段；
• 映射结束后，得到的已经是统一的T_Source数据。

说明：原始数据可能使用不同名称表示正文：text、context、body、article_content，经过source_record_mapper后，都会整理到统一字段中，例如 tweet_topic,时间、URL和id也是如此。

说明：pdf_source_processor不只是做字段映射，它主要负责：

打开和解析 PDF；

• 提取正文；
• 提取 Markdown；
• 整理PDF中的图片；
• 复制或记录相关资源文件；
• MinerU失败时进行降级解析；
• 最后调用SourceRecordMapper完成标准字段映射。

source_topic_builder.py

它的任务不是写数据库，而是把各个信源里的标题、正文、转推内容、Wiki 内容等整理成一个干净的tweet_topic，供后面的主题抽取、事件匹配、事件对齐继续使用。

普通文本处理 纯图片数据处理

一句话总结：先把各种原始文本转成字符串 → 再把多个文本拼起来 → 再统一清洗 → 最后得到干净的tweet_topic。如果只有图片，没有文本，就尝试通过多模态模型生成一段文本，再走同样的清洗流程。

专业术语解释：

resource_receive:资源接受目录，可理解为“系统收货区”

这是外部采集程序投放原始数据的入口。Twitter、新闻、Wiki、Facebook、PDF等采集结果先放到这里。

watching：正在观察，可以理解为：货已经到了，但还没卸完，先不要处理。

表示系统已经发现某个采集批次，但还不能确认文件是否传输完成。系统会连续检查：文件数量、文件总大小、最后修改时间，如果这些数据还在变化，就保持watching。

resource_target：资源目标目录，可以理解为：验收后的正式存档区。

采集批次确认稳定后，会从resource_receive复制到resource_target，用于：保存原始数据副本、后续处理 、出错后重新运行。

iterator_status：迭代器状态文件，可以理解为：批次处理进度表或书签。

迭代器状态文件，通常记录：当前处理到第几条；总共有多少条；当前资源的source_id；批量wash.json的文件指纹；当前状态；上一条成功完成的资源。

item_in_progress：表示当前有一条资源正在处理中，可以理解为：这条任务已经发出，但还没收到“完成”确认。

此时游标不会前进。如果程序中断，下次仍会重新输出这条资源。

mark_done：表示明确告诉迭代器：“当前这条处理成功了。可以理解为“消息队列中的“确认收货”。

它会 记录当前资源已经完成；游标加一；清除当前wash.json；准备处理下一条。

batch_done：表示当前批量wash.json中允许处理的资源已经全部完成。

MinerU：

BCP：原始数据文件，就是装原始信源数据的文件容器，后续程序从这里把数据读出来再做统一处理，一行 = 一条json记录。

source_id:一条资源在整个系统中的唯一编号

tweet_topic:资源的主要文本内容

T_Source:原始资源表,可以理解为：标准化后的原材料仓库。

T_Derived:派生信息表

wash_items

wash.json:wash 在这里表示“清洗并标准化后的数据”，不是一种特殊文件格式，本质上还是JSON。项目中有两种wash.json.

大 wash.json = 一批任务 当前 wash.json = 现在正在处理的一条任务

event_id:一个事件的唯一编号

core_id:当前事件的代表核心资源ID

core_ids:当前事件包含的所有核心资源ID

association_ids:与事件相关、但不作为事件事实核心的资源ID

core_ids = 用来证明或定义这件事的材料 association_ids = 围绕这件事产生的相关讨论材料

source_id:一条资源

source_ids:多条资源ID组成的列表

T_Entity_Bind：实体绑定表，通常一个QID对应一行

继承父推文绑定：推文之间可能存在回复、转发、引用，其中当前推文是“子推文”，被回复、转发或引用的推文是“父推文”。如果父推文已经绑定到某个事件，子推文可以直接继承父推文的事件绑定。口语化理解：父推文已经确定在讨论某个事件，那么它的回复、转发或引用大概率也在讨论这个事件，所以先把子推文跟过去，再由语义校验进行复核。

MinerU：一个文档解析工具，主要用于处理PDF。项目中主要使用MinerU提取PDF文本和图片。如果MinerU不能运行，还会尝试普通PDF文本提取作为降级方案。

Milvus:一个向量数据库，Milvus擅长查询：哪段文字或哪张图片在语义上最相似

Markdown：一种轻量文本格式

PDF或网页解析结果中可能包含Markdown格式。进入模型前，Align1会根据配置删除部分Markdown符号，减少格式噪声。

HTML：网页的结构标记语言，网页采集内容可能混有HTML标签。预处理或文本融合时通常会去除标签，只保留可读文字。

URL:网页或网络资源地址,URL可以用于保存原始资源位置，但如果直接混入关键词和主题文本，可能产生无意义的词元，所以Align1通常会在融合文本中移除URL。

学习顺序

pipline_data_preprocessing.py：数据预处理总开T_Source

先判断外部有没有传配置，没有就自动创建config_manager。从config.yaml中读取：

1. resource_receive在哪里；
2. resource_target在哪里；
3. 批次稳定需要检查几轮；
4. wash.json写到哪里；
5. MySQL连接参数等。

再判断外部有没有传数据库操作对象，如果没有，就根据配置创建一个，用于写入T_Source。

1.扫描接受目录，将这些信源按照resource_receive下的一级目录归并成批次。整个batch_001被当成一个批次。

resource_receive/
└── batch_001/
├── twitter-xxx/
├── news-xxx/
└── documents/

2.判断批次是否传输完成，系统不会发现问题立马处理，因为采集程序可能还在向目录写数据。它会记录目标快照：文件数量、文件总大小、最后修改时间，和上一轮比较。如果快照发生变化，说明文件还在传输，状态保持watching，暂时不处理，如果连续多轮快照没有变化，说明文件基本传输完毕，并且批次状态变为stable，开始处理。

3，备份原始批次，批次稳定后，将整个目录复制到resource_target，目的是为了保留一份原始数据，后续人工检查，出错后重新处理，防止接受目录中的文件丢失。

4.解析不同类型的数据，程序根据资源类型不同选择不同的解析的方式。BCP信源里的Twitter、Wiki通常读取对应目录中的.bcp文件，项目里的BCP按照JSONL处理，一行 = 一条JSON资源。然后逐条交给source_record_mapper进行字段映射。PDF信源交给pdf_source_processor.对于pdf先尝试MinerU解析，提取正文和图片，MinerU失败了再使用普通pdf文本提取，将pdf整理成标准资源。

5.统一字段，不同信源的字段不同，因此要转换成统一的T_Source结构。

6.写入T_source,每条资源映射完成后，会执行repository.upsert_source(source_data)，upsert表示：数据不存在就新增、数据已经存在就更新、重跑时不会简单的重复插入。如果使用了dry_run，会跳过数据库写入，但仍会完成扫描和字段转换。

7.生成批量wash.json（wash_items表示多条已经整理好资源，wash.json是把wash_items保存到磁盘形成的json文件），所有资源处理完成后，把映射结果汇总成批量wash.json（大致结构如下）。同一轮如果有多个稳定批次：第一个稳定批次 → 覆盖原来的大 wash.json，
后面的稳定批次 → 继续追加到这份 wash.json。所以一轮结束后，当前稳定批次会被合并成一份大wash.json。生成新文件后，还会重置washjson_iterator的游标状态，让后续从新批次的第一条开始处理。

[
{
"source_id": "source_001",
"source_type": "news",
"tweet_topic": "..."
},
{
"source_id": "source_002",
"source_type": "twitter",
"tweet_topic": "..."
}
]

8.保存批次状态并返回结果，最后更新folder_status.json,记录哪些批次还在观察、哪些已经稳定、哪些已经复制、哪些已经入库、哪些处理失败、每个批次生成了多少条资源。

为什么预处理只运行一轮？

预处理只运行一轮，它把当前所有稳定批次处理完之后立即返回，让外出总流水线继续执行派生处理、对齐等阶段任务。否则预处理如果一直监听，就永远不会把程序控制权交给后面的模块。我们设置了“最多扫描三轮、每轮等待31秒”，在pipline_source_preprocessing.py中有写。

这个文件不负责什么？

它不负责：

• 生成单条当前now_washjson/wash.json；
• 遍历大wash.json；
• 文本、图片、视频派生处理；
• 实体消歧；
• Align1主题建模；
• Align2事件对齐；
• 资源绑定。

原始采集批次 → 判断稳定 → 备份 → 解析和统一字段 → 写入T_Source → 生成批量wash.json

总结：它先看收货区有没有新货，再看看货是不是已经送完。货没送完就继续等；货送完了就复制一份留档，然后让不同工作人员拆解BCP和PDF，把各种格式统一贴上系统标签，放进T_Source仓库。最后再列一张大清单wash.json，交给后面的流水线逐条处理。它做完这一轮就下班，不会一直霸占程序。

写入当前单条wash.json”的意思是：

从包含很多条数据的批量wash.json中，取出当前要处理的一条数据，单独保存到另一个wash.json文件中，供后续模块读取。

Align2

事件指纹：时间、地点、实体

预处理之后的完整流程

数据预处理流程

Align1流程

Align2流程

事件对齐语义校验

资源绑定