أتمتة استخراج فواتير Gmail إلى Google Sheets بـ Claude Haiku — تتبّع مصاريفك بدون ما تفتح إيميل

import base64
import json
import os
from datetime import datetime, timedelta

import anthropic
import gspread
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build

GMAIL_QUERY = 'newer_than:30d (invoice OR فاتورة OR receipt OR billing)'
SHEET_ID = os.environ['SHEET_ID']
MODEL = 'claude-haiku-4-5-20251001'

client = anthropic.Anthropic()
gc = gspread.service_account(filename='service_account.json')
sheet = gc.open_by_key(SHEET_ID).sheet1

EXTRACT_TOOL = [{
    'name': 'save_invoice',
    'description': 'احفظ بيانات فاتورة مستخرجة من إيميل.',
    'input_schema': {
        'type': 'object',
        'properties': {
            'is_invoice': {'type': 'boolean'},
            'vendor': {'type': 'string'},
            'invoice_number': {'type': 'string'},
            'amount': {'type': 'number'},
            'currency': {'type': 'string'},
            'issue_date': {'type': 'string'},
            'due_date': {'type': 'string'}
        },
        'required': ['is_invoice']
    }
}]

def extract_invoice(email_text: str) -> dict | None:
    msg = client.messages.create(
        model=MODEL,
        max_tokens=512,
        tools=EXTRACT_TOOL,
        tool_choice={'type': 'tool', 'name': 'save_invoice'},
        messages=[{
            'role': 'user',
            'content': f'استخرج بيانات الفاتورة من الإيميل ده. '
                       f'لو مش فاتورة، رجّع is_invoice=false.\n\n{email_text[:6000]}'
        }]
    )
    for block in msg.content:
        if block.type == 'tool_use':
            data = block.input
            return data if data.get('is_invoice') else None
    return None

def fetch_emails():
    creds = Credentials.from_authorized_user_file('token.json')
    gmail = build('gmail', 'v1', credentials=creds)
    resp = gmail.users().messages().list(
        userId='me', q=GMAIL_QUERY, maxResults=200
    ).execute()
    for ref in resp.get('messages', []):
        msg = gmail.users().messages().get(
            userId='me', id=ref['id'], format='full'
        ).execute()
        yield parse_email_body(msg)

def parse_email_body(msg) -> str:
    for part in msg['payload'].get('parts', [msg['payload']]):
        if part['mimeType'] in ('text/plain', 'text/html'):
            data = part['body'].get('data', '')
            return base64.urlsafe_b64decode(data).decode('utf-8', 'ignore')
    return ''

def run():
    seen = {row[1] for row in sheet.get_all_values()[1:] if len(row) > 1}
    new_rows = []
    for body in fetch_emails():
        invoice = extract_invoice(body)
        if not invoice or invoice['invoice_number'] in seen:
            continue
        new_rows.append([
            invoice.get('vendor', ''),
            invoice.get('invoice_number', ''),
            invoice.get('amount', 0),
            invoice.get('currency', 'USD'),
            invoice.get('issue_date', ''),
            invoice.get('due_date', ''),
        ])
    if new_rows:
        sheet.append_rows(new_rows)
    print(f'تم إضافة {len(new_rows)} فاتورة جديدة')

if __name__ == '__main__':
    run()

0 7 * * * cd /home/you/invoices && /usr/bin/python3 main.py >> run.log 2>&1

{
  "is_invoice": true,
  "vendor": "AWS",
  "invoice_number": "9876543210",
  "amount": 247.33,
  "currency": "USD",
  "issue_date": "2026-03-31",
  "due_date": "2026-04-15"
}