<- 3.1 Agent PAL core demos

[3.3.7] pal_core_07_gmail_alerts.py // real emails as sensor stream (BINGO) 25.0423

For 3.7, the cleanest interpretation is:
email inbox = real-world sensor input
then
PAL converts selected emails into normalized events
That gives you a very good bridge from:
•	earlier deterministic PAL demos
to 
•	later 3.8 natural language → plan 
•	and 3.10 UI 
________________________________________
Recommended scope for 3.7
Keep it minimal:
•	connect to Gmail via IMAP 
•	read recent matching emails 
•	extract basic fields 
•	convert to PAL event format 
•	save to JSON db 
•	avoid LLM for now 
So this demo is about sensor ingestion, not reasoning.

This keeps the demo atomic:
•	sensor = Gmail inbox 
•	ingest = IMAP fetch 
•	normalize = email → PAL event 
•	store = JSON db 
•	dedupe = message-id 
So it shows a core PAL idea:
external messy input comes in,
PAL turns it into structured machine-usable state.
________________________________________
Small note
For a first pass, I would not do OAuth or Gmail API.
Too much overhead for this demo.
Use:
•	Gmail 
•	IMAP 
•	App Password 
That is the simplest path.
# pal_core_07_gmail_alerts.py
#
# 3.7 -- real emails as sensor stream
#
# Purpose:
#   Read selected Gmail messages via IMAP and convert them into normalized PAL events.
#
# Environment variables:
#   PAL_EMAIL_ADDRESS=your_email@gmail.com
#   PAL_EMAIL_APP_PASSWORD=your_16_char_app_password
#   PAL_IMAP_SERVER=imap.gmail.com
#
# Notes:
#   - For Gmail, use an App Password, not your normal password.
#   - This demo is deterministic on purpose.
#   - No LLM used here yet. We only ingest and normalize.
#
# Commands:
#   python pal_core_07_gmail_alerts.py reset
#   python pal_core_07_gmail_alerts.py fetch
#   python pal_core_07_gmail_alerts.py show
#   python pal_core_07_gmail_alerts.py loop --interval 60
#
# Example:
#   Add these values to .env:
#   PAL_EMAIL_ADDRESS=you@gmail.com
#   PAL_EMAIL_APP_PASSWORD=xxxx xxxx xxxx xxxx
#   PAL_IMAP_SERVER=imap.gmail.com
#   python pal_core_07_gmail_alerts.py fetch

import argparse
import email
import imaplib
import json
import os
import re
import sys
import time
from datetime import datetime, timezone
from email.message import Message
from email.header import decode_header
from typing import Any, Dict, List, Optional, Tuple

# --------------------------------------------------
# 1. Config
# --------------------------------------------------

# CODEX CHANGE: Load PAL Gmail settings from a local .env file before reading os.getenv.
def load_dotenv_file(path: str = ".env") -> None:
    if not os.path.exists(path):
        return

    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue

            key, value = line.split("=", 1)
            key = key.strip()
            value = value.strip().strip('"').strip("'")

            if key and key not in os.environ:
                os.environ[key] = value

# CODEX CHANGE: Make .env values available to the config below.
load_dotenv_file()

DB_FILE = "pal_events07.json"
STATE_FILE = "pal_email_state07.json"

# CODEX CHANGE: PAL_IMAP_SERVER now comes from .env, with Gmail as the fallback.
DEFAULT_IMAP_SERVER = os.getenv("PAL_IMAP_SERVER", "imap.gmail.com")
DEFAULT_LABEL = "INBOX"
DEFAULT_MAX_RESULTS = 10

# Optional filters to keep the demo clean.
# You can expand these later.
WATCH_SENDERS = [
    # examples:
    # "alerts@company.com",
    # "noreply@monitoring.com",
]
WATCH_SUBJECT_KEYWORDS = [
    "alert",
    "incident",
    "delay",
    "blocked",
    "failure",
    "warning",
    "shipment",
    "delivery",
]

# --------------------------------------------------
# 2. Small helpers
# --------------------------------------------------

def utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()

def load_json(path: str, default: Any) -> Any:
    if not os.path.exists(path):
        return default
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)

def save_json(path: str, data: Any) -> None:
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)

def decode_mime_words(value: Optional[str]) -> str:
    if not value:
        return ""
    parts = decode_header(value)
    out = []
    for part, enc in parts:
        if isinstance(part, bytes):
            out.append(part.decode(enc or "utf-8", errors="replace"))
        else:
            out.append(part)
    return "".join(out)

def normalize_whitespace(s: str) -> str:
    return re.sub(r"\s+", " ", s).strip()

def safe_lower(s: Optional[str]) -> str:
    return (s or "").strip().lower()

def ensure_db() -> Dict[str, Any]:
    db = load_json(DB_FILE, {"events": []})
    if "events" not in db or not isinstance(db["events"], list):
        db = {"events": []}
    return db

def ensure_state() -> Dict[str, Any]:
    state = load_json(STATE_FILE, {"seen_email_ids": []})
    if "seen_email_ids" not in state or not isinstance(state["seen_email_ids"], list):
        state = {"seen_email_ids": []}
    return state

# --------------------------------------------------
# 3. Email body extraction
# --------------------------------------------------

# CODEX CHANGE: Use Message directly so the type annotation does not depend on email.message being loaded as a module attribute.
def extract_text_from_message(msg: Message) -> str:
    """
    Return a plain text body if possible.
    Keep this simple and deterministic.
    """
    if msg.is_multipart():
        chunks = []
        for part in msg.walk():
            content_type = part.get_content_type()
            content_disposition = str(part.get("Content-Disposition", "")).lower()

            if "attachment" in content_disposition:
                continue

            if content_type == "text/plain":
                payload = part.get_payload(decode=True)
                if payload:
                    charset = part.get_content_charset() or "utf-8"
                    chunks.append(payload.decode(charset, errors="replace"))
        return normalize_whitespace(" ".join(chunks))

    payload = msg.get_payload(decode=True)
    if payload:
        charset = msg.get_content_charset() or "utf-8"
        return normalize_whitespace(payload.decode(charset, errors="replace"))
    return ""

# --------------------------------------------------
# 4. Deterministic parsing
# --------------------------------------------------

def infer_status(subject: str, body: str) -> str:
    text = f"{subject} {body}".lower()

    if any(x in text for x in ["blocked", "stuck", "halted"]):
        return "blocked"
    if any(x in text for x in ["delayed", "delay", "late"]):
        return "delayed"
    if any(x in text for x in ["failed", "failure", "error", "exception"]):
        return "failed"
    if any(x in text for x in ["warning", "degraded", "risk"]):
        return "warning"
    if any(x in text for x in ["delivered", "resolved", "recovered", "restored"]):
        return "resolved"
    return "alert"

def infer_priority(status: str, subject: str, body: str) -> int:
    text = f"{subject} {body}".lower()

    if "critical" in text or "sev1" in text or status == "blocked":
        return 9
    if "high" in text or "sev2" in text or status in ("failed", "delayed"):
        return 7
    if status == "warning":
        return 5
    if status == "resolved":
        return 2
    return 4

def extract_entity_id(subject: str, body: str) -> str:
    text = f"{subject} {body}"

    patterns = [
        r"\b(truck[_\- ]?\d+)\b",
        r"\b(shipment[_\- ]?\d+)\b",
        r"\b(order[_\- ]?\d+)\b",
        r"\b(site[_\- ]?\d+)\b",
        r"\b(sensor[_\- ]?\d+)\b",
        r"\b(device[_\- ]?\d+)\b",
    ]
    for pat in patterns:
        m = re.search(pat, text, re.IGNORECASE)
        if m:
            return m.group(1).replace(" ", "_").lower()

    return "email_signal"

def extract_location(subject: str, body: str) -> str:
    text = f"{subject} {body}".lower()

    known_locations = [
        "taipei",
        "tainan",
        "kaohsiung",
        "tokyo",
        "berlin",
        "kyiv",
        "warsaw",
        "new york",
        "hoboken",
        "site_1",
        "site_2",
        "site_3",
    ]
    for loc in known_locations:
        if loc in text:
            return loc.replace(" ", "_")
    return "unknown"

def infer_event_type(status: str, subject: str, body: str) -> str:
    text = f"{subject} {body}".lower()

    if any(x in text for x in ["shipment", "delivery", "carrier", "warehouse"]):
        return "logistics_alert"
    if any(x in text for x in ["server", "service", "api", "database", "latency"]):
        return "system_alert"
    if any(x in text for x in ["supplier", "vendor"]):
        return "supplier_alert"
    if status == "resolved":
        return "resolution_notice"
    return "email_alert"

def should_keep_email(sender: str, subject: str) -> bool:
    sender_l = safe_lower(sender)
    subject_l = safe_lower(subject)

    sender_match = (not WATCH_SENDERS) or any(x in sender_l for x in WATCH_SENDERS)
    subject_match = any(k in subject_l for k in WATCH_SUBJECT_KEYWORDS)

    return sender_match or subject_match

# CODEX CHANGE: Use Message directly for the same reason as above.
def email_to_pal_event(msg: Message, raw_body: str) -> Dict[str, Any]:
    subject = decode_mime_words(msg.get("Subject", ""))
    sender = decode_mime_words(msg.get("From", ""))
    msg_id = decode_mime_words(msg.get("Message-ID", "")) or f"no_msgid_{int(time.time()*1000)}"
    date_raw = decode_mime_words(msg.get("Date", ""))

    status = infer_status(subject, raw_body)
    priority = infer_priority(status, subject, raw_body)
    entity_id = extract_entity_id(subject, raw_body)
    location = extract_location(subject, raw_body)
    event_type = infer_event_type(status, subject, raw_body)

    return {
        "source": "gmail",
        "source_message_id": msg_id,
        "timestamp": utc_now_iso(),
        "event_time_raw": date_raw,
        "entity": entity_id,
        "event_type": event_type,
        "status": status,
        "priority": priority,
        "location": location,
        "subject": subject,
        "sender": sender,
        "note": raw_body[:500],
    }

# --------------------------------------------------
# 5. IMAP fetch
# --------------------------------------------------

def connect_imap() -> imaplib.IMAP4_SSL:
    email_addr = os.getenv("PAL_EMAIL_ADDRESS", "").strip()
    email_pw = os.getenv("PAL_EMAIL_APP_PASSWORD", "").strip()

    if not email_addr or not email_pw:
        print("ERROR: Missing PAL_EMAIL_ADDRESS or PAL_EMAIL_APP_PASSWORD")
        sys.exit(1)

    mail = imaplib.IMAP4_SSL(DEFAULT_IMAP_SERVER)
    mail.login(email_addr, email_pw)
    return mail

# CODEX CHANGE: Use Message directly for the return type annotation.
def fetch_recent_emails(max_results: int = DEFAULT_MAX_RESULTS, label: str = DEFAULT_LABEL) -> List[Message]:
    mail = connect_imap()
    try:
        status, _ = mail.select(label)
        if status != "OK":
            print(f"ERROR: could not open mailbox {label}")
            return []

        # Pull latest emails. Keep the demo simple.
        status, data = mail.search(None, "ALL")
        if status != "OK" or not data or not data[0]:
            return []

        ids = data[0].split()
        ids = ids[-max_results:]

        msgs = []
        for email_id in ids:
            status, msg_data = mail.fetch(email_id, "(RFC822)")
            if status != "OK":
                continue
            for response_part in msg_data:
                if isinstance(response_part, tuple):
                    msg = email.message_from_bytes(response_part[1])
                    msgs.append(msg)
        return msgs
    finally:
        try:
            mail.close()
        except Exception:
            pass
        mail.logout()

# --------------------------------------------------
# 6. Commands
# --------------------------------------------------

def cmd_reset() -> None:
    save_json(DB_FILE, {"events": []})
    save_json(STATE_FILE, {"seen_email_ids": []})
    print(json.dumps({"ok": True, "reset": True}, indent=2))

def cmd_show() -> None:
    db = ensure_db()
    print(json.dumps(db, indent=2, ensure_ascii=False))

def cmd_fetch(max_results: int, label: str) -> None:
    db = ensure_db()
    state = ensure_state()
    seen = set(state["seen_email_ids"])

    msgs = fetch_recent_emails(max_results=max_results, label=label)

    added = []
    skipped = 0

    for msg in msgs:
        subject = decode_mime_words(msg.get("Subject", ""))
        sender = decode_mime_words(msg.get("From", ""))
        msg_id = decode_mime_words(msg.get("Message-ID", ""))

        if not msg_id:
            skipped += 1
            continue

        if msg_id in seen:
            skipped += 1
            continue

        if not should_keep_email(sender, subject):
            skipped += 1
            continue

        body = extract_text_from_message(msg)
        event = email_to_pal_event(msg, body)

        db["events"].append(event)
        seen.add(msg_id)
        added.append(event)

    state["seen_email_ids"] = sorted(seen)
    save_json(DB_FILE, db)
    save_json(STATE_FILE, state)

    print(json.dumps({
        "ok": True,
        "fetched_messages": len(msgs),
        "added_events": len(added),
        "skipped": skipped,
        "events": added,
    }, indent=2, ensure_ascii=False))

def cmd_loop(interval_sec: int, max_results: int, label: str) -> None:
    print(f"[pal_core_07] polling gmail every {interval_sec}s...")
    while True:
        try:
            cmd_fetch(max_results=max_results, label=label)
        except KeyboardInterrupt:
            print("\n[pal_core_07] stopped")
            break
        except Exception as e:
            print(json.dumps({"ok": False, "error": str(e)}, indent=2))
        time.sleep(interval_sec)

# --------------------------------------------------
# 7. Main
# --------------------------------------------------

def main():
    parser = argparse.ArgumentParser(description="PAL Core 07 - Gmail alerts as sensor stream")
    sub = parser.add_subparsers(dest="cmd", required=True)

    sub.add_parser("reset")
    sub.add_parser("show")

    p_fetch = sub.add_parser("fetch")
    p_fetch.add_argument("--max_results", type=int, default=DEFAULT_MAX_RESULTS)
    p_fetch.add_argument("--label", type=str, default=DEFAULT_LABEL)

    p_loop = sub.add_parser("loop")
    p_loop.add_argument("--interval", type=int, default=60)
    p_loop.add_argument("--max_results", type=int, default=DEFAULT_MAX_RESULTS)
    p_loop.add_argument("--label", type=str, default=DEFAULT_LABEL)

    args = parser.parse_args()

    if args.cmd == "reset":
        cmd_reset()
    elif args.cmd == "show":
        cmd_show()
    elif args.cmd == "fetch":
        cmd_fetch(max_results=args.max_results, label=args.label)
    elif args.cmd == "loop":
        cmd_loop(interval_sec=args.interval, max_results=args.max_results, label=args.label)

if __name__ == "__main__":
    main()
$ python pal_core_07_gmail_alerts.py fetch
{
  "ok": true,
  "fetched_messages": 10,
  "added_events": 10,
  "skipped": 0,
  "events": [
    {
      "source": "gmail",
      "source_message_id":
..........................

$ python pal_core_07_gmail_alerts.py show
{
  "events": [
    {
      "source": "gmail",
      "source_message_id":
.............................


$ python pal_core_07_gmail_alerts.py loop --interval 10
[pal_core_07] polling gmail every 10s...
{
.........................................
{
  "ok": true,
  "fetched_messages": 10,
  "added_events": 1,
  "skipped": 9,
  "events": [
    {
      "source": "gmail",
      "source_message_id": "<,,,,,,,,,,,,,,,,,,,,@mail.gmail.com>",     
      "timestamp": "2026-04-23T17:02:24.902957+00:00",
      "event_time_raw": "Thu, 23 Apr 2026 13:02:02 -0400",
      "entity": "email_signal",
      "event_type": "email_alert",
      "status": "alert",
      "priority": 4,
      "location": "unknown",
      "subject": "test1",
      "sender": "T T <terrytaylorbonn@gmail.com>",
      "note": "111111111111"
    }
  ]
}


26.0618