app/proofread_core.py

import re
from typing import Any, AbstractSet
from urllib.parse import unquote_to_bytes


def utf16_length(value: str) -> int:
    return len(value.encode("utf-16-le")) // 2


def delta_edit_cost(delta: str, original: str) -> int | None:
    if not isinstance(delta, str):
        return None
    consumed = 0
    cost = 0
    for token in delta.split("\t") if delta else []:
        if not token:
            return None
        operation = token[0]
        value = token[1:]
        if operation in {"=", "-"}:
            if not value.isdigit():
                return None
            size = int(value)
            consumed += size
            if operation == "-":
                cost += size
        elif operation == "+":
            try:
                inserted = unquote_to_bytes(value.replace("+", "%2B")).decode("utf-8")
            except (UnicodeDecodeError, ValueError):
                return None
            cost += utf16_length(inserted)
        else:
            return None
    return cost if consumed == utf16_length(original) else None


def diff_cost(delta: str) -> int | None:
    if not isinstance(delta, str):
        return None
    total = 0
    for token in delta.split("\t") if delta else []:
        if not token or token[0] not in {"=", "-", "+"}:
            return None
        if token[0] in {"=", "-"}:
            if not token[1:].isdigit():
                return None
            if token[0] == "-":
                total += int(token[1:])
        else:
            try:
                total += utf16_length(unquote_to_bytes(token[1:].replace("+", "%2B")).decode("utf-8"))
            except (UnicodeDecodeError, ValueError):
                return None
    return total


def _patch_index(raw: Any, minimum: int, maximum: int) -> int:
    if not isinstance(raw, str) or not re.fullmatch(r"0|[1-9][0-9]*", raw):
        raise ValueError("patch index is invalid")
    value = int(raw)
    if value < minimum or value > maximum:
        raise ValueError("patch index is out of range")
    return value


def _content_part(value: Any, part_types: AbstractSet[str]) -> dict[str, str]:
    if not isinstance(value, dict) or set(value) != {"text", "type"}:
        raise ValueError("inserted part is invalid")
    if not isinstance(value["text"], str) or value["type"] not in part_types:
        raise ValueError("inserted part is invalid")
    if len(value["text"]) > 20000:
        raise ValueError("inserted part is too long")
    return {"text": value["text"], "type": value["type"]}


def validate_proofread_patch(
    patch: Any,
    article: dict[str, Any],
    part_types: AbstractSet[str],
) -> dict[str, Any]:
    if not isinstance(patch, dict) or patch.get("version") != 2:
        raise ValueError("invalid OCR patch")
    if set(patch) - {"version", "parts", "comments", "description", "newComments"}:
        raise ValueError("invalid OCR patch fields")
    parts = patch.get("parts")
    comments = patch.get("comments")
    if not isinstance(parts, dict) or not isinstance(comments, dict):
        raise ValueError("patch parts and comments must be objects")
    article_parts = article.get("parts") if isinstance(article.get("parts"), list) else []
    article_comments = article.get("comments") if isinstance(article.get("comments"), list) else []
    description = patch.get("description", "")
    normalized: dict[str, Any] = {"version": 2, "parts": {}, "comments": {}, "description": ""}
    if not isinstance(description, str):
        raise ValueError("patch description must be a string")
    if description:
        original_description = str(article.get("description") or "")
        if delta_edit_cost(description, original_description) is None:
            raise ValueError("description diff is invalid")
        if apply_text_delta(original_description, description) != original_description:
            normalized["description"] = description
    for raw_index, change in parts.items():
        index = _patch_index(raw_index, 0, len(article_parts) - 1)
        if not isinstance(change, dict) or not set(change) <= {"insertBefore", "insertAfter", "delete", "diff", "type"}:
            raise ValueError("part patch is invalid")
        if not change:
            raise ValueError("part patch is empty")
        item: dict[str, Any] = {}
        for key in ("insertBefore", "insertAfter"):
            if key in change:
                if not isinstance(change[key], list):
                    raise ValueError("inserted parts must be a list")
                values = [_content_part(value, part_types) for value in change[key]]
                if values:
                    item[key] = values
        if "delete" in change:
            if not isinstance(change["delete"], bool):
                raise ValueError("part delete must be boolean")
            if change["delete"] and any(key in change for key in ("diff", "type")):
                raise ValueError("deleted part cannot also be changed")
            if change["delete"]:
                item["delete"] = True
        if "diff" in change:
            original = article_part_text(article, index)
            if not isinstance(change["diff"], str) or delta_edit_cost(change["diff"], original) is None:
                raise ValueError("part diff is invalid")
            if apply_text_delta(original, change["diff"]) != original:
                item["diff"] = change["diff"]
        if "type" in change:
            if change["type"] not in part_types:
                raise ValueError("part type is invalid")
            original_type = article_parts[index].get("type") if isinstance(article_parts[index], dict) else None
            if change["type"] != (original_type or "paragraph"):
                item["type"] = change["type"]
        if item:
            normalized["parts"][str(index)] = item
    for raw_index, change in comments.items():
        index = _patch_index(raw_index, 1, len(article_comments))
        if not isinstance(change, dict) or not set(change) <= {"insertBefore", "insertAfter", "delete", "diff"} or not change:
            raise ValueError("comment patch is invalid")
        item = {}
        for key in ("insertBefore", "insertAfter"):
            if key in change:
                if not isinstance(change[key], list):
                    raise ValueError("inserted comments must be a list")
                values = []
                for value in change[key]:
                    if not isinstance(value, dict) or set(value) - {"id", "text"} or not isinstance(value.get("text"), str):
                        raise ValueError("inserted comment is invalid")
                    values.append({key: value[key] for key in value})
                if values:
                    item[key] = values
        if "delete" in change:
            if not isinstance(change["delete"], bool):
                raise ValueError("comment delete must be boolean")
            if change["delete"]:
                item["delete"] = True
        if "diff" in change:
            original = str(article_comments[index - 1] or "")
            if not isinstance(change["diff"], str) or delta_edit_cost(change["diff"], original) is None:
                raise ValueError("comment diff is invalid")
            if apply_text_delta(original, change["diff"]) != original:
                item["diff"] = change["diff"]
        if item:
            normalized["comments"][str(index)] = item
    if "newComments" in patch:
        if not isinstance(patch["newComments"], list) or any(not isinstance(value, str) for value in patch["newComments"]):
            raise ValueError("new comments are invalid")
        if not patch["newComments"] and not normalized["parts"] and not normalized["comments"] and not normalized["description"]:
            raise ValueError("patch contains no changes")
        if patch["newComments"]:
            normalized["newComments"] = list(patch["newComments"])
    if not normalized["parts"] and not normalized["comments"] and not normalized["description"] and "newComments" not in normalized:
        raise ValueError("patch contains no changes")
    return normalized


def article_part_text(article: dict[str, Any], index: int) -> str:
    parts = article.get("parts") if isinstance(article.get("parts"), list) else []
    if index < 0 or index >= len(parts) or not isinstance(parts[index], dict):
        raise ValueError("part index is invalid")
    values = list(str(parts[index].get("text") or ""))
    pivots = article.get("comment_pivots") if isinstance(article.get("comment_pivots"), list) else []
    selected = [item for item in pivots if isinstance(item, dict) and item.get("part_idx") == index]
    for pivot in sorted(selected, key=lambda item: int(item.get("offset", 0)), reverse=True):
        offset = int(pivot.get("offset", 0))
        if offset < 0 or offset > len(values):
            raise ValueError("comment pivot offset is invalid")
        values.insert(offset, f"〔{pivot.get('index')}〕")
    return "".join(values)


def apply_text_delta(text: str, delta: str) -> str:
    source = text.encode("utf-16-le")
    cursor = 0
    output = bytearray()
    for token in delta.split("\t") if delta else []:
        if not token:
            continue
        operation = token[0]
        value = token[1:]
        if operation == "=":
            if value.isdigit():
                size = int(value) * 2
                output.extend(source[cursor:cursor + size])
                cursor += size
        elif operation == "-":
            if value.isdigit():
                cursor += int(value) * 2
        elif operation == "+":
            try:
                inserted = unquote_to_bytes(value.replace("+", "%2B")).decode("utf-8")
                output.extend(inserted.encode("utf-16-le"))
            except (UnicodeDecodeError, ValueError):
                continue
    if cursor != len(source):
        raise ValueError("delta length mismatch")
    return output.decode("utf-16-le")


def _clip_text(value: Any, limit: int | None = None) -> str:
    text = str(value if value is not None else "")
    if limit is None or len(text) <= limit:
        return text
    return text[:limit] + "…"


def _preview_value(value: Any, limit: int | None = None) -> Any:
    return value if limit is None else _clip_text(value, limit)


def proofread_change_preview(
    patch: dict[str, Any] | None,
    article: dict[str, Any],
    metadata: dict[str, Any] | None,
    source: dict[str, Any],
    clip: int | None = None,
) -> list[dict[str, Any]]:
    article = article or {}
    changes: list[dict[str, Any]] = []
    if patch:
        article_parts = article.get("parts") if isinstance(article.get("parts"), list) else []
        article_comments = article.get("comments") if isinstance(article.get("comments"), list) else []
        for raw_index in sorted((patch.get("parts") or {}), key=int):
            index = int(raw_index)
            change = patch["parts"][raw_index]
            original_full = article_part_text(article, index)
            original = _clip_text(original_full, clip)
            if "diff" in change:
                edited_full = apply_text_delta(original_full, change["diff"])
                if edited_full != original_full:
                    changes.append({"kind": "part", "index": index + 1, "original": original, "edited": _clip_text(edited_full, clip)})
            if change.get("delete"):
                changes.append({"kind": "part", "index": index + 1, "delete": True, "original": original})
            for key in ("insertBefore", "insertAfter"):
                for part in change.get(key) or []:
                    changes.append({
                        "kind": "part", "index": index + 1, "insert": key == "insertAfter",
                        "text": _clip_text(part.get("text"), clip), "part_type": part.get("type"),
                    })
            if "type" in change:
                old_type = article_parts[index].get("type") if 0 <= index < len(article_parts) and isinstance(article_parts[index], dict) else None
                if old_type != change["type"]:
                    changes.append({"kind": "part_type", "index": index + 1, "old": old_type, "new": change["type"]})
        for raw_index in sorted((patch.get("comments") or {}), key=int):
            index = int(raw_index) - 1
            change = patch["comments"][raw_index]
            base = str(article_comments[index] or "") if 0 <= index < len(article_comments) else ""
            original = _clip_text(base, clip)
            if "diff" in change:
                edited = apply_text_delta(base, change["diff"])
                if edited != base:
                    changes.append({"kind": "comment", "index": index + 1, "original": original, "edited": _clip_text(edited, clip)})
            if change.get("delete"):
                changes.append({"kind": "comment", "index": index + 1, "delete": True, "original": original})
            for key in ("insertBefore", "insertAfter"):
                for comment in change.get(key) or []:
                    changes.append({
                        "kind": "comment", "index": index + 1, "insert": key == "insertAfter",
                        "text": _clip_text(comment.get("text"), clip), "part_type": "comment",
                    })
        for offset, text in enumerate(patch.get("newComments") or [], start=1):
            changes.append({"kind": "new_comment", "index": len(article_comments) + offset, "text": _clip_text(text, clip)})
        if patch.get("description"):
            original = str(article.get("description") or "")
            edited = apply_text_delta(original, patch["description"])
            if edited != original:
                changes.append({
                    "kind": "description",
                    "original": _clip_text(original, clip),
                    "edited": _clip_text(edited, clip),
                })
    if metadata:
        article_meta = metadata.get("article") or {}
        for field in ("title", "authors", "dates", "tags"):
            if field in article_meta and article.get(field) != article_meta[field]:
                changes.append({"kind": "metadata", "field": field, "old": _preview_value(article.get(field), clip), "new": _preview_value(article_meta[field], clip)})
        source_meta = metadata.get("source") or {}
        source_fields = {"name": "publication_name", "author": "publication_author", "type": "publication_type", "files": "source_files"}
        for field, key in source_fields.items():
            if field in source_meta and source.get(key) != source_meta[field]:
                changes.append({"kind": "metadata", "field": field, "old": _preview_value(source.get(key), clip), "new": _preview_value(source_meta[field], clip)})
    return changes


def patch_is_safe_to_merge(
    patch: dict[str, Any] | None,
    _article: dict[str, Any],
    max_diff_cost: int,
    max_net_paragraph_delta: int,
) -> bool:
    if patch is None:
        return True
    if patch.get("version") != 2 or patch.get("newComments") or patch.get("description"):
        return False
    parts = patch.get("parts") or {}
    comments = patch.get("comments") or {}
    if not isinstance(parts, dict) or not isinstance(comments, dict):
        return False
    if not parts and not comments:
        return False
    delta = 0
    cost = 0
    for change in parts.values():
        if not isinstance(change, dict) or not change:
            return False
        if set(change) - {"diff", "type", "insertBefore", "insertAfter", "delete"}:
            return False
        if "delete" in change and change["delete"] is not True:
            return False
        if change.get("delete"):
            delta -= 1
        for key in ("insertBefore", "insertAfter"):
            if key in change:
                if not isinstance(change[key], list):
                    return False
                delta += len(change[key])
                for part in change[key]:
                    if not isinstance(part, dict) or not isinstance(part.get("text"), str):
                        return False
                    cost += utf16_length(part["text"])
        if "diff" in change:
            if not isinstance(change["diff"], str):
                return False
            item_cost = diff_cost(change["diff"])
            if item_cost is None:
                return False
            cost += item_cost
    for change in comments.values():
        if not isinstance(change, dict) or set(change) != {"diff"} or not isinstance(change.get("diff"), str):
            return False
        item_cost = diff_cost(change["diff"])
        if item_cost is None:
            return False
        cost += item_cost
    return -max_net_paragraph_delta <= delta <= max_net_paragraph_delta and cost <= max_diff_cost