smart_ide/services/ia_dev/git-issues/mail-to-issue.py

#!/usr/bin/env python3
"""
Create Gitea issues from unread emails via IMAP (e.g. Proton Mail Bridge).

**Preferred flow (agent-driven):** do not chain directly. Use mail-list-unread.sh
to list unread emails, then for each: formalize the issue or send a reply (mail-send-reply.sh);
only when a correction/evolution is ready, create the issue (mail-create-issue-from-email.sh
with optional formalized title/body), treat it (fix/evol), then comment on the issue and
reply to the email via the Bridge.

This script (mail-to-issue) is a **batch** fallback: it creates one issue per unread
message with title=subject and body=text+From, then marks messages as read. Use only
when the agent-driven flow is not used.

Reads IMAP config from .secrets/git-issues/imap-bridge.env (or env vars).
Reads Gitea token from GITEA_TOKEN or .secrets/git-issues/token.
"""

from __future__ import annotations

import email
import imaplib
import ssl
import sys
from email.header import decode_header
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from mail_common import (
    create_gitea_issue,
    imap_search_criterion_unseen,
    load_gitea_config,
    load_imap_config,
    repo_root,
    sanitize_title,
)


def _decode_header_value(header: str | None) -> str:
    if not header:
        return ""
    parts = decode_header(header)
    result = []
    for part, charset in parts:
        if isinstance(part, bytes):
            result.append(part.decode(charset or "utf-8", errors="replace"))
        else:
            result.append(part)
    return "".join(result)


def _get_text_body(msg: email.message.Message) -> str:
    if msg.is_multipart():
        for part in msg.walk():
            if part.get_content_type() == "text/plain":
                payload = part.get_payload(decode=True)
                if payload:
                    return payload.decode(part.get_content_charset() or "utf-8", errors="replace")
        return ""
    payload = msg.get_payload(decode=True)
    if not payload:
        return ""
    return payload.decode(msg.get_content_charset() or "utf-8", errors="replace")


def main() -> None:
    imap_cfg = load_imap_config()
    if not imap_cfg["user"] or not imap_cfg["password"]:
        root = repo_root()
        env_path = root / ".secrets" / "git-issues" / "imap-bridge.env"
        print("[git-issues] ERROR: IMAP_USER and IMAP_PASSWORD required.", file=sys.stderr)
        sys.exit(1)
    gitea_cfg = load_gitea_config()
    if not gitea_cfg["token"]:
        print("[git-issues] ERROR: GITEA_TOKEN not set.", file=sys.stderr)
        sys.exit(1)

    mail = imaplib.IMAP4(imap_cfg["host"], int(imap_cfg["port"]))
    if imap_cfg["use_starttls"]:
        mail.starttls(ssl.create_default_context())
    mail.login(imap_cfg["user"], imap_cfg["password"])
    mail.select("INBOX")
    criterion = imap_search_criterion_unseen()
    _, nums = mail.search(None, criterion)
    ids = nums[0].split()
    if not ids:
        print("[git-issues] No unread messages.")
        mail.logout()
        return

    created = 0
    for uid in ids:
        uid_s = uid.decode("ascii")
        _, data = mail.fetch(uid, "(RFC822)")
        if not data or not data[0]:
            continue
        msg = email.message_from_bytes(data[0][1])
        subject = _decode_header_value(msg.get("Subject"))
        from_ = _decode_header_value(msg.get("From"))
        body_text = _get_text_body(msg)
        body_for_issue = f"**From:** {from_}\n\n{body_text}".strip()
        title = sanitize_title(subject)
        issue = create_gitea_issue(title, body_for_issue)
        if issue:
            created += 1
            print(f"[git-issues] Created issue #{issue.get('number', '?')}: {title[:60]}")
            mail.store(uid_s, "+FLAGS", "\\Seen")
        else:
            print(f"[git-issues] Skipped (API failed): {title[:60]}", file=sys.stderr)

    mail.logout()
    print(f"[git-issues] Done. Created {created} issue(s).")


if __name__ == "__main__":
    main()