Optimize a campaign

id,channel,draft
step-1,email,"Hi {{first_name}},

I hope this finds you well! I'm reaching out because {{company}} is growing fast and I imagine onboarding new reps is taking up a lot of your managers' time.

Rampwise gives every new rep a guided first 30 days, so managers spend less time shadowing calls and more time coaching.

Would you be open to a quick call next week to see if it could help {{company}}?

Thanks,
{{sender_first_name}}"
step-2,linkedin,"Hi {{first_name}}, thanks for connecting! ..."

#!/usr/bin/env python3
"""Optimize every draft in a CSV with the Amdahl Message Optimizer.

Usage: AMDAHL_KEY=amdhl_... python3 optimize_campaign.py drafts.csv optimized.csv
Input columns: id, draft, and optionally channel (email or linkedin).
Re-runnable: an id already in the output file is skipped.
"""
import csv, json, os, sys, time, urllib.error, urllib.request

API = "https://app.amdahl.ai/api/platform/v1/messages/optimize"
KEY = os.environ["AMDAHL_KEY"]
SRC = sys.argv[1] if len(sys.argv) > 1 else "drafts.csv"
OUT = sys.argv[2] if len(sys.argv) > 2 else "optimized.csv"
# Applied to every draft. Leave a list empty, or fill it from your style guide.
CONTEXT = {"rules": [], "voice_examples": []}
COLUMNS = ["id", "channel", "outcome", "draft", "returned", "summary",
           "edits", "questions", "notes", "reason", "run_id"]


def post(body):
    """POST one draft. Returns (HTTP status, parsed body, seconds to wait on a 429)."""
    req = urllib.request.Request(
        API, data=json.dumps(body).encode("utf-8"), method="POST",
        headers={"X-API-Key": KEY, "Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=180) as res:
            return res.status, json.load(res), 0
    except urllib.error.HTTPError as err:
        raw = err.read().decode("utf-8", "replace")
        retry_after = err.headers.get("Retry-After") or ""
        wait = int(retry_after) if retry_after.isdigit() else 60
        try:
            return err.code, json.loads(raw), wait
        except ValueError:  # the per-IP limiter answers in plain text
            return err.code, {"error": {"code": f"http_{err.code}", "message": raw}}, wait


def optimize(row):
    body = {"message": row["draft"]}
    if (row.get("channel") or "").strip():
        body["channel"] = row["channel"].strip()
    context = {k: v for k, v in CONTEXT.items() if v}
    if context:
        body["context"] = context
    while True:
        try:
            status, payload, wait = post(body)
        except OSError as err:  # no answer within 180 seconds, or the network failed
            return {"ok": False, "reason": f"no_answer: {err}"}
        data = payload.get("data", payload)
        error = data.get("error") if isinstance(data.get("error"), dict) else {}
        code = error.get("code")
        if code == "quota_exceeded":
            sys.exit("Monthly Optimizer cap reached. Re-run next month; finished rows are kept.")
        if status in (401, 403):
            sys.exit(f"{status}: {error.get('message') or payload}. Mint a Customer agent key.")
        if status == 429:  # a rate limit, not the cap: wait and retry this draft
            time.sleep(wait)
            continue
        if status != 200:
            return {"ok": False, "reason": code or f"http_{status}"}
        return data


fresh = not os.path.exists(OUT) or os.path.getsize(OUT) == 0
done = set()
if not fresh:
    with open(OUT, newline="", encoding="utf-8") as f:
        done = {r["id"] for r in csv.DictReader(f)}

with open(SRC, newline="", encoding="utf-8") as f:
    rows = [r for r in csv.DictReader(f) if r["id"] not in done]

with open(OUT, "a", newline="", encoding="utf-8") as f:
    out = csv.DictWriter(f, fieldnames=COLUMNS)
    if fresh:
        out.writeheader()
    for n, row in enumerate(rows, 1):
        r = optimize(row)
        if not r.get("ok"):
            outcome = "failed"
        elif r.get("unchanged"):
            outcome = "kept"
        else:
            outcome = "rewritten"
        out.writerow({
            "id": row["id"],
            "channel": (r.get("extraction") or {}).get("channel") or row.get("channel", ""),
            "outcome": outcome,
            "draft": row["draft"],
            "returned": r.get("message", ""),
            "summary": r.get("summary", ""),
            "edits": " | ".join(s["says"] for s in r.get("kept_suggestions") or []),
            "questions": " | ".join(q["asks"] for q in r.get("ask") or []),
            "notes": " ".join(note["code"] for note in r.get("notes") or []),
            "reason": r.get("reason", ""),
            "run_id": r.get("run_id", ""),
        })
        f.flush()
        print(f"[{n}/{len(rows)}] {row['id']}: {outcome}")

export AMDAHL_KEY="amdhl_your_key_here"
python3 optimize_campaign.py drafts.csv optimized.csv

curl -s "https://app.amdahl.ai/api/platform/v1/setup/status" -H "X-API-Key: $AMDAHL_KEY" | jq '.data.optimize'

Hi {{first_name}},

Saw that {{company}} has been growing quickly, which usually means a wave of new reps to onboard and managers who suddenly have very little time left for anything else.

Rampwise gives each new rep a guided first 30 days, so managers spend less time shadowing calls and more time actually coaching.

Is that the kind of thing you're running into right now?

Thanks,
{{sender_first_name}}