153 lines
5 KiB
Python
153 lines
5 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"(?:^|\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*> ?"),
|
|
]
|
|
|
|
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 reply_fields_for_message(
|
|
message: dict,
|
|
*,
|
|
body_text: str | None,
|
|
body_html: str | None = None,
|
|
) -> dict[str, str | bool]:
|
|
if "new_reply_content" in message:
|
|
return {
|
|
"new_reply_content": message.get("new_reply_content", ""),
|
|
"quoted_content": message.get("quoted_content", ""),
|
|
"has_quoted_text": bool(message.get("has_quoted_text") or message.get("quoted_content")),
|
|
}
|
|
built = build_reply_display(body_text=body_text, body_html=body_html)
|
|
return {
|
|
"new_reply_content": built["new_reply_content"],
|
|
"quoted_content": built["quoted_content"],
|
|
"has_quoted_text": built["has_quoted_text"],
|
|
}
|
|
|
|
|
|
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
|