app.py

"""ColdScore — cold-email deliverability scoring on an open-weights model.

Hugging Face Space. Loads a fine-tuned ModernBERT from the Hub and runs it in
this container. No proprietary model API is called.

Set MODEL_ID to your Hub repo before deploying.
"""

import csv
import io
import os
import re

import gradio as gr
import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer

MODEL_ID = os.environ.get("MODEL_ID", "Ashsinha1/coldscore-modernbert")
MAX_LEN = 512

# Lazy-load: the Space is deployed before the model is pushed, and loading at
# import time would crash-loop the container until the repo exists. This way the
# UI boots immediately and picks the model up on the first request after it lands.
_M = {"tok": None, "model": None, "error": None}


def get_model():
    if _M["model"] is None:
        try:
            _M["tok"] = AutoTokenizer.from_pretrained(MODEL_ID)
            m = AutoModelForSequenceClassification.from_pretrained(MODEL_ID)
            m.eval()
            _M["model"], _M["error"] = m, None
        except Exception as e:  # repo missing, still uploading, or gated
            _M["error"] = (
                f"Model `{MODEL_ID}` isn't loadable yet.\n\n"
                f"Train and push it first (see the Colab one-liner), then hit **Score it** "
                f"again — no redeploy needed.\n\n<sub>{type(e).__name__}: {e}</sub>"
            )
    return _M

GRADES = [
    (0.15, "A", "Should land in the primary inbox."),
    (0.35, "B", "Likely to deliver. A couple of things worth tightening."),
    (0.60, "C", "Coin flip. The promotions tab is a real possibility."),
    (0.85, "D", "High spam risk. Rewrite before sending at volume."),
    (1.01, "F", "Very high spam risk. Sending this will hurt your domain reputation."),
]


def grade_for(risk):
    for cutoff, letter, verdict in GRADES:
        if risk < cutoff:
            return letter, verdict
    return "F", GRADES[-1][2]


@torch.inference_mode()
def score_many(pairs):
    """Score a list of (subject, body). Batched — one forward pass per chunk."""
    M = get_model()
    if M["model"] is None:
        raise RuntimeError(M["error"])
    tokenizer, model = M["tok"], M["model"]
    out = []
    for i in range(0, len(pairs), 16):
        chunk = pairs[i:i + 16]
        enc = tokenizer([s for s, _ in chunk], [b for _, b in chunk],
                        truncation=True, padding=True, max_length=MAX_LEN,
                        return_tensors="pt")
        probs = model(**enc).logits.float().softmax(-1)[:, 1]
        out.extend(float(p) for p in probs)
    return out


# --- fix suggestions -------------------------------------------------------
# These are deterministic checks against published deliverability guidance, NOT
# the model's reasoning. ModernBERT can't tell you why it scored something —
# claiming otherwise would be inventing an explanation. Kept separate on purpose.

URL_RE = re.compile(r"https?://\S+|www\.\S+")
MERGE_RE = re.compile(r"\{\{?\s*[a-z_]+\s*\}?\}", re.I)


def suggestions(subject, body):
    tips = []
    full = f"{subject}\n{body}"
    letters = [c for c in full if c.isalpha()]
    caps = sum(1 for c in letters if c.isupper()) / len(letters) if letters else 0

    if MERGE_RE.search(full):
        tips.append("**Unrendered merge tag** — something like `{{first_name}}` will send literally. "
                    "This is the single most damaging thing in the list.")
    if caps > 0.3:
        tips.append(f"**{caps:.0%} of letters are capitals.** Shouting is a classic filter trigger.")
    if re.search(r"\b[A-Z]{4,}\b", full):
        tips.append("**ALL-CAPS words.** Drop them to sentence case.")
    n_ex = full.count("!")
    if n_ex > 2:
        tips.append(f"**{n_ex} exclamation marks.** One is plenty; zero is usually better.")
    links = len(URL_RE.findall(body))
    if links > 2:
        tips.append(f"**{links} links.** Multiple links in a first-touch email reads as bulk mail. Cut to one.")
    if re.match(r"^\s*(re|fwd?)\s*:", subject, re.I):
        tips.append("**Fake `Re:`/`Fwd:` prefix.** It implies a conversation that never happened — "
                    "recipients mark it as spam, and that costs you domain reputation.")
    if len(body.split()) < 25:
        tips.append("**Very short body.** Too little content to look like a real 1:1 email.")
    if len(body.split()) > 220:
        tips.append(f"**{len(body.split())} words.** Long first-touch emails lose replies; aim for 90–130.")
    if len(subject) > 60:
        tips.append(f"**{len(subject)}-character subject.** It'll be truncated on mobile. Aim under 50.")
    if "unsubscribe" not in body.lower():
        tips.append("_No opt-out._ Not required for genuine 1:1 mail, but at volume its absence "
                    "raises complaint rates — which is what actually burns a domain.")

    if not tips:
        return "No mechanical issues found. Anything left is about the writing itself."
    return "\n\n".join("- " + t for t in tips)


def score_single(subject, body):
    subject = (subject or "").strip()
    body = (body or "").strip()
    if not subject and not body:
        return {}, "Enter a subject and a body.", ""

    try:
        risk = score_many([(subject, body)])[0]
    except RuntimeError as e:
        return {}, str(e), ""
    letter, verdict = grade_for(risk)
    headline = f"### Grade {letter} — {risk:.1%} spam risk\n{verdict}"
    return ({"spam-prone": risk, "deliverable": 1 - risk}, headline, suggestions(subject, body))


# --- batch -----------------------------------------------------------------

def score_batch(file, pasted):
    """CSV with subject,body columns — or one email per line as 'subject | body'."""
    pairs, errors = [], []

    if file is not None:
        try:
            with open(file.name, encoding="utf-8-sig", newline="") as fh:
                reader = csv.DictReader(fh)
                cols = {c.lower().strip(): c for c in (reader.fieldnames or [])}
                if "subject" not in cols or "body" not in cols:
                    return [], f"CSV needs `subject` and `body` columns. Found: {reader.fieldnames}"
                for row in reader:
                    pairs.append(((row[cols["subject"]] or "").strip(),
                                  (row[cols["body"]] or "").strip()))
        except Exception as e:
            return [], f"Could not read that CSV: {e}"

    if pasted and pasted.strip():
        for line in pasted.strip().splitlines():
            if not line.strip():
                continue
            subject, _, body = line.partition("|")
            pairs.append((subject.strip(), body.strip()))

    if not pairs:
        return [], "Upload a CSV or paste some emails first."

    pairs = pairs[:200]
    try:
        risks = score_many(pairs)
    except RuntimeError as e:
        return [], str(e)
    rows = []
    for (s, b), r in zip(pairs, risks):
        letter, _ = grade_for(r)
        rows.append([letter, f"{r:.1%}", s[:70], f"{len(b.split())}w"])
    rows.sort(key=lambda x: -float(x[1].rstrip("%")))

    risky = sum(1 for r in risks if r >= 0.6)
    note = (f"Scored **{len(pairs)}** emails. **{risky}** at grade C or worse "
            f"({risky / len(pairs):.0%}). Sorted worst first.")
    if len(pairs) == 200:
        note += "\n\n_Capped at 200 rows for this demo._"
    return rows, note


GOOD = ("quick question about Northwind's onboarding",
        "Hi Sarah,\n\nI noticed Northwind shipped a self-serve trial last month. Curious how "
        "you're handling the handoff from trial to sales conversation right now.\n\nWe built "
        "something for that and I'd rather hear how you do it before I assume anything.\n\n"
        "Worth 15 minutes next week?\n\nAshish")
BAD = ("ACT NOW!!! Northwind qualifies for FREE revenue growth!!!",
       "Dear Sir/Madam,\n\nCONGRATULATIONS!!! Northwind has been SELECTED for our EXCLUSIVE "
       "offer!\n\nGUARANTEED 10X ROI or your money back!!! This is a LIMITED TIME offer that "
       "EXPIRES at midnight!\n\nCLICK HERE NOW: http://bit.ly/xy12\n\nDON'T MISS OUT!!! ACT FAST!!!")
MERGE = ("{{first_name}} - I have an amazing offer for {{company}}",
         "Hi {{first_name}},\n\nWe are the best in the world at what we do and our results are "
         "guaranteed.\n\nSign up free today! Click here to buy now: http://example.com/signup\n\n"
         "Best,\nSales Team")

with gr.Blocks(title="ColdScore", theme=gr.themes.Soft()) as demo:
    gr.Markdown(
        f"""
        # 🧲 ColdScore
        ### Will this cold email reach the inbox — or the spam folder?

        A fine-tuned **[ModernBERT](https://huggingface.co/answerdotai/ModernBERT-base)**
        (Apache-2.0, 149M params) running on a free CPU. Open weights, MIT licence.
        **No proprietary model API is called** — the weights are public, and you can run
        this identical model inside your own network.
        """
    )

    with gr.Tab("Score an email"):
        with gr.Row():
            with gr.Column(scale=3):
                subject = gr.Textbox(label="Subject", placeholder="quick question about onboarding")
                body = gr.Textbox(label="Body", lines=12, placeholder="Hi Sarah,\n\nI noticed...")
                btn = gr.Button("Score it", variant="primary", size="lg")
            with gr.Column(scale=2):
                headline = gr.Markdown()
                label = gr.Label(label="Model output", num_top_classes=2)

        gr.Markdown("#### What to fix")
        tips = gr.Markdown()
        gr.Markdown(
            "_The fix list is deterministic checks against published deliverability guidance — "
            "not the model's reasoning. A 149M-parameter transformer can't tell you why it "
            "scored something, and inventing an explanation would be worse than not having one._"
        )

        gr.Examples(examples=[list(GOOD), list(BAD), list(MERGE)],
                    inputs=[subject, body], label="Try one")

        btn.click(score_single, [subject, body], [label, headline, tips])

    with gr.Tab("Score a campaign"):
        gr.Markdown(
            "Score a whole list at once — the actual job if you're sending at volume.\n\n"
            "Upload a **CSV with `subject` and `body` columns**, or paste one email per line "
            "as `subject | body`."
        )
        with gr.Row():
            up = gr.File(label="CSV", file_types=[".csv"])
            paste = gr.Textbox(label="Or paste", lines=6,
                               placeholder="quick question about onboarding | Hi Sarah, I noticed...")
        bbtn = gr.Button("Score all", variant="primary")
        bnote = gr.Markdown()
        table = gr.Dataframe(headers=["Grade", "Risk", "Subject", "Length"],
                             datatype=["str", "str", "str", "str"],
                             wrap=True, label="Worst first")
        bbtn.click(score_batch, [up, paste], [table, bnote])

    with gr.Tab("Use it yourself"):
        gr.Markdown(
            f"""
            The model is public. Run it wherever your data is allowed to live:

            ```python
            from transformers import pipeline

            clf = pipeline("text-classification", model="{MODEL_ID}")
            clf({{"text": "quick question about onboarding", "text_pair": "Hi Sarah, I noticed..."}})
            # [{{'label': 'deliverable', 'score': 0.98}}]
            ```

            ### Why an open model instead of a frontier API

            Scoring a cold email means handing over a prospect's name, their email address, and
            your pitch. That's customer data. Sending it to a third-party model provider is a
            decision that shows up in your customers' security review.

            This model doesn't force that decision: the weights are public, inference is a CPU
            process, and it runs inside your own network with the four lines above. A frontier
            model would score these emails better — but for the half of the pipeline that touches
            every contact in the database, "good and local" beats "excellent and elsewhere".

            ### Limitations — read before trusting a number

            1. **Content only.** Inbox placement is dominated by things the model cannot see:
               domain and IP reputation, SPF/DKIM/DMARC alignment, sending volume, list hygiene,
               complaint rates. A perfect email from a burned domain still lands in spam.
            2. **The benchmark is easy.** Trained on 100 hand-written emails whose spam examples
               are unsubtle. A keyword blocklist already scores ~0.94 F1 on that corpus, so
               held-out numbers are an optimistic upper bound, not real-world accuracy.
            3. **Labels are judgements, not outcomes.** Nobody sent these emails and measured
               where they landed.
            4. **English only.** Not a phishing detector — well-written phishing scores clean.

            Treat it as a content lint, not a deliverability guarantee.

            ### Privacy

            Text you enter is processed in this container and is not stored or logged. If that
            still isn't good enough for your data — which is a reasonable position, and rather
            the point of the project — clone the Space or run the model locally.
            """
        )

if __name__ == "__main__":
    demo.launch()