grants-rag / app /notify.py
michaellupo74's picture
feat: notifications module; ignore exports and env files
4a182f7
raw
history blame
3.86 kB
# 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"<div style='color:#64748b'>Original: {deadline_text}</div>" if deadline_text else ""
syn = (synopsis or "").replace("\n", "<br/>")[:1200]
return f"""
<div style="font-family:system-ui,-apple-system,Segoe UI,Roboto,Arial,sans-serif">
<h2 style="margin:0 0 8px 0">{title}</h2>
<div>Deadline: <strong>{dl}</strong></div>
{raw}
<p style="margin-top:12px">{syn}</p>
<p><a href="{url}">{url}</a></p>
</div>
"""
# ---------- 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"