Weft is the foundation for AI systems

import os, json, time, base64, threading, logging
from email.mime.text import MIMEText
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
import anthropic, psycopg2, psycopg2.extras
from flask import Flask, request, render_template_string, abort

SCOPES = ["https://www.googleapis.com/auth/gmail.readonly", "https://www.googleapis.com/auth/gmail.send"]
log = logging.getLogger("inbox-bot")
app = Flask(__name__)

def gmail_service():
    creds = Credentials.from_authorized_user_file("token.json", SCOPES) if os.path.exists("token.json") else None
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token: creds.refresh(Request())
        else: raise RuntimeError("run the oauth flow first")  # TODO: who does this on the server?
        open("token.json", "w").write(creds.to_json())
    return build("gmail", "v1", credentials=creds, cache_discovery=False)

gmail, llm = gmail_service(), anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
db = psycopg2.connect(os.environ["DATABASE_URL"], cursor_factory=psycopg2.extras.RealDictCursor)

def decode_body(payload):
    if payload.get("body", {}).get("data"): return base64.urlsafe_b64decode(payload["body"]["data"]).decode("utf-8", "replace")
    for part in payload.get("parts", []) or []:
        if part.get("mimeType") == "text/plain" and part["body"].get("data"): return decode_body(part)
    for part in payload.get("parts", []) or []:
        found = decode_body(part)
        if found: return found  # html only? strip tags? attachments?
    return ""

def draft_reply(text, attempt=0):
    try:
        r = llm.messages.create(model="claude-sonnet-5-5", max_tokens=1024, messages=[{"role": "user", "content": text}])
        return "".join(b.text for b in r.content if b.type == "text")
    except (anthropic.RateLimitError, anthropic.APIConnectionError) as e:
        if attempt >= 5: raise
        time.sleep(2 ** attempt); return draft_reply(text, attempt + 1)

def poll_forever():
    seen = {r["msg_id"] for r in query("SELECT msg_id FROM reviews")}
    while True:
        try:
            res = gmail.users().messages().list(userId="me", q="label:inbox -from:me", maxResults=50).execute()
        except HttpError as e:
            log.warning("gmail list failed: %s", e); time.sleep(30); continue
        for m in res.get("messages", []):
            if m["id"] in seen: continue
            msg = gmail.users().messages().get(userId="me", id=m["id"], format="full").execute()
            headers = {h["name"].lower(): h["value"] for h in msg["payload"]["headers"]}
            draft = draft_reply(decode_body(msg["payload"]))
            execute("INSERT INTO reviews (msg_id, thread_id, sender, message_id, draft, status) VALUES (%s,%s,%s,%s,%s,'pending')",
                    (m["id"], msg["threadId"], headers.get("from"), headers.get("message-id"), draft))
            seen.add(m["id"]); notify_reviewer(m["id"])  # slack? email? sms?
        time.sleep(60)

@app.route("/review/<msg_id>", methods=["GET", "POST"])
def review(msg_id):
    row = one("SELECT * FROM reviews WHERE msg_id = %s", (msg_id,)) or abort(404)
    if request.method == "GET": return render_template_string(REVIEW_PAGE, row=row)  # auth? csrf?
    if request.form.get("decision") != "send": return mark(msg_id, "rejected")
    reply = MIMEText(request.form["reply"])
    reply["To"], reply["In-Reply-To"], reply["References"] = row["sender"], row["message_id"], row["message_id"]
    raw = base64.urlsafe_b64encode(reply.as_bytes()).decode()
    gmail.users().messages().send(userId="me", body={"raw": raw, "threadId": row["thread_id"]}).execute()
    ...

# query, execute, one, mark, notify_reviewer, REVIEW_PAGE, the
# schema, the deploy, restarts mid-review, a second worker, ...
...

import os, json, time, base64, threading, logging
from email.mime.text import MIMEText
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
import anthropic, psycopg2, psycopg2.extras
from flask import Flask, request, render_template_string, abort

SCOPES = ["https://www.googleapis.com/auth/gmail.readonly", "https://www.googleapis.com/auth/gmail.send"]
log = logging.getLogger("inbox-bot")
app = Flask(__name__)

def gmail_service():
    creds = Credentials.from_authorized_user_file("token.json", SCOPES) if os.path.exists("token.json") else None
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token: creds.refresh(Request())
        else: raise RuntimeError("run the oauth flow first")  # TODO: who does this on the server?
        open("token.json", "w").write(creds.to_json())
    return build("gmail", "v1", credentials=creds, cache_discovery=False)

gmail, llm = gmail_service(), anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
db = psycopg2.connect(os.environ["DATABASE_URL"], cursor_factory=psycopg2.extras.RealDictCursor)

def decode_body(payload):
    if payload.get("body", {}).get("data"): return base64.urlsafe_b64decode(payload["body"]["data"]).decode("utf-8", "replace")
    for part in payload.get("parts", []) or []:
        if part.get("mimeType") == "text/plain" and part["body"].get("data"): return decode_body(part)
    for part in payload.get("parts", []) or []:
        found = decode_body(part)
        if found: return found  # html only? strip tags? attachments?
    return ""

def draft_reply(text, attempt=0):
    try:
        r = llm.messages.create(model="claude-sonnet-5-5", max_tokens=1024, messages=[{"role": "user", "content": text}])
        return "".join(b.text for b in r.content if b.type == "text")
    except (anthropic.RateLimitError, anthropic.APIConnectionError) as e:
        if attempt >= 5: raise
        time.sleep(2 ** attempt); return draft_reply(text, attempt + 1)

def poll_forever():
    seen = {r["msg_id"] for r in query("SELECT msg_id FROM reviews")}
    while True:
        try:
            res = gmail.users().messages().list(userId="me", q="label:inbox -from:me", maxResults=50).execute()
        except HttpError as e:
            log.warning("gmail list failed: %s", e); time.sleep(30); continue
        for m in res.get("messages", []):
            if m["id"] in seen: continue
            msg = gmail.users().messages().get(userId="me", id=m["id"], format="full").execute()
            headers = {h["name"].lower(): h["value"] for h in msg["payload"]["headers"]}
            draft = draft_reply(decode_body(msg["payload"]))
            execute("INSERT INTO reviews (msg_id, thread_id, sender, message_id, draft, status) VALUES (%s,%s,%s,%s,%s,'pending')",
                    (m["id"], msg["threadId"], headers.get("from"), headers.get("message-id"), draft))
            seen.add(m["id"]); notify_reviewer(m["id"])  # slack? email? sms?
        time.sleep(60)

@app.route("/review/<msg_id>", methods=["GET", "POST"])
def review(msg_id):
    row = one("SELECT * FROM reviews WHERE msg_id = %s", (msg_id,)) or abort(404)
    if request.method == "GET": return render_template_string(REVIEW_PAGE, row=row)  # auth? csrf?
    if request.form.get("decision") != "send": return mark(msg_id, "rejected")
    reply = MIMEText(request.form["reply"])
    reply["To"], reply["In-Reply-To"], reply["References"] = row["sender"], row["message_id"], row["message_id"]
    raw = base64.urlsafe_b64encode(reply.as_bytes()).decode()
    gmail.users().messages().send(userId="me", body={"raw": raw, "threadId": row["thread_id"]}).execute()
    ...

# query, execute, one, mark, notify_reviewer, REVIEW_PAGE, the
# schema, the deploy, restarts mid-review, a second worker, ...
...