AI Automation with Python:
Beginner to Production

import openai, json, os
client = openai.OpenAI(api_key=os.environ["OPENAI_API_KEY"])

PROMPT = (
    "You are an email triage specialist. "
    "Classify into: BILLING | SUPPORT | SALES | ADMIN | PERSONAL | SPAM | OTHER. "
    "Return ONLY valid JSON: "
    '{"category": "CATEGORY", "urgency": 1-5, "summary": "max 10 words"}'
)

def classify(subject, body):
    try:
        r = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": PROMPT},
                {"role": "user", "content": f"Subject: {subject}
Body: {body[:400]}"}
            ],
            temperature=0.1, max_tokens=120,
            response_format={"type": "json_object"}
        )
        result = json.loads(r.choices[0].message.content)
        valid = {"BILLING","SUPPORT","SALES","ADMIN","PERSONAL","SPAM","OTHER"}
        if result.get("category") not in valid:
            result["category"] = "OTHER"
        return result
    except Exception as e:
        return {"category": "OTHER", "urgency": 1,
                "summary": "Classification failed", "error": str(e)}

import schedule, time
from datetime import datetime
from dotenv import load_dotenv
from classifier import classify

load_dotenv()

def process_emails():
    print(f"[{datetime.now().isoformat()}] Processing...")
    # Replace with real Gmail IMAP or Gmail API calls
    emails = [
        {"id":"001","subject":"Invoice overdue","body":"Your payment..."},
        {"id":"002","subject":"How do I export data?","body":"Hi team..."},
    ]
    for e in emails:
        r = classify(e["subject"], e["body"])
        print(f"  [{r['category']}] urgency={r['urgency']} — {e['subject']}")
        # TODO: write r to Google Sheets or CRM

process_emails()  # Run once immediately
schedule.every(15).minutes.do(process_emails)

while True:
    schedule.run_pending()
    time.sleep(30)

from googleapiclient.discovery import build
from google.oauth2.service_account import Credentials

def sheets_client(creds_file="service-account.json"):
    c = Credentials.from_service_account_file(
        creds_file, scopes=["https://www.googleapis.com/auth/spreadsheets"]
    )
    return build("sheets","v4",credentials=c).spreadsheets()

def append_row(spreadsheet_id, sheet, row):
    sheets_client().values().append(
        spreadsheetId=spreadsheet_id,
        range=f"{sheet}!A:Z",
        valueInputOption="USER_ENTERED",
        body={"values": [row]}
    ).execute()

import requests, os

def slack_notify(message, urgent=False):
    emoji = ":rotating_light:" if urgent else ":email:"
    requests.post(
        os.environ["SLACK_WEBHOOK_URL"],
        json={"text": f"{emoji} {message}"},
        timeout=5
    )

import requests, time

def api_call(url, method="GET", headers=None, data=None, retries=3):
    for attempt in range(retries):
        try:
            r = requests.request(
                method, url,
                headers={**{"Content-Type":"application/json"}, **(headers or {})},
                json=data, timeout=10
            )
            r.raise_for_status()
            return r.json()
        except requests.RequestException:
            if attempt == retries - 1: raise
            time.sleep(2 ** attempt)

# /etc/systemd/system/email-automation.service
[Unit]
Description=Email AI Automation Service
After=network.target

[Service]
Type=simple
User=ubuntu
WorkingDirectory=/home/ubuntu/email-automation
ExecStart=/usr/bin/python3 main.py
Restart=always
RestartSec=10
EnvironmentFile=/home/ubuntu/email-automation/.env

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable email-automation
sudo systemctl start email-automation
sudo systemctl status email-automation  # Should show "active (running)"
sudo journalctl -u email-automation -f  # Live log view

import logging, json
from datetime import datetime

class JSONLogger(logging.Formatter):
    def format(self, record):
        entry = {
            "ts": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "msg": record.getMessage()
        }
        if hasattr(record, "extra"):
            entry.update(record.extra)
        return json.dumps(entry)

handler = logging.StreamHandler()
handler.setFormatter(JSONLogger())
logger = logging.getLogger("automation")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Log each classification
logger.info("classified", extra={
    "email_id": "001", "category": "BILLING",
    "urgency": 4, "tokens": 127, "cost_usd": 0.00002
})