openclaw/workspace/scripts/gmail_label_classify.py

#!/usr/bin/env python3
"""Apply Gmail labels for owner mailbox automation.

Currently classifies self-sent briefing mails into existing/created Gmail labels
such as "브리핑" and "주식". Designed for OpenClaw cron at 01:00 Asia/Seoul.
"""
from __future__ import annotations

import argparse
import json
import subprocess
import sys
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Iterable

ACCOUNT = "mini.snowoyh@gmail.com"
STATE_DIR = Path(__file__).resolve().parents[1] / "state"
LOG_PATH = STATE_DIR / "gmail_label_classify.log"


@dataclass(frozen=True)
class Rule:
    name: str
    label: str
    query: str
    remove_labels: tuple[str, ...] = ()


@dataclass(frozen=True)
class CleanupRule:
    name: str
    query: str


RULES = [
    Rule(
        name="briefing",
        label="브리핑",
        query='newer_than:7d subject:("[오전 브리핑]" OR "[오후 브리핑]")',
    ),
    Rule(
        name="stock_analysis",
        label="종목분석",
        query='newer_than:30d subject:"[비하이브 종목분석]"',
        remove_labels=("주식",),
    ),
    Rule(
        name="stock_briefing",
        label="주식브리핑",
        query='newer_than:30d subject:"[주식 리포트]"',
        remove_labels=("주식",),
    ),
]

CLEANUP_RULES = [
    CleanupRule(
        name="test_mail_24h",
        query=(
            'older_than:1d -in:trash '
            'from:mini.snowoyh@gmail.com to:mini.snowoyh@gmail.com '
            '(subject:테스트 OR subject:test OR subject:TEST)'
        ),
    ),
]


def run_gog(args: list[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
    cmd = ["gog", "gmail", *args, "--account", ACCOUNT, "--no-input"]
    return subprocess.run(cmd, text=True, capture_output=True, check=check)


def list_labels() -> set[str]:
    cp = run_gog(["labels", "list", "--json"])
    data = json.loads(cp.stdout or "{}")
    labels = data.get("labels") or data.get("Labels") or []
    names: set[str] = set()
    for item in labels:
        if isinstance(item, dict):
            name = item.get("name") or item.get("Name")
            if name:
                names.add(name)
    if not names:
        # Fallback for older gog JSON shapes: parse plain output.
        cp = run_gog(["labels", "list", "--plain"])
        for line in cp.stdout.splitlines()[1:]:
            parts = line.split("\t")
            if len(parts) >= 2:
                names.add(parts[1])
    return names


def ensure_label(label: str, dry_run: bool) -> None:
    if label in list_labels():
        return
    if dry_run:
        print(f"[dry-run] create label: {label}")
        return
    run_gog(["labels", "create", label])


def search_threads(query: str) -> list[dict]:
    cp = run_gog(["search", query, "--max", "100", "--json"])
    data = json.loads(cp.stdout or "{}")
    return data.get("threads") or []


def has_label(thread: dict, label: str) -> bool:
    labels = thread.get("labels") or []
    return label in labels


def apply_label(thread_id: str, label: str, dry_run: bool) -> None:
    if dry_run:
        print(f"[dry-run] add {label} to thread {thread_id}")
        return
    run_gog(["labels", "modify", thread_id, "--add", label])


def remove_label(thread_id: str, label: str, dry_run: bool) -> None:
    if dry_run:
        print(f"[dry-run] remove {label} from thread {thread_id}")
        return
    run_gog(["labels", "modify", thread_id, "--remove", label])


def log(line: str) -> None:
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    now = datetime.now().isoformat(timespec="seconds")
    with LOG_PATH.open("a", encoding="utf-8") as f:
        f.write(f"{now} {line}\n")


def cleanup_messages(dry_run: bool = False) -> tuple[int, int]:
    total_seen = 0
    total_trashed = 0
    for rule in CLEANUP_RULES:
        threads = search_threads(rule.query)
        total_seen += len(threads)
        if dry_run:
            for thread in threads:
                print(f"[dry-run] trash {rule.name}: {thread.get('id')} {thread.get('subject')}")
        elif threads:
            run_gog(["trash", "--query", rule.query, "--max", "100"])
        total_trashed += len(threads)
        print(f"{rule.name}: seen={len(threads)} trashed={0 if dry_run else len(threads)}")
    return total_seen, total_trashed


def classify(dry_run: bool = False) -> int:
    total_seen = 0
    total_changed = 0
    for rule in RULES:
        ensure_label(rule.label, dry_run)
        threads = search_threads(rule.query)
        total_seen += len(threads)
        changed = 0
        for thread in threads:
            thread_id = thread.get("id")
            if not thread_id:
                continue
            if not has_label(thread, rule.label):
                apply_label(thread_id, rule.label, dry_run)
                changed += 1
            for old_label in rule.remove_labels:
                if has_label(thread, old_label):
                    remove_label(thread_id, old_label, dry_run)
                    changed += 1
        total_changed += changed
        print(f"{rule.name}: seen={len(threads)} changed={changed}")

    cleanup_seen, cleanup_trashed = cleanup_messages(dry_run=dry_run)
    status = "dry-run" if dry_run else "ok"
    summary = (
        f"status={status} seen={total_seen} labeled={total_changed} "
        f"cleanup_seen={cleanup_seen} cleanup_trashed={0 if dry_run else cleanup_trashed}"
    )
    print(summary)
    if not dry_run:
        log(summary)
    return 0


def main(argv: Iterable[str] | None = None) -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args(argv)
    try:
        return classify(dry_run=args.dry_run)
    except subprocess.CalledProcessError as e:
        print(e.stdout, end="", file=sys.stderr)
        print(e.stderr, end="", file=sys.stderr)
        return e.returncode or 1


if __name__ == "__main__":
    raise SystemExit(main())