ia_dev/gitea-issues/mail_common.py

# Shared config and helpers for gitea-issues mail scripts (IMAP/SMTP, Gitea).
# Used by mail-list-unread, mail-send-reply, mail-create-issue-from-email, mail-mark-read.

from __future__ import annotations

import json
import os
import re
import ssl
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

# Only consider messages on or after this date (IMAP format DD-Mon-YYYY). Override with env MAIL_SINCE_DATE.
MAIL_SINCE_DATE_DEFAULT = "10-Mar-2026"


def imap_since_date() -> str:
    """Return IMAP SINCE date (DD-Mon-YYYY). Messages before this date are ignored by fetch/list scripts."""
    return os.environ.get("MAIL_SINCE_DATE", MAIL_SINCE_DATE_DEFAULT).strip() or MAIL_SINCE_DATE_DEFAULT


def imap_search_criterion_all() -> str:
    """IMAP search: all messages on or after MAIL_SINCE_DATE."""
    return f"SINCE {imap_since_date()}"


def imap_search_criterion_unseen() -> str:
    """IMAP search: unread messages on or after MAIL_SINCE_DATE."""
    return f"(UNSEEN SINCE {imap_since_date()})"


def repo_root() -> Path:
    # Root = directory containing gitea-issues (ia_dev). .secrets and logs live under ia_dev (./.secrets, ./logs)
    # so they do not depend on the parent project; same ia_dev works for any project (../ai_project_id).
    env_root = os.environ.get("REPO_ROOT")
    if env_root:
        return Path(env_root).resolve()
    issues_dir = os.environ.get("GITEA_ISSUES_DIR")
    if issues_dir:
        return Path(issues_dir).resolve().parent
    return Path(__file__).resolve().parent.parent


def load_env_file(path: Path) -> None:
    if not path.is_file():
        return
    with open(path, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            if "=" in line:
                key, _, value = line.partition("=")
                key = key.strip()
                value = value.strip().strip("'\"")
                if key and key not in os.environ:
                    os.environ[key] = value


def load_imap_config() -> dict[str, str]:
    root = repo_root()
    env_path = root / ".secrets" / "gitea-issues" / "imap-bridge.env"
    load_env_file(env_path)
    ssl_verify_raw = os.environ.get("IMAP_SSL_VERIFY", "true").lower()
    ssl_verify = ssl_verify_raw not in ("0", "false", "no")
    return {
        "host": os.environ.get("IMAP_HOST", "127.0.0.1"),
        "port": os.environ.get("IMAP_PORT", "1143"),
        "user": os.environ.get("IMAP_USER", ""),
        "password": os.environ.get("IMAP_PASSWORD", ""),
        "use_starttls": os.environ.get("IMAP_USE_STARTTLS", "true").lower() in ("1", "true", "yes"),
        "ssl_verify": ssl_verify,
        "filter_to": os.environ.get("MAIL_FILTER_TO", "ai.support.lecoffreio@4nkweb.com").strip().lower(),
    }


def imap_ssl_context(ssl_verify: bool = True) -> ssl.SSLContext:
    """Return SSL context for IMAP STARTTLS. Use ssl_verify=False only for local Bridge with self-signed cert."""
    if ssl_verify:
        return ssl.create_default_context()
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return ctx


def load_smtp_config() -> dict[str, str]:
    root = repo_root()
    env_path = root / ".secrets" / "gitea-issues" / "imap-bridge.env"
    load_env_file(env_path)
    ssl_verify_raw = os.environ.get("IMAP_SSL_VERIFY", os.environ.get("SMTP_SSL_VERIFY", "true")).lower()
    ssl_verify = ssl_verify_raw not in ("0", "false", "no")
    return {
        "host": os.environ.get("SMTP_HOST", os.environ.get("IMAP_HOST", "127.0.0.1")),
        "port": os.environ.get("SMTP_PORT", "1025"),
        "user": os.environ.get("SMTP_USER", os.environ.get("IMAP_USER", "")),
        "password": os.environ.get("SMTP_PASSWORD", os.environ.get("IMAP_PASSWORD", "")),
        "use_starttls": os.environ.get("SMTP_USE_STARTTLS", "true").lower() in ("1", "true", "yes"),
        "ssl_verify": ssl_verify,
    }


def load_gitea_config() -> dict[str, str]:
    root = repo_root()
    token = os.environ.get("GITEA_TOKEN")
    if not token:
        token_path = root / ".secrets" / "gitea-issues" / "token"
        if token_path.is_file():
            token = token_path.read_text(encoding="utf-8").strip()
    return {
        "api_url": os.environ.get("GITEA_API_URL", "https://git.4nkweb.com/api/v1").rstrip("/"),
        "owner": os.environ.get("GITEA_REPO_OWNER", "4nk"),
        "repo": os.environ.get("GITEA_REPO_NAME", "lecoffre_ng"),
        "token": token or "",
    }


def sanitize_title(raw: str, max_len: int = 200) -> str:
    one_line = re.sub(r"\s+", " ", raw).strip()
    return one_line[:max_len] if one_line else "(no subject)"


def create_gitea_issue(title: str, body: str) -> dict | None:
    gitea = load_gitea_config()
    if not gitea["token"]:
        return None
    url = f"{gitea['api_url']}/repos/{gitea['owner']}/{gitea['repo']}/issues"
    payload = json.dumps({"title": title, "body": body}).encode("utf-8")
    req = Request(
        url,
        data=payload,
        method="POST",
        headers={
            "Accept": "application/json",
            "Content-Type": "application/json",
            "Authorization": f"token {gitea['token']}",
        },
    )
    try:
        with urlopen(req, timeout=30) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except (HTTPError, URLError):
        return None