# app/notify.py from __future__ import annotations import os, re, hashlib, time, requests, smtplib, ssl from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from typing import Optional def _env(name: str, default: Optional[str] = None) -> Optional[str]: v = os.getenv(name) return v if v is not None and v != "" else default # ---------- Slack ---------- def send_slack(message: str, webhook_url: Optional[str] = None) -> bool: url = webhook_url or _env("SLACK_WEBHOOK_URL") if not url: return False try: r = requests.post(url, json={"text": message}, timeout=10) return r.status_code // 100 == 2 except Exception: return False # ---------- Email (SMTP) ---------- def send_email(subject: str, text: str, html: Optional[str] = None, to: Optional[list[str]] = None) -> bool: host = _env("SMTP_HOST") user = _env("SMTP_USER") pwd = _env("SMTP_PASS") port = int(_env("SMTP_PORT", "587")) sender = _env("SMTP_FROM", user) recipients = to or (_env("NOTIFY_TO","") or "").replace(";", ",").split(",") recipients = [x.strip() for x in recipients if x.strip()] if not (host and user and pwd and sender and recipients): return False msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = sender msg["To"] = ", ".join(recipients) msg.attach(MIMEText(text, "plain", "utf-8")) if html: msg.attach(MIMEText(html, "html", "utf-8")) try: ctx = ssl.create_default_context() with smtplib.SMTP(host, port, timeout=15) as s: s.starttls(context=ctx) s.login(user, pwd) s.sendmail(sender, recipients, msg.as_string()) return True except Exception: return False def htmlify(title: str, url: str, synopsis: str, deadline_iso: str | None, deadline_text: str | None) -> str: dl = deadline_iso or "TBD" raw = f"
Original: {deadline_text}
" if deadline_text else "" syn = (synopsis or "").replace("\n", "
")[:1200] return f"""

{title}

Deadline: {dl}
{raw}

{syn}

{url}

""" # ---------- ICS (calendar) ---------- def _safe_name(s: str) -> str: s = re.sub(r"[^\w\-. ]+", "", s.strip()) or "event" return re.sub(r"\s+", "_", s)[:64] def build_ics_all_day(summary: str, date_yyyy_mm_dd: Optional[str], description: str = "", url: str = "") -> str: # All-day event on deadline date; if None => today from datetime import datetime, timezone uid = hashlib.sha1(f"{summary}|{date_yyyy_mm_dd}|{url}".encode("utf-8")).hexdigest() + "@grants-rag" dtstamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ") if not date_yyyy_mm_dd: date_yyyy_mm_dd = datetime.utcnow().date().isoformat() y, m, d = date_yyyy_mm_dd.split("-") dt = f"{y}{m}{d}" # VALUE=DATE desc = description or "" if url: desc = (desc + "\n" + url).strip() ics = [ "BEGIN:VCALENDAR", "VERSION:2.0", "PRODID:-//grants-rag//EN", "CALSCALE:GREGORIAN", "METHOD:PUBLISH", "BEGIN:VEVENT", f"UID:{uid}", f"DTSTAMP:{dtstamp}", f"DTSTART;VALUE=DATE:{dt}", f"SUMMARY:{summary}", f"DESCRIPTION:{desc}", f"URL:{url}" if url else "", "END:VEVENT", "END:VCALENDAR", "", ] return "\n".join([x for x in ics if x]) def filename_for_ics(title: str, deadline_iso: Optional[str]) -> str: tag = (deadline_iso or "TBD").replace("-", "") return f"{_safe_name(title)}_{tag}.ics"