from datetime import datetime
from html import escape
def _clean_external_subject(subject: str) -> str:
value = (subject or "(No subject)").strip()
for prefix in ("Re: ", "RE: ", "re: "):
if value.startswith(prefix):
value = value[len(prefix) :].strip()
for prefix in ("Human Handoff · ", "Human Handoff: "):
if value.lower().startswith(prefix.lower()):
value = value[len(prefix) :].strip()
return value or "(No subject)"
def handoff_email_subject(external_subject: str) -> str:
return f"Human Handoff · {_clean_external_subject(external_subject)}"
def external_reply_subject(external_subject: str) -> str:
cleaned = _clean_external_subject(external_subject)
return f"Re: {cleaned}"
def format_sender_line(sender: dict) -> str:
name = (sender.get("name") or "").strip()
address = (sender.get("address") or "").strip()
if not address:
return name or "unknown sender"
local = address.split("@", 1)[0].lower()
if name and name.lower() not in {local, address.lower()}:
return f"{name} <{address}>"
return address
def sender_display_name(sender: dict | None) -> str:
sender = sender or {}
name = (sender.get("name") or "").strip()
address = (sender.get("address") or "").strip()
if name:
return name
if address:
return address.split("@", 1)[0].replace(".", " ").replace("_", " ").replace("-", " ").title()
return "unknown sender"
def _attribution_date(value) -> str:
if not isinstance(value, datetime):
return ""
return value.strftime("%a, %b %d, %Y at %I:%M %p").replace(" 0", " ")
def build_gmail_quote(
*,
sender: dict | None,
event_at,
body_text: str | None,
body_html: str | None,
) -> tuple[str, str]:
identity = format_sender_line(sender or {})
date = _attribution_date(event_at)
attribution = f"On {date} {identity} wrote:" if date else f"{identity} wrote:"
text_source = (body_text or "").rstrip()
quoted_lines = [f"> {line}" if line else ">" for line in text_source.splitlines()]
quoted_text = f"{attribution}\n" + "\n".join(quoted_lines)
inner_html = (body_html or "").strip() or (
f"{escape(text_source).replace(chr(10), '
')}
"
)
quoted_html = (
f''
f'
{escape(attribution)}
'
f'
'
f"{inner_html}"
f"
{escape(new_text).replace(chr(10), '
')}
"
)
body_html = f"{visible_html}
{quoted_html}"
return body_text, body_html
def _footer_html(agent_email: str) -> str:
return f"""
"""
def _footer_text(agent_email: str) -> str:
return (
"↩ Reply normally\n\n"
f"Your reply will be sent to the original sender through {agent_email}.\n"
"Only the new text you write will be sent."
)
def _header_html(sender_line: str, external_subject: str) -> str:
subject = escape(_clean_external_subject(external_subject))
return f"""
✦ Human Handoff
{escape(sender_line)}
| {subject}
"""
def _header_text(sender_line: str, external_subject: str) -> str:
return f"✦ Human Handoff\n{sender_line} | {_clean_external_subject(external_subject)}"
def build_mirror_handoff_html(
*,
sender_line: str,
external_subject: str,
original_html: str,
agent_email: str,
) -> str:
return f"""
{_header_html(sender_line, external_subject)}
{original_html}
{_footer_html(agent_email)}
"""
def build_mirror_handoff_text(
*,
sender_line: str,
external_subject: str,
original_body_text: str,
agent_email: str,
) -> str:
return "\n\n".join(
[
_header_text(sender_line, external_subject),
"─" * 56,
original_body_text.strip(),
"─" * 56,
_footer_text(agent_email),
]
)
def build_mirror_update_html(
*,
sender_line: str,
external_subject: str,
original_html: str,
agent_email: str,
) -> str:
subject = escape(_clean_external_subject(external_subject))
return f"""
✦ Customer replied
{escape(sender_line)}
| {subject}
{original_html}
{_footer_html(agent_email)}
"""
def build_mirror_update_text(
*,
sender_line: str,
external_subject: str,
original_body_text: str,
agent_email: str,
) -> str:
return "\n\n".join(
[
f"✦ Customer replied\n{sender_line} | {_clean_external_subject(external_subject)}",
"─" * 56,
original_body_text.strip(),
"─" * 56,
_footer_text(agent_email),
]
)
def build_relay_confirmation_text(customer_body: str) -> str:
return f"Sent to customer:\n\n{customer_body.strip()}"