327 lines
12 KiB
Python
327 lines
12 KiB
Python
import copy
|
||
import re
|
||
|
||
def convert_numbered_headings(text):
|
||
"""
|
||
将编号转换为对应的Markdown标题：
|
||
- `1.2` → `## 1.2`
|
||
- `1.2.1` → `### 1.2.1`
|
||
- `1. 3. 1` → `### 1.3.1`（去掉空格）
|
||
"""
|
||
# 替换全角 `．` 为半角 `.`
|
||
text = text.replace('．', '.')
|
||
|
||
# 遍历每一行，处理标题
|
||
lines = text.split('\n')
|
||
|
||
for i in range(len(lines)):
|
||
line = lines[i].strip()
|
||
if not line:
|
||
continue
|
||
|
||
# 只匹配 `1.2` 或 `1.2.1` 形式的编号，允许 `1. 3. 1` 这种格式
|
||
match = re.match(r'^(#*\s*)(\d+(?:\s*\.\s*\d+)+)', line)
|
||
if match:
|
||
prefix, num = match.groups()
|
||
level = num.count('.') # `.` 的个数决定层级
|
||
new_prefix = '#' * (level + 1) # 1.2 → ##, 1.2.1 → ###
|
||
|
||
# 去掉数字和 `.` 之间的空格
|
||
clean_num = re.sub(r'\s*\.\s*', '.', num)
|
||
|
||
# 替换标题
|
||
lines[i] = f"{new_prefix} {clean_num}{line[match.end():]}"
|
||
|
||
return '\n'.join(lines)
|
||
|
||
|
||
|
||
def custom_convert_numbered_headings(text, max_level=9):
|
||
"""
|
||
将编号转换为对应的Markdown标题，支持自定义最大分级层数：
|
||
- `1.2` → `## 1.2` (当max_level>=2)
|
||
- `1.2.1` → `### 1.2.1` (当max_level>=3)
|
||
- `1. 3. 1` → `### 1.3.1`（去掉空格）
|
||
|
||
参数:
|
||
text: 要处理的文本
|
||
max_level: 最大支持的分级层数(默认为3)
|
||
"""
|
||
# 替换全角 `．` 为半角 `.`
|
||
text = text.replace('．', '.')
|
||
|
||
# 遍历每一行，处理标题
|
||
lines = text.split('\n')
|
||
|
||
for i in range(len(lines)):
|
||
line = lines[i].strip()
|
||
if not line:
|
||
continue
|
||
|
||
# 只匹配 `1.2` 或 `1.2.1` 形式的编号，允许 `1. 3. 1` 这种格式
|
||
match = re.match(r'^(#*\s*)(\d+(?:\s*\.\s*\d+)+)', line)
|
||
if match:
|
||
prefix, num = match.groups()
|
||
level = num.count('.') # `.` 的个数决定层级
|
||
|
||
# 只处理不超过max_level的标题
|
||
if level <= max_level:
|
||
new_prefix = '#' * (level + 1) # 1.2 → ##, 1.2.1 → ###
|
||
|
||
# 去掉数字和 `.` 之间的空格
|
||
clean_num = re.sub(r'\s*\.\s*', '.', num)
|
||
|
||
# 替换标题
|
||
lines[i] = f"{new_prefix} {clean_num}{line[match.end():]}"
|
||
else:
|
||
# 超过max_level的标题保持不变
|
||
lines[i] = line
|
||
|
||
return lines #'\n'.join(lines)
|
||
|
||
def parse_markdown(md_text, max_level=6):
|
||
# md_text = convert_numbered_headings(md_text)
|
||
md_text = custom_convert_numbered_headings(md_text, max_level=max_level)
|
||
# lines = md_text.strip().split("\n")
|
||
lines = md_text
|
||
results = []
|
||
current_headers = [None] * max_level # 动态存储各级标题（索引0=H1，1=H2...）
|
||
current_text = []
|
||
|
||
def add_entry():
|
||
nonlocal current_text
|
||
if current_text or any(current_headers):
|
||
# 拼接有效标题到 text 中（跳过 None）
|
||
full_text = []
|
||
for level in range(max_level):
|
||
if current_headers[level] is not None:
|
||
full_text.append(f"{'#' * (level + 1)} {current_headers[level]}")
|
||
full_text.extend(current_text)
|
||
|
||
# 生成结果（动态分配 Header_1, Header_2...）
|
||
entry = {"text": "".join(full_text)}
|
||
for level in range(max_level):
|
||
entry[f"Header_{level + 1}"] = current_headers[level]
|
||
results.append(entry)
|
||
current_text = []
|
||
|
||
# 动态生成正则表达式，匹配 1 到 max_level 个 #
|
||
header_pattern = re.compile(r'^(#{1,%d})\s*(.*)' % max_level)
|
||
|
||
for line in lines:
|
||
line = line
|
||
header_match = header_pattern.match(line)
|
||
|
||
if header_match:
|
||
add_entry()
|
||
level = len(header_match.group(1)) - 1 # 转换为 0-based 索引
|
||
title = header_match.group(2)
|
||
|
||
if level < max_level:
|
||
# 更新当前标题，并清空更低层级的标题
|
||
current_headers[level] = title
|
||
for l in range(level + 1, max_level):
|
||
current_headers[l] = None
|
||
elif line:
|
||
current_text.append(line)
|
||
|
||
add_entry()
|
||
return results
|
||
|
||
# def parse_markdown(md_text,max_level=3):
|
||
# # md_text = convert_numbered_headings(md_text)
|
||
# md_text = custom_convert_numbered_headings(md_text,max_level=max_level)
|
||
# lines = md_text.strip().split("\n")
|
||
# results = []
|
||
# current_h1, current_h2, current_h3 = None, None, None
|
||
# current_text = []
|
||
|
||
# def add_entry():
|
||
# if current_h1 or current_h2 or current_h3:
|
||
# results.append({
|
||
# "Header_1": current_h1,
|
||
# "Header_2": current_h2,
|
||
# "Header_3": current_h3,
|
||
# "text": "\n".join(current_text).strip()
|
||
# })
|
||
# current_text.clear()
|
||
|
||
# for line in lines:
|
||
# line = line.strip()
|
||
# header_match = re.match(r'^(#{1,3})\s*(.*)', line)
|
||
|
||
# if header_match:
|
||
# add_entry()
|
||
# level = len(header_match.group(1))
|
||
# title = header_match.group(2).strip()
|
||
|
||
# if level == 1:
|
||
# current_h1, current_h2, current_h3 = title, None, None
|
||
# elif level == 2:
|
||
# current_h2, current_h3 = title, None
|
||
# elif level == 3:
|
||
# current_h3 = title
|
||
# else:
|
||
# current_text.append(line)
|
||
|
||
# add_entry()
|
||
# return results
|
||
|
||
def header_postprocess(md_header_json):
|
||
processed_results = []
|
||
buffer_entry = None
|
||
|
||
for md in md_header_json:
|
||
headers = {k: md.get(k) for k in md if k.startswith("Header_")}
|
||
text = md.get("text", "")
|
||
|
||
md["origin_text"] = copy.deepcopy(text) # 确保完全独立
|
||
|
||
# 修复后的判断逻辑
|
||
is_header_only = (
|
||
not text.strip() or
|
||
any(text.strip() == headers.get(f"Header_{i}", "") for i in range(1, 5)
|
||
))
|
||
|
||
if is_header_only:
|
||
if buffer_entry is None:
|
||
buffer_entry = md.copy()
|
||
else:
|
||
# 合并标题，保留最高级标题
|
||
for level in range(1, 5):
|
||
key = f"Header_{level}"
|
||
if key in md and md[key] is not None:
|
||
buffer_entry[key] = md[key]
|
||
else:
|
||
if buffer_entry is not None:
|
||
# 合并缓存的标题和当前内容
|
||
new_text = []
|
||
for level in range(1, 5):
|
||
key = f"Header_{level}"
|
||
if buffer_entry.get(key):
|
||
new_text.append(f"{'#' * level} {buffer_entry[key]}")
|
||
new_text.append(text)
|
||
md["text"] = "".join(new_text)
|
||
|
||
for level in range(1, 5):
|
||
key = f"Header_{level}"
|
||
md[key] = buffer_entry.get(key)
|
||
|
||
buffer_entry = None
|
||
processed_results.append(md)
|
||
|
||
if buffer_entry is not None:
|
||
processed_results.append(buffer_entry)
|
||
|
||
return processed_results
|
||
|
||
def postprocess_chunks(chunks):
|
||
"""合并被分割的原型图上下文（适配_process_chunk生成的结构）"""
|
||
new_chunks = []
|
||
i = 0
|
||
|
||
# 修正后的图片匹配正则表达式（确保括号配对）
|
||
image_pattern = re.compile(
|
||
r'(!\[.*?\]\(.*?\))|' # Markdown图片 
|
||
r'(<img.*?src=["\'].*?["\'].*?>)|' # HTML <img>标签
|
||
r'(\[.*?\]\(.*?\))|' # Markdown链接 [text](url)
|
||
r'(https?://\S+\.(?:jpe?g|png|gif|bmp|svg|webp)(?:\?\S*)?)', # 纯图片URL
|
||
re.IGNORECASE
|
||
)
|
||
|
||
# 修正后的标题匹配正则表达式
|
||
title_pattern = re.compile(
|
||
r'(#\s*[0-9一二三四五六七八九十]+[\.．、:：]?\s*[原形图圖示示])|'
|
||
r'(prototype|diagram|screenshot|界面图)',
|
||
re.IGNORECASE
|
||
)
|
||
|
||
while i < len(chunks):
|
||
current_chunk = chunks[i]
|
||
current_text = current_chunk.get("text", "")
|
||
|
||
# 检查是否是原型图标题块
|
||
is_prototype_title = title_pattern.search(current_text)
|
||
|
||
# 情况1：当前是原型图标题但未包含图片
|
||
if is_prototype_title and not image_pattern.search(current_text):
|
||
# 向后查找最多3个块寻找图片
|
||
for j in range(i+1, min(i+4, len(chunks))):
|
||
next_chunk = chunks[j]
|
||
next_text = next_chunk.get("text", "")
|
||
|
||
if image_pattern.search(next_text):
|
||
# 合并元数据
|
||
merged_chunk = {
|
||
"id": current_chunk["id"],
|
||
"Header_1": current_chunk.get("Header_1", "") or next_chunk.get("Header_1", ""),
|
||
"Header_2": current_chunk.get("Header_2", "") or next_chunk.get("Header_2", ""),
|
||
"Header_3": current_chunk.get("Header_3", "") or next_chunk.get("Header_3", ""),
|
||
"text": current_text + next_text,
|
||
"resource": current_chunk["resource"],
|
||
"img_path": ",".join(filter(None, [
|
||
current_chunk.get("img_path", ""),
|
||
next_chunk.get("img_path", "")
|
||
])) or None
|
||
}
|
||
new_chunks.append(merged_chunk)
|
||
i = j + 1 # 跳过已合并的块
|
||
break
|
||
else:
|
||
# 没找到图片则保留原块
|
||
new_chunks.append(current_chunk)
|
||
i += 1
|
||
else:
|
||
# 情况2：普通块处理
|
||
new_chunks.append(current_chunk)
|
||
i += 1
|
||
|
||
return new_chunks
|
||
|
||
# def header_postprocess(md_header_json):
|
||
# processed_results = []
|
||
# buffer_entry = None
|
||
|
||
# for md in md_header_json:
|
||
# header_1 = md.get("Header_1")
|
||
# header_2 = md.get("Header_2")
|
||
# header_3 = md.get("Header_3")
|
||
# text = md.get("text", "")
|
||
|
||
# # 如果当前条目只有标题没有正文，则缓存起来
|
||
# if not text.strip() or text.strip() == header_1 or text.strip() == header_2 or text.strip() == header_3:
|
||
# if buffer_entry is None:
|
||
# buffer_entry = md.copy()
|
||
# else:
|
||
# # 合并标题
|
||
# for level in range(1, 4):
|
||
# key = f"Header_{level}"
|
||
# if key in md and md[key] is not None:
|
||
# buffer_entry[key] = md[key]
|
||
# else:
|
||
# # 如果有正文内容，检查是否有缓存的标题
|
||
# if buffer_entry is not None:
|
||
# # 合并缓存的标题和当前内容
|
||
# new_text = []
|
||
# for level in range(1, 4):
|
||
# key = f"Header_{level}"
|
||
# if buffer_entry.get(key):
|
||
# new_text.append(f"{'#' * level} {buffer_entry[key]}")
|
||
# new_text.append(text)
|
||
# md["text"] = "\n".join(new_text)
|
||
|
||
# # 更新标题信息
|
||
# for level in range(1, 4):
|
||
# key = f"Header_{level}"
|
||
# md[key] = buffer_entry.get(key)
|
||
|
||
# buffer_entry = None
|
||
# processed_results.append(md)
|
||
|
||
# # 处理最后可能剩余的缓存条目
|
||
# if buffer_entry is not None:
|
||
# # 如果最后只剩下标题没有正文，单独作为一个条目
|
||
# processed_results.append(buffer_entry)
|
||
|
||
# return processed_results
|