109 lines
3.9 KiB
Python
109 lines
3.9 KiB
Python
"""Shared, deterministic comparison summary for whole-document rewrites."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from collections import Counter
|
|
from difflib import SequenceMatcher
|
|
from typing import Any
|
|
|
|
_BLOCK_SPLIT_RE = re.compile(r"\n\s*\n+")
|
|
_HEADING_RE = re.compile(r"^(#{1,6})\s+(.+?)\s*$")
|
|
|
|
|
|
def visible_char_count(content: str) -> int:
|
|
return len("".join(content.split()))
|
|
|
|
|
|
def _blocks(content: str) -> list[str]:
|
|
return [block.strip() for block in _BLOCK_SPLIT_RE.split(content) if block.strip()]
|
|
|
|
|
|
def changed_block_count(original: str, rewritten: str) -> int:
|
|
"""Count affected Markdown blocks, not merely the net block-count delta."""
|
|
before = _blocks(original)
|
|
after = _blocks(rewritten)
|
|
changed = 0
|
|
for tag, before_start, before_end, after_start, after_end in SequenceMatcher(
|
|
a=before,
|
|
b=after,
|
|
autojunk=False,
|
|
).get_opcodes():
|
|
if tag != "equal":
|
|
changed += max(before_end - before_start, after_end - after_start)
|
|
return changed
|
|
|
|
|
|
def _headings(content: str) -> list[tuple[int, str]]:
|
|
output: list[tuple[int, str]] = []
|
|
for line in content.splitlines():
|
|
match = _HEADING_RE.match(line.strip())
|
|
if match:
|
|
output.append((len(match.group(1)), match.group(2).strip()))
|
|
return output
|
|
|
|
|
|
def _requested_optimizations(instruction: str) -> list[str]:
|
|
keywords = (
|
|
("重复", "减少重复表述"),
|
|
("衔接", "改善段落衔接"),
|
|
("结构", "调整文章结构"),
|
|
("语气", "统一写作语气"),
|
|
("精炼", "压缩冗余内容"),
|
|
("润色", "改善表达流畅度"),
|
|
("扩写", "补充原文已有信息的说明"),
|
|
)
|
|
return [label for token, label in keywords if token in instruction] or ["按用户要求整体改写"]
|
|
|
|
|
|
def build_document_rewrite_summary(
|
|
*,
|
|
source_display_name: str,
|
|
instruction: str,
|
|
model_name: str,
|
|
original: str,
|
|
rewritten: str,
|
|
validation_warnings: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
before_headings = _headings(original)
|
|
after_headings = _headings(rewritten)
|
|
before_titles = [title for _, title in before_headings]
|
|
after_titles = [title for _, title in after_headings]
|
|
before_counter = Counter(before_titles)
|
|
after_counter = Counter(after_titles)
|
|
added = list((after_counter - before_counter).elements())
|
|
removed = list((before_counter - after_counter).elements())
|
|
common_before = [title for title in before_titles if title in after_counter]
|
|
common_after = [title for title in after_titles if title in before_counter]
|
|
levels_before = {title: level for level, title in before_headings}
|
|
levels_after = {title: level for level, title in after_headings}
|
|
level_warnings = [
|
|
f"标题「{title}」层级从 H{levels_before[title]} 调整为 H{levels_after[title]}"
|
|
for title in levels_before.keys() & levels_after.keys()
|
|
if levels_before[title] != levels_after[title]
|
|
]
|
|
before_chars = visible_char_count(original)
|
|
after_chars = visible_char_count(rewritten)
|
|
return {
|
|
"sourceDisplayName": source_display_name,
|
|
"instruction": instruction,
|
|
"modelName": model_name,
|
|
"originalCharCount": before_chars,
|
|
"rewrittenCharCount": after_chars,
|
|
"charChangeRatio": round((after_chars - before_chars) / before_chars, 4) if before_chars else 0,
|
|
"changedBlockCount": changed_block_count(original, rewritten),
|
|
"headingCountBefore": len(before_headings),
|
|
"headingCountAfter": len(after_headings),
|
|
"headingChanges": {
|
|
"added": added,
|
|
"removed": removed,
|
|
"reordered": common_before != common_after,
|
|
"levelWarnings": level_warnings,
|
|
},
|
|
"requestedOptimizations": _requested_optimizations(instruction),
|
|
"validationWarnings": validation_warnings or [],
|
|
}
|
|
|
|
|
|
__all__ = ["build_document_rewrite_summary", "changed_block_count", "visible_char_count"]
|