119 lines
3.7 KiB
Python
119 lines
3.7 KiB
Python
import re
|
|
from html import unescape
|
|
|
|
# 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"\nOn .+? wrote:\s*\n",
|
|
re.IGNORECASE,
|
|
),
|
|
re.compile(
|
|
r"\nOn .+?, .+? at .+? .+? wrote:\s*\n",
|
|
re.IGNORECASE,
|
|
),
|
|
re.compile(
|
|
r"\nFrom:\s*.+\nSent:\s*.+\nTo:\s*.+\nSubject:\s*.+",
|
|
re.IGNORECASE,
|
|
),
|
|
re.compile(r"\n_{5,}\n"),
|
|
]
|
|
|
|
HTML_QUOTE_PATTERNS = [
|
|
re.compile(r"<blockquote\b[^>]*>.*?</blockquote>", re.IGNORECASE | re.DOTALL),
|
|
re.compile(r'<div[^>]+class="[^"]*gmail_quote[^"]*"[^>]*>.*?</div>', re.IGNORECASE | re.DOTALL),
|
|
re.compile(r'<div[^>]+class="[^"]*gmail_attr[^"]*"[^>]*>.*?</div>', re.IGNORECASE | re.DOTALL),
|
|
]
|
|
|
|
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")
|
|
|
|
|
|
def _html_to_text(html: str) -> str:
|
|
cleaned = html
|
|
for pattern in HTML_QUOTE_PATTERNS:
|
|
cleaned = pattern.sub("", cleaned)
|
|
cleaned = re.sub(r"<br\s*/?>", "\n", cleaned, flags=re.IGNORECASE)
|
|
cleaned = re.sub(r"</p\s*>", "\n", cleaned, flags=re.IGNORECASE)
|
|
cleaned = re.sub(r"<[^>]+>", "", cleaned)
|
|
return unescape(cleaned)
|
|
|
|
|
|
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_stripped = body_html
|
|
for pattern in HTML_QUOTE_PATTERNS:
|
|
html_stripped = pattern.sub("", html_stripped)
|
|
html_text = _html_to_text(html_stripped).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 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
|