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图片 ![alt](url) r'()|' # HTML 标签 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