217 lines
7.0 KiB
Python
217 lines
7.0 KiB
Python
"""Markdown 解析与 Note Block 切分。
|
||
|
||
Block 由 Markdown 文本生成：标题行独立成块（heading_path 含自身），正文按空行分段，
|
||
每块记录其在原文中的 start_offset / end_offset，用于 Citation 跳转定位。block_id 由
|
||
(note_id, heading_path, content) 稳定派生，内容不变则 ID 稳定。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from app.contracts import NoteBlock
|
||
from app.textutils import count_tokens
|
||
|
||
_HEADING_RE = re.compile(r"^(#{1,6})[ \t]+(.*?)\s*$")
|
||
_FRONTMATTER_KEY_RE = re.compile(r"^([A-Za-z0-9_-]+)\s*:\s*(.*)$")
|
||
_FENCE_RE = re.compile(r"^[ \t]{0,3}(`{3,}|~{3,})(?:[^`]*)$")
|
||
|
||
|
||
@dataclass
|
||
class ParsedNote:
|
||
note_id: str
|
||
title: str
|
||
file_path: str
|
||
folder: str
|
||
tags: list[str]
|
||
created_at: datetime
|
||
updated_at: datetime
|
||
blocks: list[NoteBlock] = field(default_factory=list)
|
||
embedding_local_only: bool = False
|
||
|
||
|
||
def note_id_for_path(rel_path: str) -> str:
|
||
"""由相对路径派生稳定 note_id（路径哈希而非路径本身，见团队约定「不用路径当 ID」）。
|
||
|
||
MVP 阶段 ID 随文件移动而变化；后续 move 流程会保留原 ID。"""
|
||
normalized = rel_path.replace("\\", "/").strip("/")
|
||
return "note_" + hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:16]
|
||
|
||
|
||
def parse_note(
|
||
*,
|
||
markdown: str,
|
||
file_path: str,
|
||
folder: str,
|
||
tags: list[str] | None = None,
|
||
created_at: datetime,
|
||
updated_at: datetime,
|
||
note_id: str | None = None,
|
||
) -> ParsedNote:
|
||
"""解析一篇 Markdown，生成 ParsedNote（元数据 + Block 列表）。"""
|
||
note_id = note_id or note_id_for_path(file_path)
|
||
frontmatter = _extract_frontmatter(markdown)
|
||
|
||
fallback_title = Path(file_path).stem
|
||
title = frontmatter.get("title") or _first_heading(markdown) or fallback_title
|
||
resolved_tags = list(tags) if tags is not None else _parse_tags(frontmatter.get("tags"))
|
||
|
||
blocks = parse_blocks(markdown, note_id)
|
||
return ParsedNote(
|
||
note_id=note_id,
|
||
title=title,
|
||
file_path=file_path,
|
||
folder=folder,
|
||
tags=resolved_tags,
|
||
created_at=created_at,
|
||
updated_at=updated_at,
|
||
blocks=blocks,
|
||
embedding_local_only=str(frontmatter.get("embedding_local_only", "")).lower() == "true",
|
||
)
|
||
|
||
|
||
def parse_blocks(markdown: str, note_id: str) -> list[NoteBlock]:
|
||
"""把 Markdown 切成 Block，offset 相对原文（含 frontmatter）。"""
|
||
lines = _split_lines(markdown)
|
||
content_start = _content_start(markdown)
|
||
|
||
blocks: list[NoteBlock] = []
|
||
heading_stack: list[str] = []
|
||
body: list[tuple[str, int]] = []
|
||
id_counters: dict[str, int] = {}
|
||
fence_marker: str | None = None
|
||
|
||
def make_block(path: list[str], chunk: list[tuple[str, int]]) -> None:
|
||
if not chunk:
|
||
return
|
||
content = "\n".join(line for line, _ in chunk)
|
||
start = chunk[0][1]
|
||
end = chunk[-1][1] + _utf16_len(chunk[-1][0])
|
||
block_id = _stable_block_id(note_id, path, content, id_counters)
|
||
blocks.append(
|
||
NoteBlock(
|
||
block_id=block_id,
|
||
note_id=note_id,
|
||
heading_path=list(path),
|
||
start_offset=start,
|
||
end_offset=end,
|
||
content=content,
|
||
content_hash=hashlib.sha256(content.encode("utf-8")).hexdigest()[:16],
|
||
token_count=count_tokens(content),
|
||
)
|
||
)
|
||
|
||
def flush_body() -> None:
|
||
nonlocal body
|
||
make_block(heading_stack, body)
|
||
body = []
|
||
|
||
for line, offset in lines:
|
||
if offset < content_start:
|
||
continue # 跳过 frontmatter 区域，但保留 offset 准确性
|
||
|
||
fence = _FENCE_RE.match(line)
|
||
if fence_marker is not None:
|
||
body.append((line, offset))
|
||
marker = fence.group(1) if fence else ""
|
||
if marker.startswith(fence_marker[0]) and len(marker) >= len(fence_marker):
|
||
fence_marker = None
|
||
flush_body()
|
||
continue
|
||
if fence:
|
||
flush_body()
|
||
fence_marker = fence.group(1)
|
||
body.append((line, offset))
|
||
continue
|
||
|
||
heading = _HEADING_RE.match(line)
|
||
if heading:
|
||
flush_body()
|
||
level = len(heading.group(1))
|
||
title = heading.group(2).strip()
|
||
heading_stack = heading_stack[: level - 1] + [title]
|
||
# 标题自身作为一个 Block，便于按章节定位
|
||
make_block(heading_stack, [(line, offset)])
|
||
elif line.strip() == "":
|
||
flush_body() # 空行分隔段落
|
||
else:
|
||
body.append((line, offset))
|
||
|
||
flush_body()
|
||
return blocks
|
||
|
||
|
||
def _stable_block_id(note_id: str, path: list[str], content: str, counters: dict[str, int]) -> str:
|
||
base = hashlib.sha256(
|
||
f"{note_id}\x1f{chr(31).join(path)}\x1f{content}".encode("utf-8")
|
||
).hexdigest()[:16]
|
||
block_id = f"blk_{base}"
|
||
# 同一篇笔记内极少出现的重复段落用后缀消歧，保证唯一
|
||
n = counters.get(block_id, 0)
|
||
counters[block_id] = n + 1
|
||
return block_id if n == 0 else f"{block_id}_{n}"
|
||
|
||
|
||
def _split_lines(text: str) -> list[tuple[str, int]]:
|
||
"""按行拆分并记录 UTF-16 code unit 偏移，直接兼容浏览器编辑器。"""
|
||
result: list[tuple[str, int]] = []
|
||
start = 0
|
||
for raw in text.splitlines(keepends=True):
|
||
line = raw
|
||
if line.endswith("\r\n"):
|
||
line = line[:-2]
|
||
elif line.endswith("\n") or line.endswith("\r"):
|
||
line = line[:-1]
|
||
result.append((line, start))
|
||
start += _utf16_len(raw)
|
||
return result
|
||
|
||
|
||
def _content_start(markdown: str) -> int:
|
||
"""返回正文起始 UTF-16 偏移：有 frontmatter 时跳过 --- 分隔块。"""
|
||
if markdown.startswith("---"):
|
||
end = markdown.find("\n---", 3)
|
||
if end != -1:
|
||
return _utf16_len(markdown[: end + 4])
|
||
return 0
|
||
|
||
|
||
def _utf16_len(text: str) -> int:
|
||
return len(text.encode("utf-16-le")) // 2
|
||
|
||
|
||
def _extract_frontmatter(markdown: str) -> dict[str, str]:
|
||
"""极简 frontmatter 解析，只提取 key: value 行。"""
|
||
if not markdown.startswith("---"):
|
||
return {}
|
||
end = markdown.find("\n---", 3)
|
||
if end == -1:
|
||
return {}
|
||
meta: dict[str, str] = {}
|
||
for line in markdown[3:end].splitlines():
|
||
m = _FRONTMATTER_KEY_RE.match(line)
|
||
if m:
|
||
meta[m.group(1).lower()] = m.group(2).strip()
|
||
return meta
|
||
|
||
|
||
def _first_heading(markdown: str) -> str | None:
|
||
for line in markdown.splitlines():
|
||
m = re.match(r"^#\s+(.*?)\s*$", line)
|
||
if m and m.group(1).strip():
|
||
return m.group(1).strip()
|
||
return None
|
||
|
||
|
||
def _parse_tags(raw: str | None) -> list[str]:
|
||
if not raw:
|
||
return []
|
||
raw = raw.strip()
|
||
if raw.startswith("[") and raw.endswith("]"):
|
||
raw = raw[1:-1]
|
||
return [t.strip().strip("'\"") for t in raw.split(",") if t.strip()]
|