"""
# ---- Top narratives (with bull/bear split) ----
def narr_rows():
out = ""
pn = {p["name"]: p for p in per_narr}
for i, n in enumerate(narratives, 1):
name = n.get("name", "")
avg = n.get("avg_sentiment", 0)
tag = "BULLISH" if avg >= 0.15 else "BEARISH" if avg <= -0.15 else "NEUTRAL"
tcol = _color(avg)
p = pn.get(name, {})
split = f"{p.get('bullish_pct',0):.0f}% / {p.get('bearish_pct',0):.0f}%" if p else "—"
out += f"""
{i}
{_esc(name)}
{split}
{n.get('score',0):.2f}
{tag}
"""
return out
# ---- Top headlines (the meat — what actually happened) ----
def headline_rows():
out = ""
# prefer the most recent, mix of bull/bear, skip pure-neutral filler
picked = [n for n in news if n.get("sentiment") in ("bullish", "bearish")][:7]
if len(picked) < 7:
picked += [n for n in news if n not in picked][: 7 - len(picked)]
for n in picked[:7]:
sent = n.get("sentiment", "neutral")
scol = "#2ECC71" if sent == "bullish" else "#E05252" if sent == "bearish" else "#999"
stag = "BULL" if sent == "bullish" else "BEAR" if sent == "bearish" else "NEUT"
src = (n.get("source", "") or "").replace("GoogleNews: ", "").replace("Reddit/", "")
title = _esc(n.get("title", ""))[:110]
url = _esc(n.get("url", ""))
title_html = f'{title}' if url else title
out += f"""
{stag}{title_html}
{_esc(src)}
"""
return out
def mover_chips(coins):
chips = ""
for c in coins:
chg = c.get("change_24h") or 0
col = "#2ECC71" if chg >= 0 else "#E05252"
sign = "+" if chg >= 0 else ""
chips += f'{c.get("symbol","")}{sign}{chg:.1f}%'
return chips
trend_chips = "".join(
f'#{i+1} {_esc(c.get("name",""))} ({_esc(c.get("symbol",""))})'
for i, c in enumerate(trending)
)
return f"""
■ CRYPTO NARRATIVE BRIEF
{when}
24-HOUR MARKET SENTIMENT
{idx}/100
{label}
{arrow} {('+' if dpts>=0 else '')}{dpts} pts vs 24h ago · {o.get('count',0)} stories analysed
"""
# Recipients are BCC'd in one call; providers cap how many per message, so we
# send in batches. 90 keeps us safely under Brevo's ~99-per-message limit.
def send_brief(recipients: list[str], subject: str, html: str) -> tuple[bool, str]:
"""Send the HTML brief to each recipient INDIVIDUALLY (one personal email
per person). No BCC and no copy to the sender — so you don't receive your
own test/brief sends, recipients never see each other, and deliverability
improves (BCC-to-many looks like spam). Prefers Brevo HTTP, falls back to
SMTP."""
if not recipients:
return False, "No recipients"
if not (brevo_configured() or smtp_configured()):
return False, "Email not configured (set BREVO_API_KEY, or SMTP_* in .env)"
sent, failures = 0, []
for r in recipients:
ok, detail = send_transactional(r, subject, html)
if ok:
sent += 1
else:
failures.append(detail)
if sent == 0:
return False, f"All sends failed: {'; '.join(failures)[:200]}"
if failures:
return True, f"Sent to {sent}/{len(recipients)} ({len(failures)} failed)"
return True, f"Sent to {sent} recipient(s)"
def send_transactional(to_email: str, subject: str, html: str) -> tuple[bool, str]:
"""Send a single direct email (e.g. a signup confirmation) to one address.
Uses Brevo HTTP if a key is set, else falls back to SMTP."""
name, email = _parse_sender()
if not email:
return False, "No sender email (set SMTP_FROM or SMTP_USER)"
if os.getenv("BREVO_API_KEY"):
payload = {
"sender": {"name": name, "email": email},
"to": [{"email": to_email}],
"replyTo": {"email": email, "name": name},
"subject": subject,
"htmlContent": html,
"textContent": _html_to_text(html),
"headers": _deliverability_headers(),
}
try:
resp = requests.post(
BREVO_URL,
headers={
"api-key": os.getenv("BREVO_API_KEY", ""),
"content-type": "application/json",
"accept": "application/json",
},
data=json.dumps(payload),
timeout=20,
)
if resp.status_code in (200, 201):
return True, "sent"
return False, f"Brevo error {resp.status_code}: {resp.text[:200]}"
except Exception as exc:
return False, str(exc)
# SMTP fallback
return _send_smtp([to_email], subject, html)
def _send_brevo(recipients: list[str], subject: str, html: str) -> tuple[bool, str]:
"""Send via Brevo's HTTP API — works on hosts that block SMTP (e.g. HF Spaces)."""
name, email = _parse_sender()
if not email:
return False, "No sender email (set SMTP_FROM or SMTP_USER)"
payload = {
"sender": {"name": name, "email": email},
"to": [{"email": email}], # send to self
"bcc": [{"email": r} for r in recipients], # clients hidden from each other
"replyTo": {"email": email, "name": name},
"subject": subject,
"htmlContent": html,
"textContent": _html_to_text(html),
"headers": _deliverability_headers(),
}
try:
resp = requests.post(
BREVO_URL,
headers={
"api-key": os.getenv("BREVO_API_KEY", ""),
"content-type": "application/json",
"accept": "application/json",
},
data=json.dumps(payload),
timeout=20,
)
if resp.status_code in (200, 201):
return True, f"Sent to {len(recipients)} recipient(s) via Brevo"
return False, f"Brevo error {resp.status_code}: {resp.text[:200]}"
except Exception as exc:
logger.warning(f"[Mailer] Brevo send failed: {exc}")
return False, str(exc)
def _send_smtp(recipients: list[str], subject: str, html: str) -> tuple[bool, str]:
"""Send via SMTP (Gmail App Password). For local / SMTP-allowed hosts."""
host = os.getenv("SMTP_HOST")
port = int(os.getenv("SMTP_PORT", "587"))
user = os.getenv("SMTP_USER")
pw = os.getenv("SMTP_PASS")
sender = os.getenv("SMTP_FROM", user)
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = sender
msg["To"] = sender # recipients go via BCC for privacy
msg.attach(MIMEText("Your email client does not support HTML. Open the web terminal for the brief.", "plain"))
msg.attach(MIMEText(html, "html"))
try:
ctx = ssl.create_default_context()
with smtplib.SMTP(host, port, timeout=20) as server:
server.starttls(context=ctx)
server.login(user, pw)
server.sendmail(sender, [sender] + recipients, msg.as_string())
return True, f"Sent to {len(recipients)} recipient(s)"
except Exception as exc:
logger.warning(f"[Mailer] send failed: {exc}")
return False, str(exc)