redmine/redmine_ticket_drafts.py

#!/usr/bin/env python3
"""Local draft mutation queue for internal notes vs customer emails.

CRITICAL DISTINCTION (do not collapse these):

- internal_note  → Redmine journal note only. NEVER emails the customer.
                   Optional private_notes. PUT /issues/:id.json

- customer_email → Helpdesk "Send note" path. ALWAYS emails the customer
                   (when applied). POST /helpdesk/email_note
                   Cannot be private.

Drafts are stored under <cache-dir>/drafts/*.json until applied or cancelled.
"""

from __future__ import annotations

import json
import re
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import urlparse


KIND_INTERNAL_NOTE = "internal_note"
KIND_CUSTOMER_EMAIL = "customer_email"
KIND_FIELDS = "fields"
KIND_CONTACT_FIELDS = "contact_fields"
# Legacy aliases from earlier draft CLI
LEGACY_KIND_MAP = {
    "note": KIND_INTERNAL_NOTE,
    "helpdesk_reply": KIND_CUSTOMER_EMAIL,
}

# Hosts allowed for write/apply without --allow-prod-writes
DEFAULT_SAFE_WRITE_HOST_PATTERNS = (
    r"^localhost$",
    r"^127\.0\.0\.1$",
    r"^192\.168\.\d+\.\d+$",
    r"^10\.\d+\.\d+\.\d+$",
    r"^retired\.int\.ldrprep\.com$",
    r"^.*\.170$",  # common LAN Redmine
)


def now_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def normalize_kind(kind: str) -> str:
    k = (kind or "").strip()
    k = LEGACY_KIND_MAP.get(k, k)
    if k not in {KIND_INTERNAL_NOTE, KIND_CUSTOMER_EMAIL, KIND_FIELDS, KIND_CONTACT_FIELDS}:
        raise ValueError(
            f"Unsupported draft kind {kind!r}. "
            f"Use {KIND_INTERNAL_NOTE!r}, {KIND_CUSTOMER_EMAIL!r}, "
            f"{KIND_FIELDS!r}, or {KIND_CONTACT_FIELDS!r}."
        )
    return k


def kind_label(kind: str) -> str:
    kind = normalize_kind(kind)
    if kind == KIND_INTERNAL_NOTE:
        return "INTERNAL NOTE (does NOT email customer)"
    if kind == KIND_CUSTOMER_EMAIL:
        return "CUSTOMER EMAIL (WILL email customer via Helpdesk)"
    if kind == KIND_FIELDS:
        return "FIELD UPDATE (does NOT email customer; changes issue fields)"
    if kind == KIND_CONTACT_FIELDS:
        return "CONTACT UPDATE (does NOT email; changes CRM contact fields/tags)"
    return kind


def drafts_dir(cache_dir: Path) -> Path:
    path = Path(cache_dir) / "drafts"
    path.mkdir(parents=True, exist_ok=True)
    return path


def assert_kind_flags(kind: str, *, private: bool) -> str:
    kind = normalize_kind(kind)
    if kind == KIND_CUSTOMER_EMAIL and private:
        raise ValueError(
            "customer_email drafts cannot be private. "
            "Private notes never email; customer emails always email. "
            "Use kind=internal_note with --private for staff-only notes."
        )
    return kind


def add_draft(
    cache_dir: Path,
    *,
    issue_id: int | None = None,
    contact_id: int | None = None,
    kind: str,
    body: str = "",
    private: bool = False,
    fields: dict[str, Any] | None = None,
    project: str | None = None,
    base_url: str | None = None,
) -> Path:
    kind = assert_kind_flags(kind, private=private)
    body = (body or "").strip()
    if kind in {KIND_INTERNAL_NOTE, KIND_CUSTOMER_EMAIL} and not body:
        raise ValueError("Draft body is required for notes/emails.")
    if kind == KIND_FIELDS and not fields:
        raise ValueError("Field-update drafts require a non-empty fields payload.")
    if kind == KIND_CONTACT_FIELDS:
        if contact_id is None:
            raise ValueError("contact_fields drafts require contact_id.")
        if not fields:
            raise ValueError("contact_fields drafts require a non-empty fields payload.")
    if kind in {KIND_INTERNAL_NOTE, KIND_CUSTOMER_EMAIL, KIND_FIELDS} and issue_id is None:
        raise ValueError(f"{kind} drafts require issue_id.")
    draft_id = uuid.uuid4().hex[:12]
    payload = {
        "id": draft_id,
        "created_at": now_iso(),
        "status": "pending",
        "project": project,
        "base_url": base_url,
        "issue_id": int(issue_id) if issue_id is not None else None,
        "contact_id": int(contact_id) if contact_id is not None else None,
        "kind": kind,
        "kind_label": kind_label(kind),
        "body": body,
        "private": bool(private) if kind == KIND_INTERNAL_NOTE else False,
        "fields": fields or {},
        "applied_at": None,
        "error": None,
        "emails_customer": kind == KIND_CUSTOMER_EMAIL,
    }
    path = drafts_dir(cache_dir) / f"{draft_id}.json"
    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return path


def load_draft(cache_dir: Path, draft_id: str) -> tuple[Path, dict[str, Any]]:
    path = drafts_dir(cache_dir) / f"{draft_id}.json"
    if not path.is_file():
        matches = list(drafts_dir(cache_dir).glob(f"{draft_id}*.json"))
        if len(matches) != 1:
            raise FileNotFoundError(f"Draft not found or ambiguous: {draft_id}")
        path = matches[0]
    data = json.loads(path.read_text(encoding="utf-8"))
    data["kind"] = normalize_kind(str(data.get("kind") or ""))
    return path, data


def list_drafts(cache_dir: Path, status: str | None = "pending") -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    root = drafts_dir(cache_dir)
    for path in sorted(root.glob("*.json")):
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            continue
        if status and data.get("status") != status:
            continue
        try:
            data["kind"] = normalize_kind(str(data.get("kind") or ""))
        except ValueError:
            pass
        data["kind_label"] = kind_label(str(data.get("kind") or "internal_note")) if data.get("kind") else ""
        data["_path"] = str(path)
        items.append(data)
    return items


def cancel_draft(cache_dir: Path, draft_id: str) -> bool:
    try:
        path, data = load_draft(cache_dir, draft_id)
    except FileNotFoundError:
        return False
    data["status"] = "cancelled"
    data["cancelled_at"] = now_iso()
    path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return True


def cancel_all_pending(cache_dir: Path) -> int:
    count = 0
    for draft in list_drafts(cache_dir, status="pending"):
        if cancel_draft(cache_dir, str(draft["id"])):
            count += 1
    return count


def mark_draft_applied(cache_dir: Path, draft_id: str, *, result: dict[str, Any] | None = None) -> None:
    path, data = load_draft(cache_dir, draft_id)
    data["status"] = "applied"
    data["applied_at"] = now_iso()
    data["error"] = None
    data["apply_result"] = result or {}
    path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")


def mark_draft_error(cache_dir: Path, draft_id: str, error: str) -> None:
    path, data = load_draft(cache_dir, draft_id)
    data["status"] = "error"
    data["error"] = error
    data["error_at"] = now_iso()
    path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")


def host_is_safe_for_writes(base_url: str, patterns: tuple[str, ...] = DEFAULT_SAFE_WRITE_HOST_PATTERNS) -> bool:
    host = (urlparse(base_url).hostname or "").lower()
    if not host:
        return False
    return any(re.search(pat, host) for pat in patterns)


def assert_write_target_allowed(base_url: str, *, allow_prod_writes: bool) -> None:
    if allow_prod_writes:
        return
    if host_is_safe_for_writes(base_url):
        return
    raise RuntimeError(
        f"Refusing writes to {base_url!r}. "
        "Default allow-list is LAN/test hosts only. "
        "Re-run against the test Redmine, or pass --allow-prod-writes if you truly intend production."
    )