-## License
-
-版权说明：本项目在 Ragflow 项目基础上进行二开，需要遵守 Ragflow 的开源协议，如下
-
-This repository is available under the [Ragflow
- Open Source License](LICENSE), which is essentially Apache 2.0 with a few additional restrictions.
-
## 鸣谢
本项目基于以下开源项目开发：
diff --git a/api/apps/document_app.py b/api/apps/document_app.py
index ec7db5f..c27a4d9 100644
--- a/api/apps/document_app.py
+++ b/api/apps/document_app.py
@@ -349,6 +349,8 @@ def rm():
@validate_request("doc_ids", "run")
def run():
req = request.json
+
+ # 检查用户是否有权限操作这些文档
for doc_id in req["doc_ids"]:
if not DocumentService.accessible(doc_id, current_user.id):
return get_json_result(
@@ -357,29 +359,42 @@ def run():
code=settings.RetCode.AUTHENTICATION_ERROR
)
try:
+ # 遍历所有需要处理的文档ID
for id in req["doc_ids"]:
+ # 准备更新文档状态的信息
info = {"run": str(req["run"]), "progress": 0}
+ # 如果是运行状态且需要删除旧数据，则重置统计信息
if str(req["run"]) == TaskStatus.RUNNING.value and req.get("delete", False):
info["progress_msg"] = ""
info["chunk_num"] = 0
info["token_num"] = 0
+
+ # 更新文档状态
DocumentService.update_by_id(id, info)
+ # 获取租户ID
tenant_id = DocumentService.get_tenant_id(id)
if not tenant_id:
return get_data_error_result(message="Tenant not found!")
+ # 获取文档详情
e, doc = DocumentService.get_by_id(id)
if not e:
return get_data_error_result(message="Document not found!")
+ # 如果需要删除旧数据
if req.get("delete", False):
+ # 删除相关任务记录
TaskService.filter_delete([Task.doc_id == id])
+ # 如果索引存在，则删除索引中的文档数据
if settings.docStoreConn.indexExist(search.index_name(tenant_id), doc.kb_id):
settings.docStoreConn.delete({"doc_id": id}, search.index_name(tenant_id), doc.kb_id)
-
+
+ # 如果是运行状态，则创建解析任务
if str(req["run"]) == TaskStatus.RUNNING.value:
e, doc = DocumentService.get_by_id(id)
doc = doc.to_dict()
doc["tenant_id"] = tenant_id
+ # 获取文档存储位置
bucket, name = File2DocumentService.get_storage_address(doc_id=doc["id"])
+ # 将任务加入队列
queue_tasks(doc, bucket, name)
return get_json_result(data=True)
diff --git a/api/db/services/document_service.py b/api/db/services/document_service.py
index 16482c8..470f24a 100644
--- a/api/db/services/document_service.py
+++ b/api/db/services/document_service.py
@@ -515,6 +515,7 @@ def doc_upload_and_parse(conversation_id, file_objs, user_id):
ParserType.EMAIL.value: email
}
parser_config = {"chunk_token_num": 4096, "delimiter": "\n!?;。；！？", "layout_recognize": "Plain Text"}
+ # 使用线程池执行解析任务
exe = ThreadPoolExecutor(max_workers=12)
threads = []
doc_nm = {}
diff --git a/api/db/services/task_service.py b/api/db/services/task_service.py
index 2d3147c..d61feef 100644
--- a/api/db/services/task_service.py
+++ b/api/db/services/task_service.py
@@ -202,76 +202,138 @@ class TaskService(CommonService):
def queue_tasks(doc: dict, bucket: str, name: str):
+ """
+ 将文档解析任务分割并加入队列处理。
+
+ 该函数根据文档类型(PDF、表格等)将文档分割成多个子任务，计算任务摘要，
+ 检查是否可以重用之前的任务结果，并将未完成的任务加入Redis队列进行处理。
+
+ 参数:
+ doc (dict): 文档信息字典，包含id、type、parser_id、parser_config等信息
+ bucket (str): 存储桶名称
+ name (str): 文件名称
+
+ 流程:
+ 1. 根据文档类型(PDF/表格)将文档分割成多个子任务
+ 2. 为每个任务生成唯一摘要(digest)
+ 3. 尝试重用之前任务的处理结果
+ 4. 清理旧任务并更新文档状态
+ 5. 将新任务批量插入数据库
+ 6. 将未完成的任务加入Redis队列
+ """
def new_task():
+ """
+ 创建一个新的任务字典，包含基本任务信息。
+
+ 返回:
+ dict: 包含任务ID、文档ID、进度和页面范围的任务字典
+ """
return {"id": get_uuid(), "doc_id": doc["id"], "progress": 0.0, "from_page": 0, "to_page": 100000000}
+ # 存储所有解析任务的数组
parse_task_array = []
+ # PDF文档处理逻辑
if doc["type"] == FileType.PDF.value:
+ # 从存储中获取文件内容
file_bin = STORAGE_IMPL.get(bucket, name)
+ # 获取布局识别方式，默认为"DeepDOC"
do_layout = doc["parser_config"].get("layout_recognize", "DeepDOC")
+ # 获取PDF总页数
pages = PdfParser.total_page_number(doc["name"], file_bin)
+ # 获取每个任务处理的页数，默认为12页
page_size = doc["parser_config"].get("task_page_size", 12)
+ # 对于学术论文类型，默认任务页数为22
if doc["parser_id"] == "paper":
page_size = doc["parser_config"].get("task_page_size", 22)
+ # 对于特定解析器或非DeepDOC布局识别，将整个文档作为一个任务处理
if doc["parser_id"] in ["one", "knowledge_graph"] or do_layout != "DeepDOC":
page_size = 10 ** 9
+ # 获取需要处理的页面范围，默认为全部页面
page_ranges = doc["parser_config"].get("pages") or [(1, 10 ** 5)]
+ # 根据页面范围和任务页数分割任务
for s, e in page_ranges:
+ # 调整页码（从0开始）
s -= 1
s = max(0, s)
+ # 确保结束页不超过文档总页数
e = min(e - 1, pages)
+ # 按照任务页数分割任务
for p in range(s, e, page_size):
task = new_task()
task["from_page"] = p
task["to_page"] = min(p + page_size, e)
parse_task_array.append(task)
+ # 表格文档处理逻辑
elif doc["parser_id"] == "table":
+ # 从存储中获取文件内容
file_bin = STORAGE_IMPL.get(bucket, name)
+ # 获取表格总行数
rn = RAGFlowExcelParser.row_number(doc["name"], file_bin)
+ # 每3000行作为一个任务
for i in range(0, rn, 3000):
task = new_task()
task["from_page"] = i
task["to_page"] = min(i + 3000, rn)
parse_task_array.append(task)
+ # 其他类型文档，整个文档作为一个任务处理
else:
parse_task_array.append(new_task())
+ # 获取文档的分块配置
chunking_config = DocumentService.get_chunking_config(doc["id"])
+ # 为每个任务生成唯一摘要(digest)
for task in parse_task_array:
+ # 创建哈希对象
hasher = xxhash.xxh64()
+ # 对分块配置中的每个字段进行哈希
for field in sorted(chunking_config.keys()):
if field == "parser_config":
+ # 移除不需要参与哈希计算的特定配置项
for k in ["raptor", "graphrag"]:
if k in chunking_config[field]:
del chunking_config[field][k]
+ # 将配置字段添加到哈希计算中
hasher.update(str(chunking_config[field]).encode("utf-8"))
+ # 将任务特定字段添加到哈希计算中
for field in ["doc_id", "from_page", "to_page"]:
hasher.update(str(task.get(field, "")).encode("utf-8"))
+ # 生成任务摘要并设置初始进度
task_digest = hasher.hexdigest()
task["digest"] = task_digest
task["progress"] = 0.0
+ # 获取文档之前的任务记录
prev_tasks = TaskService.get_tasks(doc["id"])
+ # 记录重用的块数量
ck_num = 0
if prev_tasks:
+ # 尝试重用之前任务的处理结果
for task in parse_task_array:
ck_num += reuse_prev_task_chunks(task, prev_tasks, chunking_config)
+ # 删除文档之前的任务记录
TaskService.filter_delete([Task.doc_id == doc["id"]])
+ # 收集需要删除的块ID
chunk_ids = []
for task in prev_tasks:
if task["chunk_ids"]:
chunk_ids.extend(task["chunk_ids"].split())
+ # 从文档存储中删除这些块
if chunk_ids:
settings.docStoreConn.delete({"id": chunk_ids}, search.index_name(chunking_config["tenant_id"]),
chunking_config["kb_id"])
+ # 更新文档的块数量
DocumentService.update_by_id(doc["id"], {"chunk_num": ck_num})
+ # 将新任务批量插入数据库
bulk_insert_into_db(Task, parse_task_array, True)
+ # 开始解析文档
DocumentService.begin2parse(doc["id"])
+ # 筛选出未完成的任务
unfinished_task_array = [task for task in parse_task_array if task["progress"] < 1.0]
+ # 将未完成的任务加入Redis队列
for unfinished_task in unfinished_task_array:
assert REDIS_CONN.queue_product(
SVR_QUEUE_NAME, message=unfinished_task
diff --git a/deepdoc/parser/docx_parser.py b/deepdoc/parser/docx_parser.py
index dfe3f37..a15acca 100644
--- a/deepdoc/parser/docx_parser.py
+++ b/deepdoc/parser/docx_parser.py
@@ -23,16 +23,65 @@ from io import BytesIO
class RAGFlowDocxParser:
+ """
+ Word文档(.docx)解析器，用于提取文档中的文本内容和表格。
+
+ 该解析器能够：
+ 1. 按页面范围提取文档中的段落文本及其样式
+ 2. 识别文档中的表格并将其转换为结构化文本
+ 3. 智能处理表格头部和内容，生成语义化的文本描述
+ """
def __extract_table_content(self, tb):
+ """
+ 从Word表格对象中提取内容并转换为DataFrame
+
+ 参数:
+ tb: docx库的Table对象
+
+ 返回:
+ 处理后的表格内容文本列表
+ """
df = []
for row in tb.rows:
df.append([c.text for c in row.cells])
return self.__compose_table_content(pd.DataFrame(df))
def __compose_table_content(self, df):
+ """
+ 将表格DataFrame转换为语义化的文本描述
+
+ 通过识别表格的结构特征(如表头、数据类型等)，将表格转换为更易于理解的文本形式
+
+ 参数:
+ df: 包含表格内容的DataFrame
+
+ 返回:
+ 表格内容的文本表示列表
+ """
def blockType(b):
+ """
+ 识别单元格内容的类型
+
+ 通过正则表达式和文本特征分析，将单元格内容分类为不同类型：
+ - Dt: 日期类型
+ - Nu: 数字类型
+ - Ca: 代码/ID类型
+ - En: 英文文本
+ - NE: 数字和文本混合
+ - Sg: 单字符
+ - Tx: 短文本
+ - Lx: 长文本
+ - Nr: 人名
+ - Ot: 其他类型
+
+ 参数:
+ b: 单元格文本内容
+
+ 返回:
+ 内容类型的字符串标识
+ """
patt = [
("^(20|19)[0-9]{2}[年/-][0-9]{1,2}[月/-][0-9]{1,2}日*$", "Dt"),
(r"^(20|19)[0-9]{2}年$", "Dt"),
@@ -62,14 +111,21 @@ class RAGFlowDocxParser:
return "Ot"
+ # 表格至少需要两行才能处理
if len(df) < 2:
return []
+
+ # 统计表格中最常见的内容类型
max_type = Counter([blockType(str(df.iloc[i, j])) for i in range(
1, len(df)) for j in range(len(df.iloc[i, :]))])
max_type = max(max_type.items(), key=lambda x: x[1])[0]
+ # 获取表格列数
colnm = len(df.iloc[0, :])
- hdrows = [0] # header is not nessesarily appear in the first line
+ # 默认第一行为表头
+ hdrows = [0] # 表头不一定出现在第一行
+
+ # 如果表格主要是数字类型，则识别非数字行作为表头
if max_type == "Nu":
for r in range(1, len(df)):
tys = Counter([blockType(str(df.iloc[r, j]))
@@ -78,18 +134,26 @@ class RAGFlowDocxParser:
if tys != max_type:
hdrows.append(r)
+ # 处理表格内容，将每行转换为文本
lines = []
for i in range(1, len(df)):
+ # 跳过表头行
if i in hdrows:
continue
+
+ # 计算当前行之前的表头行
hr = [r - i for r in hdrows]
hr = [r for r in hr if r < 0]
+
+ # 找到最近的连续表头行
t = len(hr) - 1
while t > 0:
if hr[t] - hr[t - 1] > 1:
hr = hr[t:]
break
t -= 1
+
+ # 为每列构建表头描述
headers = []
for j in range(len(df.iloc[i, :])):
t = []
@@ -102,6 +166,8 @@ class RAGFlowDocxParser:
if t:
t += ": "
headers.append(t)
+
+ # 构建每行的文本表示
cells = []
for j in range(len(df.iloc[i, :])):
if not str(df.iloc[i, j]):
@@ -109,31 +175,53 @@ class RAGFlowDocxParser:
cells.append(headers[j] + str(df.iloc[i, j]))
lines.append(";".join(cells))
+ # 根据列数决定返回格式
if colnm > 3:
return lines
return ["\n".join(lines)]
def __call__(self, fnm, from_page=0, to_page=100000000):
+ """
+ 解析Word文档，提取指定页面范围内的文本和表格
+
+ 参数:
+ fnm: 文件名或二进制内容
+ from_page: 起始页码(从0开始)
+ to_page: 结束页码
+
+ 返回:
+ 元组(secs, tbls)，其中:
+ - secs: 段落内容列表，每项为(文本内容, 样式名称)的元组
+ - tbls: 表格内容列表
+ """
+ # 根据输入类型创建Document对象
self.doc = Document(fnm) if isinstance(
fnm, str) else Document(BytesIO(fnm))
- pn = 0 # parsed page
- secs = [] # parsed contents
+ pn = 0 # 当前解析页码
+ secs = [] # 存储解析的段落内容
+
+ # 遍历文档中的所有段落
for p in self.doc.paragraphs:
+ # 如果超出指定页码范围，停止解析
if pn > to_page:
break
- runs_within_single_paragraph = [] # save runs within the range of pages
+ runs_within_single_paragraph = [] # 保存在页面范围内的文本片段
+ # 遍历段落中的所有文本片段(run)
for run in p.runs:
if pn > to_page:
break
+ # 如果当前页码在指定范围内且段落有内容，则添加文本
if from_page <= pn < to_page and p.text.strip():
- runs_within_single_paragraph.append(run.text) # append run.text first
+ runs_within_single_paragraph.append(run.text) # 先添加文本片段
- # wrap page break checker into a static method
+ # 检查页面分隔符
if 'lastRenderedPageBreak' in run._element.xml:
pn += 1
- secs.append(("".join(runs_within_single_paragraph), p.style.name if hasattr(p.style, 'name') else '')) # then concat run.text as part of the paragraph
+ # 将段落文本和样式添加到结果列表
+ secs.append(("".join(runs_within_single_paragraph), p.style.name if hasattr(p.style, 'name') else '')) # 然后将文本片段连接为段落的一部分
+ # 提取所有表格内容
tbls = [self.__extract_table_content(tb) for tb in self.doc.tables]
return secs, tbls
diff --git a/deepdoc/parser/pdf_parser.py b/deepdoc/parser/pdf_parser.py
index 30a30f5..b8a14ca 100644
--- a/deepdoc/parser/pdf_parser.py
+++ b/deepdoc/parser/pdf_parser.py
@@ -284,6 +284,24 @@ class RAGFlowPdfParser:
b["SP"] = ii
def __ocr(self, pagenum, img, chars, ZM=3):
+ """
+ 对PDF页面图像进行OCR处理，识别文本内容并合并PDF提取的字符
+
+ 该方法完成以下任务：
+ 1. 使用OCR模型检测图像中的文本框
+ 2. 将PDF提取的字符与OCR检测的文本框匹配合并
+ 3. 对没有文本的框进行文本识别
+ 4. 计算文本高度统计信息
+
+ 参数:
+ pagenum: 当前处理的页码
+ img: 页面图像对象
+ chars: 从PDF中提取的字符列表
+ ZM: 缩放因子，用于坐标转换
+
+ 返回:
+ 无直接返回值，但会更新self.boxes列表
+ """
start = timer()
bxs = self.ocr.detect(np.array(img))
logging.info(f"__ocr detecting boxes of a image cost ({timer() - start}s)")
@@ -965,6 +983,26 @@ class RAGFlowPdfParser:
def __images__(self, fnm, zoomin=3, page_from=0,
page_to=299, callback=None):
+ """
+ 处理PDF文件并提取图像和文本内容
+
+ 该方法是PDF解析的核心，负责：
+ 1. 将PDF页面转换为高分辨率图像
+ 2. 提取页面中的字符和布局信息
+ 3. 使用OCR识别文本内容
+ 4. 分析文档语言(英文或中文)
+ 5. 提取文档大纲结构
+
+ 参数:
+ fnm: 文件名或二进制内容
+ zoomin: 图像缩放因子，用于提高OCR精度
+ page_from: 起始页码(从0开始)
+ page_to: 结束页码
+ callback: 进度回调函数，用于报告处理进度
+
+ 返回:
+ 无直接返回值，但会更新类的内部状态
+ """
self.lefted_chars = []
self.mean_height = []
self.mean_width = []
diff --git a/deepdoc/vision/ocr.py b/deepdoc/vision/ocr.py
index 87ba2b6..ddd0834 100644
--- a/deepdoc/vision/ocr.py
+++ b/deepdoc/vision/ocr.py
@@ -123,7 +123,19 @@ def load_model(model_dir, nm):
class TextRecognizer:
+ """
+ 文本识别器类，用于识别检测到的文本区域中的具体文字内容
+
+ 该类使用基于CTC(Connectionist Temporal Classification)的文本识别模型，
+ 能够将图像中的文本区域转换为文字内容
+ """
def __init__(self, model_dir):
+ """
+ 初始化文本识别器
+
+ 参数:
+ model_dir: 模型文件所在目录
+ """
self.rec_image_shape = [int(v) for v in "3, 48, 320".split(",")]
self.rec_batch_num = 16
postprocess_params = {
@@ -136,6 +148,16 @@ class TextRecognizer:
self.input_tensor = self.predictor.get_inputs()[0]
def resize_norm_img(self, img, max_wh_ratio):
+ """
+ 调整图像大小并进行归一化处理，保持宽高比
+
+ 参数:
+ img: 输入图像
+ max_wh_ratio: 最大宽高比
+
+ 返回:
+ 处理后的图像张量
+ """
imgC, imgH, imgW = self.rec_image_shape
assert imgC == img.shape[2]
@@ -394,6 +416,12 @@ class TextRecognizer:
class TextDetector:
+ """
+ 文本检测器类，用于检测图像中的文本区域
+
+ 该类使用基于DB(Differentiable Binarization)的文本检测模型，
+ 能够准确定位图像中的文本区域，并返回文本框的坐标信息
+ """
def __init__(self, model_dir):
pre_process_list = [{
'DetResizeForTest': {
diff --git a/management/server/app.py b/management/server/app.py
index 918e098..206d875 100644
--- a/management/server/app.py
+++ b/management/server/app.py
@@ -1,89 +1,20 @@
from flask import Flask, jsonify, request
from flask_cors import CORS
import database
+from routes import register_routes
app = Flask(__name__)
# 启用CORS，允许前端访问
CORS(app, resources={r"/api/*": {"origins": "*"}}, supports_credentials=True)
-# 添加 v1 前缀
+# 注册所有路由
+register_routes(app)
+
+# 登录路由保留在主文件中
@app.route('/api/v1/auth/login', methods=['POST'])
def login():
# 实现登录逻辑
return {"code": 0, "data": {"token": "your-token"}, "message": "登录成功"}
-
-
-@app.route('/api/v1/users/me', methods=['GET'])
-def get_current_user():
- return jsonify({
- "code": 0,
- "data": {
- "username": "admin",
- "roles": ["admin"]
- },
- "message": "获取用户信息成功"
- })
-
-
-@app.route('/api/v1/users', methods=['GET'])
-def get_users():
- """获取用户的API端点,支持分页和条件查询"""
- try:
- # 获取查询参数
- current_page = int(request.args.get('currentPage', 1))
- page_size = int(request.args.get('size', 10))
- username = request.args.get('username', '')
- email = request.args.get('email', '')
-
- # 调用数据库函数获取分页和筛选后的用户数据
- users, total = database.get_users_with_pagination(current_page, page_size, username, email)
-
- # 返回符合前端期望格式的数据
- return jsonify({
- "code": 0, # 成功状态码
- "data": {
- "list": users,
- "total": total
- },
- "message": "获取用户列表成功"
- })
- except Exception as e:
- # 错误处理
- return jsonify({
- "code": 500,
- "message": f"获取用户列表失败: {str(e)}"
- }), 500
-
-@app.route('/api/v1/users/团队名称：{{ currentTeam.name }}
+负责人：{{ currentTeam.ownerName }}
+