agentic.email/backend/app/services/reply_extractor.py

import re
from html import unescape
from html.parser import HTMLParser

# Patterns that mark the start of quoted previous-message content.
TEXT_QUOTE_MARKERS = [
    re.compile(r"\n-{2,}\s*Original Message\s*-{2,}", re.IGNORECASE),
    re.compile(r"(?:^|\n)\s*On .+? wrote:\s*", re.IGNORECASE),
    re.compile(
        r"\nFrom:\s*.+\nSent:\s*.+\nTo:\s*.+\nSubject:\s*.+",
        re.IGNORECASE,
    ),
    re.compile(r"\n_{5,}\n"),
    re.compile(r"\n\s*> ?"),
]

PRIVATE_LEAK_MARKERS = [
    "human handoff",
    "a conversation needs your attention",
    "why this was handed off",
    "why you're receiving this",
    "gentic human handoff",
    "no handoff instructions were set",
    "step passes",
    "classifier returned",
    "rule evaluation succeeded",
    "reply normally to respond through",
    "latest message",
    "private agent ↔ human conversation",
]


def _normalize_newlines(text: str) -> str:
    return text.replace("\r\n", "\n").replace("\r", "\n")


class _ReplyHTMLTextParser(HTMLParser):
    BLOCK_TAGS = {"div", "p", "pre", "li", "ul", "ol", "table", "tr"}
    IGNORED_TAGS = {"script", "style", "iframe", "object"}

    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.parts: list[str] = []
        self.quote_depth = 0
        self.ignored_depth = 0
        self.at_line_start = True

    def _newline(self) -> None:
        if self.parts and not self.parts[-1].endswith("\n"):
            self.parts.append("\n")
        self.at_line_start = True

    def _text(self, value: str) -> None:
        if self.ignored_depth:
            return
        normalized = re.sub(r"\s+", " ", value)
        if not normalized.strip():
            return
        if self.at_line_start:
            normalized = normalized.lstrip()
            if self.quote_depth:
                self.parts.append("> " * self.quote_depth)
        self.parts.append(normalized)
        self.at_line_start = False

    def handle_starttag(self, tag: str, attrs) -> None:
        tag = tag.lower()
        if tag in self.IGNORED_TAGS:
            self.ignored_depth += 1
            return
        if self.ignored_depth:
            return
        if tag == "br":
            self._newline()
        elif tag == "blockquote":
            self._newline()
            self.quote_depth += 1
        elif tag in self.BLOCK_TAGS:
            self._newline()

    def handle_endtag(self, tag: str) -> None:
        tag = tag.lower()
        if tag in self.IGNORED_TAGS:
            self.ignored_depth = max(0, self.ignored_depth - 1)
            return
        if self.ignored_depth:
            return
        if tag == "blockquote":
            self._newline()
            self.quote_depth = max(0, self.quote_depth - 1)
        elif tag in self.BLOCK_TAGS:
            self._newline()

    def handle_data(self, data: str) -> None:
        self._text(data)

    def text(self) -> str:
        return "".join(self.parts).strip()


def _html_to_text(html: str) -> str:
    parser = _ReplyHTMLTextParser()
    parser.feed(html)
    parser.close()
    return unescape(parser.text())


def _strip_trailing_quoted_lines(text: str) -> str:
    lines = text.split("\n")
    kept: list[str] = []
    for line in lines:
        if re.match(r"^\s*>", line):
            break
        kept.append(line)
    return "\n".join(kept).strip()


def _split_at_quote_markers(text: str) -> tuple[str, str]:
    normalized = _normalize_newlines(text)
    earliest_index = None
    for pattern in TEXT_QUOTE_MARKERS:
        match = pattern.search(normalized)
        if match and (earliest_index is None or match.start() < earliest_index):
            earliest_index = match.start()
    if earliest_index is not None:
        new_content = normalized[:earliest_index].strip()
        quoted = normalized[earliest_index:].strip()
        return new_content, quoted
    return _strip_trailing_quoted_lines(normalized), ""


def extract_new_reply_content(
    *,
    body_text: str | None,
    body_html: str | None = None,
) -> dict[str, str]:
    raw_text = _normalize_newlines((body_text or "").strip())
    if body_html and not raw_text:
        raw_text = _html_to_text(body_html).strip()

    if body_html:
        html_text = _html_to_text(body_html).strip()
        if html_text:
            new_from_html, quoted_from_html = _split_at_quote_markers(html_text)
            if new_from_html:
                return {
                    "raw_body": raw_text or html_text,
                    "new_reply_content": new_from_html,
                    "quoted_content": quoted_from_html,
                }

    new_content, quoted_content = _split_at_quote_markers(raw_text)
    return {
        "raw_body": raw_text,
        "new_reply_content": new_content,
        "quoted_content": quoted_content,
    }


def reply_fields_for_message(
    message: dict,
    *,
    body_text: str | None,
    body_html: str | None = None,
) -> dict[str, str | bool]:
    built = build_reply_display(body_text=body_text, body_html=body_html)
    new_content = built["new_reply_content"] or message.get("new_reply_content", "")
    quoted_content = built["quoted_content"] or message.get("quoted_content", "")
    return {
        "new_reply_content": new_content,
        "quoted_content": quoted_content,
        "has_quoted_text": bool(quoted_content),
    }


def build_reply_display(
    *,
    body_text: str | None,
    body_html: str | None = None,
    limit: int = 240,
) -> dict[str, str | bool]:
    extracted = extract_new_reply_content(body_text=body_text, body_html=body_html)
    new_content = extracted["new_reply_content"]
    quoted = extracted["quoted_content"]
    fallback = re.sub(r"\s+", " ", (body_text or "")).strip()
    snippet_source = new_content or fallback
    return {
        "new_reply_content": new_content,
        "quoted_content": quoted,
        "has_quoted_text": bool(quoted),
        "raw_body": extracted["raw_body"],
        "snippet": re.sub(r"\s+", " ", snippet_source).strip()[:limit],
    }


def validate_external_reply_body(body: str) -> bool:
    normalized = (body or "").strip()
    if not normalized:
        return False
    lower = normalized.lower()
    hits = sum(1 for marker in PRIVATE_LEAK_MARKERS if marker in lower)
    return hits == 0