أتمتة فرز Gmail بـ Claude Haiku و Gmail API — من 40 دقيقة لـ 30 ثانية يوميًا

import os
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from anthropic import Anthropic

SCOPES = ["https://www.googleapis.com/auth/gmail.modify"]

LABELS = {
    "Urgent": "إيميل عميل حالي يطلب رد عاجل أو يذكر مشكلة إنتاج أو عطل.",
    "Clients": "تواصل عمل من عميل أو شريك، ليس عاجلًا.",
    "Newsletters": "نشرات بريدية أو محتوى تسويقي أو إعلانات.",
    "Receipts": "فواتير، إيصالات، تأكيد مدفوعات، اشتراكات.",
    "Other": "أي شيء خارج الفئات السابقة."
}

def gmail_service():
    creds = None
    if os.path.exists("token.json"):
        creds = Credentials.from_authorized_user_file("token.json", SCOPES)
    if not creds or not creds.valid:
        flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES)
        creds = flow.run_local_server(port=0)
        open("token.json", "w").write(creds.to_json())
    return build("gmail", "v1", credentials=creds)

def classify(anth, subject, snippet):
    labels_desc = "\n".join([f"- {k}: {v}" for k, v in LABELS.items()])
    msg = anth.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=10,
        system=(
            "صنّف الإيميل في واحدة من الفئات دي فقط:\n"
            f"{labels_desc}\n"
            "رد بكلمة واحدة: اسم الفئة بالظبط، بدون شرح."
        ),
        messages=[{"role": "user", "content": f"العنوان: {subject}\nالملخص: {snippet}"}],
    )
    return msg.content[0].text.strip()

def get_or_create_label(svc, name):
    existing = svc.users().labels().list(userId="me").execute().get("labels", [])
    for lbl in existing:
        if lbl["name"] == name:
            return lbl["id"]
    new = svc.users().labels().create(userId="me", body={"name": name}).execute()
    return new["id"]

def run():
    svc = gmail_service()
    anth = Anthropic()
    label_ids = {name: get_or_create_label(svc, name) for name in LABELS}
    unread = svc.users().messages().list(
        userId="me", q="is:unread newer_than:1d", maxResults=100
    ).execute()
    for m in unread.get("messages", []):
        msg = svc.users().messages().get(
            userId="me", id=m["id"], format="metadata", metadataHeaders=["Subject"]
        ).execute()
        subject = next(
            (h["value"] for h in msg["payload"]["headers"] if h["name"] == "Subject"),
            "",
        )
        snippet = msg.get("snippet", "")[:400]
        category = classify(anth, subject, snippet)
        if category in label_ids:
            svc.users().messages().modify(
                userId="me", id=m["id"], body={"addLabelIds": [label_ids[category]]}
            ).execute()
            print(f"[{category}] {subject[:60]}")

if __name__ == "__main__":
    run()

# crontab -e — تشغيل يومي الساعة 8 صباحًا
0 8 * * * /usr/bin/python3 /home/user/gmail_sort.py >> /var/log/gmail_sort.log 2>&1