""" mailer.py — Sends a polished HTML digest of new Factors.AI account entries, with the full export attached as a CSV file. Design notes ------------ * Card-per-account layout instead of a single wide table. An 11-column table cannot fit inside an email viewport on any client, and Outlook desktop (Word rendering engine) cannot horizontally scroll — a wide table there just renders broken/clipped. Stacked cards render correctly everywhere, including Outlook, Gmail, Apple Mail and mobile clients, with zero horizontal overflow. * All styling is inline (email clients strip most
{f'" if False else ""} {attachment_html}
""" def _build_plain_text(df: pd.DataFrame, is_first_run: bool, csv_filename: Optional[str]) -> str: count = len(df) lines = ["Factors.AI - Account Alerts", "=" * 32, ""] if is_first_run: lines.append(f"Baseline export - {count} account(s) currently in Factors.AI.") else: lines.append(f"{count} new account(s) detected since the last export.") lines.append("") for _, row in df.iterrows(): name = _clean(row.get(_PRIMARY_FIELD)) or _clean(row.get(_DOMAIN_FIELD)) or "Unknown company" domain = _clean(row.get(_DOMAIN_FIELD)) header = name + (f" ({domain})" if domain and domain != name else "") lines.append(f"- {header}") for field in [*_BADGE_FIELDS, *_GRID_FIELDS, _ADDRESS_FIELD]: val = _clean(row.get(field)) if val: lines.append(f" {_display_label(field)}: {val}") linkedin = _clean(row.get(_LINKEDIN_FIELD)) if linkedin: lines.append(f" LinkedIn: {linkedin}") lines.append("") if csv_filename: lines.append(f"Full export attached: {csv_filename}") lines.append("") lines.append("This is an automated message from the Factors.AI Account Export script.") return "\n".join(lines) # ── Attachment ──────────────────────────────────────────────────────────────── def _attach_csv(msg: MIMEMultipart, csv_path: Optional[str]) -> Optional[str]: """Attach the exported CSV to the message. Returns the filename attached, or None.""" if not csv_path: logger.info("No downloaded_csv path provided — sending without an attachment.") return None path = Path(csv_path) if not path.is_file(): logger.warning(f"downloaded_csv not found, skipping attachment: {path}") return None try: data = path.read_bytes() except OSError as exc: logger.warning(f"Could not read downloaded_csv ({path}): {exc} — skipping attachment.") return None part = MIMEApplication(data, _subtype="csv") part.add_header("Content-Disposition", "attachment", filename=path.name) msg.attach(part) logger.info(f"Attached CSV: {path.name} ({len(data):,} bytes)") return path.name # ── Public API ──────────────────────────────────────────────────────────────── def send_alert( smtp_host: str, smtp_port: int, sender_email: str, sender_password: str, recipients: Sequence[str], new_entries: pd.DataFrame, downloaded_csv: Optional[str] = None, *, is_first_run: bool = False, skip_if_empty: bool = True, use_ssl: bool = False, ) -> bool: """ Compose and send the Factors.AI account alert email, with the raw CSV export attached. Parameters ---------- smtp_host / smtp_port : SMTP server. Use use_ssl=True for port 465 (implicit TLS), or leave use_ssl=False for STARTTLS (typically port 587). sender_email : The From address. sender_password : SMTP password / App Password. recipients : List of To addresses. Must be non-empty. new_entries : DataFrame of new rows to include in the digest. downloaded_csv : Path to the full CSV export to attach. If missing/unreadable, the email still sends without the attachment (logged as a warning). is_first_run : True if no previous CSV existed (baseline export copy/subject). skip_if_empty : If True (default) and new_entries is empty and this is not the first run, no email is sent — avoids empty "0 new accounts" noise. use_ssl : Use SMTP_SSL instead of STARTTLS. Returns ------- bool — True if an email was sent, False if sending was skipped (empty digest). """ if not recipients: raise ValueError("recipients must contain at least one address") if new_entries is None: raise ValueError("new_entries cannot be None") count = len(new_entries) if count == 0 and not is_first_run and skip_if_empty: logger.info("No new accounts detected — skipping email (skip_if_empty=True).") return False today = datetime.now().strftime("%Y-%m-%d") subject = ( f"[Factors.AI] Baseline Export — {count} Account{'s' if count != 1 else ''} ({today})" if is_first_run else f"[Factors.AI] {count} New Account{'s' if count != 1 else ''} Detected ({today})" ) msg = MIMEMultipart("mixed") msg["Subject"] = subject msg["From"] = sender_email msg["To"] = ", ".join(recipients) msg["Date"] = formatdate(localtime=True) msg["Message-ID"] = make_msgid() csv_filename = Path(downloaded_csv).name if downloaded_csv and Path(downloaded_csv).is_file() else None alt = MIMEMultipart("alternative") alt.attach(MIMEText(_build_plain_text(new_entries, is_first_run, csv_filename), "plain", "utf-8")) alt.attach(MIMEText(_build_html(new_entries, is_first_run, csv_filename), "html", "utf-8")) msg.attach(alt) _attach_csv(msg, downloaded_csv) logger.info(f"Connecting to {smtp_host}:{smtp_port} ...") try: smtp_cls = smtplib.SMTP_SSL if use_ssl else smtplib.SMTP with smtp_cls(smtp_host, smtp_port, timeout=30) as server: server.ehlo() if not use_ssl: server.starttls() server.ehlo() server.login(sender_email, sender_password) server.sendmail(sender_email, list(recipients), msg.as_string()) except smtplib.SMTPException: logger.exception(f"Failed to send Factors.AI alert email to {recipients}") raise logger.info(f"Email sent to: {recipients} | subject: {subject}") return True def save_preview( new_entries: pd.DataFrame, output_path: str = "preview.html", is_first_run: bool = False, csv_filename: Optional[str] = "factors_ai_export.csv", ) -> str: """Render the email to a local HTML file without sending anything — for quick design checks.""" html_body = _build_html(new_entries, is_first_run, csv_filename) Path(output_path).write_text(html_body, encoding="utf-8") return output_path # """ # mailer.py — Sends an HTML email listing new Factors.AI account entries. # """ # import logging # import smtplib # from datetime import datetime # from email.mime.multipart import MIMEMultipart # from email.mime.text import MIMEText # import pandas as pd # logger = logging.getLogger(__name__) # # Columns to show in the email table (in this order) # _EMAIL_COLS = [ # "Account Domain", # "Company Name", # "Company Industry", # "Company Employee Range", # "Company Annual Revenue", # "Last Activity", # ] # # ── HTML helpers ────────────────────────────────────────────────────────────── # _STYLES = { # "body": "font-family:Arial,sans-serif;font-size:14px;color:#333;margin:0;padding:0;background:#f4f6f9", # "wrapper": "max-width:900px;margin:32px auto;background:#fff;border-radius:8px;overflow:hidden;box-shadow:0 2px 8px rgba(0,0,0,.12)", # "header": "background:#4F46E5;padding:24px 32px", # "h1": "color:#fff;margin:0;font-size:22px", # "sub": "color:#c7d2fe;font-size:13px;margin:6px 0 0", # "body_div": "padding:24px 32px", # "table": "border-collapse:collapse;width:100%;font-size:13px", # "th": "padding:10px 14px;background:#4F46E5;color:#fff;text-align:left;white-space:nowrap", # "td_even": "padding:9px 14px;border-bottom:1px solid #eef0f3;background:#f9fafb", # "td_odd": "padding:9px 14px;border-bottom:1px solid #eef0f3;background:#fff", # "footer": "padding:16px 32px;background:#f4f6f9;font-size:11px;color:#9ca3af", # } # def _make_table(df: pd.DataFrame) -> str: # cols = [c for c in _EMAIL_COLS if c in df.columns] # if not cols: # cols = list(df.columns) # headers = "".join( # f"{c}" for c in cols # ) # rows_html = "" # for i, (_, row) in enumerate(df[cols].iterrows()): # td_style = _STYLES["td_even"] if i % 2 == 0 else _STYLES["td_odd"] # cells = "".join( # f"{str(v)}" for v in row # ) # rows_html += f"{cells}" # tbl_style = _STYLES["table"] # return ( # f"" # f"{headers}" # f"{rows_html}" # f"
" # ) # def _build_html(df: pd.DataFrame, is_first_run: bool) -> str: # count = len(df) # run_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # if is_first_run: # intro = ( # "This is the first run of the automation — no previous " # "export exists for comparison. All current accounts are listed below." # ) # else: # intro = ( # f"{count} new account{'s' if count != 1 else ''} " # f"{'have' if count != 1 else 'has'} been detected since the last export." # ) # table_html = _make_table(df) # return f""" # # #
#
#

Factors.AI — Account Profiles Report

#

Run time: {run_time}

#
#
#

{intro}

#

# Total rows in this export: {count} #

#
# {table_html} #
#
# This is an automated message generated by the Factors.AI Account # Export script. Do not reply to this email. #
#
# # # """ # # ── Public API ──────────────────────────────────────────────────────────────── # def send_alert( # smtp_host: str, # smtp_port: int, # sender_email: str, # sender_password: str, # recipients: list[str], # new_entries: pd.DataFrame, # is_first_run: bool = False, # downloaded_csv: str, # ) -> None: # """ # Compose and send the alert email. # Parameters # ---------- # smtp_host / smtp_port : SMTP server (default: Gmail STARTTLS 587). # sender_email : The From address. # sender_password : SMTP password / App Password. # recipients : List of To addresses. # new_entries : DataFrame of new rows to include. # is_first_run : True if no previous CSV existed. # """ # count = len(new_entries) # today = datetime.now().strftime("%Y-%m-%d") # subject = ( # f"[Factors.AI] First Run — {count} Accounts Exported ({today})" # if is_first_run # else f"[Factors.AI] {count} New Account{'s' if count != 1 else ''} Detected ({today})" # ) # html_body = _build_html(new_entries, is_first_run) # msg = MIMEMultipart("alternative") # msg["Subject"] = subject # msg["From"] = sender_email # msg["To"] = ", ".join(recipients) # msg.attach(MIMEText(html_body, "html", "utf-8")) # logger.info(f"Connecting to {smtp_host}:{smtp_port} …") # with smtplib.SMTP(smtp_host, smtp_port, timeout=30) as server: # server.ehlo() # server.starttls() # server.login(sender_email, sender_password) # server.sendmail(sender_email, recipients, msg.as_string()) # logger.info(f"Email sent to: {recipients} | subject: {subject}")