chatgpt-register/scripts/setup_cfmail.py

from __future__ import annotations

import argparse
import json
import os
from dataclasses import dataclass
from pathlib import Path
import secrets
import shutil
import subprocess
import sys
from typing import Any

from core.cfmail_provisioner import CfmailProvisioner, ProvisioningSettings

DEFAULT_WORKER_REPO = "https://github.com/dreamhunter2333/cloudflare_temp_email.git"
DEFAULT_WORKER_NAME = "zhuce6-cfmail"
DEFAULT_D1_NAME = "zhuce6-cfmail-db"
DEFAULT_VENDOR_DIR = Path("vendor")
DEFAULT_WORKER_DIR = DEFAULT_VENDOR_DIR / "cfmail-worker"
DEFAULT_CONFIG_DIR = Path("config")
DEFAULT_CFMAIL_ACCOUNTS_PATH = DEFAULT_CONFIG_DIR / "cfmail_accounts.json"
DEFAULT_CFMAIL_ENV_PATH = DEFAULT_CONFIG_DIR / "cfmail_provision.env"
DEFAULT_COMPATIBILITY_DATE = "2025-04-01"
EMAIL_ROUTING_FALLBACK_MX_RECORDS = (
    ("amir.mx.cloudflare.net", 13),
    ("isaac.mx.cloudflare.net", 24),
    ("linda.mx.cloudflare.net", 86),
)
EMAIL_ROUTING_FALLBACK_SPF = "v=spf1 include:_spf.mx.cloudflare.net ~all"
RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}


class SetupError(RuntimeError):
    def __init__(self, message: str, *, hint: str = "") -> None:
        super().__init__(message)
        self.hint = hint


@dataclass(frozen=True)
class WorkerLayout:
    repo_dir: Path
    worker_dir: Path
    schema_path: Path
    migration_paths: tuple[Path, ...]
    wrangler_template_path: Path | None


@dataclass(frozen=True)
class DNSRecordSpec:
    record_type: str
    name: str
    content: str
    priority: int | None = None
    ttl: int = 1
    proxied: bool | None = None


@dataclass(frozen=True)
class CfmailRuntimeConfig:
    api_token: str
    account_id: str
    zone_id: str
    worker_name: str
    worker_domain: str
    zone_name: str
    email_domain: str
    admin_password: str
    d1_name: str
    d1_database_id: str


class CloudflareClient:
    def __init__(
        self,
        api_token: str,
        *,
        auth_email: str = "",
        auth_key: str = "",
        timeout: float = 30.0,
    ) -> None:
        api_token = str(api_token or "").strip()
        auth_email = str(auth_email or "").strip()
        auth_key = str(auth_key or "").strip()
        if not api_token and not (auth_email and auth_key):
            raise SetupError(
                "缺少 Cloudflare 凭据。",
                hint="请提供 Cloudflare API Token, 或提供 CF_AUTH_EMAIL + CF_AUTH_KEY。",
            )
        try:
            import httpx  # type: ignore
        except ImportError as exc:
            raise SetupError(
                "当前 Python 环境缺少 httpx。",
                hint="请先执行 `uv sync` 或 `uv pip install httpx`。",
            ) from exc

        self._httpx = httpx
        headers = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": "zhuce6/setup_cfmail",
        }
        self._uses_api_token = bool(api_token)
        if api_token:
            headers["Authorization"] = f"Bearer {api_token}"
        else:
            headers["X-Auth-Email"] = auth_email
            headers["X-Auth-Key"] = auth_key
        self._client = httpx.Client(
            base_url="https://api.cloudflare.com/client/v4",
            timeout=timeout,
            trust_env=True,
            headers=headers,
        )

    def close(self) -> None:
        self._client.close()

    def __enter__(self) -> "CloudflareClient":
        return self

    def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
        self.close()

    def request(
        self,
        method: str,
        path: str,
        *,
        params: dict[str, Any] | None = None,
        json_body: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        last_error: Exception | None = None
        for attempt in range(1, 4):
            try:
                response = self._client.request(method, path, params=params, json=json_body)
            except self._httpx.HTTPError as exc:
                last_error = exc
                if attempt < 3:
                    continue
                raise SetupError(
                    f"Cloudflare API 请求失败: {method.upper()} {path}",
                    hint=f"请检查网络连通性后重试。原始错误: {exc}",
                ) from exc
            if response.status_code in RETRYABLE_STATUS_CODES and attempt < 3:
                continue
            try:
                payload = response.json()
            except ValueError as exc:
                raise SetupError(
                    f"Cloudflare API 返回了无法解析的 JSON: {method.upper()} {path}",
                    hint=f"HTTP {response.status_code}, 响应片段: {response.text[:300]}",
                ) from exc
            if response.is_success and payload.get("success") is True:
                return payload

            errors = payload.get("errors") or []
            message = "; ".join(
                str(item.get("message") or item.get("code") or item)
                for item in errors
                if item
            ).strip()
            if not message:
                message = response.text[:300].strip() or f"HTTP {response.status_code}"
            raise SetupError(
                f"Cloudflare API 调用失败: {method.upper()} {path} -> {message}",
                hint=self._build_api_hint(path, response.status_code),
            )
        if last_error is not None:
            raise SetupError(str(last_error))
        raise SetupError(f"Cloudflare API 调用失败: {method.upper()} {path}")

    def _build_api_hint(self, path: str, status_code: int) -> str:
        if status_code in {401, 403}:
            return (
                "请确认 API Token 具备 Zone Read, DNS Edit, Workers Scripts Write, D1 Edit, "
                "以及 Email Routing 写权限。"
            )
        if "/email/routing" in path:
            return "请先在 Cloudflare Dashboard 手动开启 Email Routing, 然后重新执行脚本。"
        return "请根据 Cloudflare 返回信息检查配置后重试。"

    def verify_token(self) -> dict[str, Any]:
        if self._uses_api_token:
            payload = self.request("GET", "/user/tokens/verify")
            result = payload.get("result")
            if not isinstance(result, dict):
                raise SetupError("Token 校验响应缺少 result 字段。")
            return result

        payload = self.request("GET", "/user")
        result = payload.get("result")
        if not isinstance(result, dict):
            raise SetupError("Cloudflare 用户校验响应缺少 result 字段。")
        if not result.get("status"):
            result = {**result, "status": "active"}
        return result

    def resolve_zone(self, zone_name: str) -> dict[str, Any]:
        payload = self.request("GET", "/zones", params={"name": zone_name})
        result = payload.get("result") or []
        matches = [item for item in result if isinstance(item, dict) and item.get("name") == zone_name]
        if not matches:
            raise SetupError(
                f"未找到 zone: {zone_name}",
                hint="请确认该域名已接入当前 Cloudflare 账号, 且 API Token 有 Zone Read 权限。",
            )
        zone = matches[0]
        account = zone.get("account") if isinstance(zone.get("account"), dict) else {}
        account_id = str(account.get("id") or "").strip()
        zone_id = str(zone.get("id") or "").strip()
        if not account_id or not zone_id:
            raise SetupError("Zone 信息中缺少 account_id 或 zone_id。")
        return zone

    def list_d1_databases(self, account_id: str, *, database_name: str = "") -> list[dict[str, Any]]:
        payload = self.request("GET", f"/accounts/{account_id}/d1/database")
        result = payload.get("result") or []
        items = [item for item in result if isinstance(item, dict)]
        if database_name:
            items = [item for item in items if str(item.get("name") or "") == database_name]
        return items

    def ensure_d1_database(self, account_id: str, database_name: str) -> dict[str, Any]:
        existing = self.list_d1_databases(account_id, database_name=database_name)
        if existing:
            return existing[0]
        payload = self.request(
            "POST",
            f"/accounts/{account_id}/d1/database",
            json_body={"name": database_name},
        )
        result = payload.get("result")
        if not isinstance(result, dict):
            raise SetupError("创建 D1 数据库成功, 但响应缺少 result。")
        return result

    def get_workers_subdomain(self, account_id: str) -> str:
        payload = self.request("GET", f"/accounts/{account_id}/workers/subdomain")
        result = payload.get("result")
        if not isinstance(result, dict):
            return ""
        return str(result.get("subdomain") or "").strip()

    def get_email_routing_status(self, zone_id: str) -> dict[str, Any]:
        payload = self.request("GET", f"/zones/{zone_id}/email/routing")
        result = payload.get("result")
        if not isinstance(result, dict):
            raise SetupError("Email Routing 状态接口返回格式异常。")
        return result

    def get_email_routing_dns_requirements(self, zone_id: str) -> list[DNSRecordSpec]:
        payload = self.request("GET", f"/zones/{zone_id}/email/routing/dns")
        result = payload.get("result")
        items: list[dict[str, Any]] = []
        if isinstance(result, list):
            items = [item for item in result if isinstance(item, dict)]
        elif isinstance(result, dict):
            for key in ("records", "items", "dns_records", "dns"):
                value = result.get(key)
                if isinstance(value, list):
                    items = [item for item in value if isinstance(item, dict)]
                    break
        records: list[DNSRecordSpec] = []
        for item in items:
            record_type = str(item.get("type") or item.get("record_type") or "").upper()
            name = str(item.get("name") or item.get("hostname") or "").strip()
            content = str(item.get("content") or item.get("value") or "").strip()
            if not record_type or not name or not content:
                continue
            priority = item.get("priority")
            try:
                parsed_priority = int(priority) if priority is not None else None
            except (TypeError, ValueError):
                parsed_priority = None
            records.append(
                DNSRecordSpec(
                    record_type=record_type,
                    name=name,
                    content=content,
                    priority=parsed_priority,
                    ttl=int(item.get("ttl") or 1),
                    proxied=item.get("proxied") if isinstance(item.get("proxied"), bool) else None,
                )
            )
        return records

    def list_dns_records(self, zone_id: str, *, name: str = "", record_type: str = "") -> list[dict[str, Any]]:
        params: dict[str, Any] = {}
        if name:
            params["name"] = name
        if record_type:
            params["type"] = record_type
        payload = self.request("GET", f"/zones/{zone_id}/dns_records", params=params)
        result = payload.get("result") or []
        return [item for item in result if isinstance(item, dict)]

    def ensure_dns_record(self, zone_id: str, spec: DNSRecordSpec) -> dict[str, Any]:
        existing = self.list_dns_records(zone_id, name=spec.name, record_type=spec.record_type)
        for record in existing:
            if self._dns_record_matches(record, spec):
                return record

        updatable = self._select_updatable_dns_record(existing, spec)
        payload = {
            "type": spec.record_type,
            "name": spec.name,
            "content": spec.content,
            "ttl": spec.ttl,
        }
        if spec.priority is not None:
            payload["priority"] = spec.priority
        if spec.proxied is not None and spec.record_type not in {"MX", "TXT"}:
            payload["proxied"] = spec.proxied

        if updatable is not None:
            response = self.request(
                "PUT",
                f"/zones/{zone_id}/dns_records/{updatable['id']}",
                json_body=payload,
            )
        else:
            response = self.request("POST", f"/zones/{zone_id}/dns_records", json_body=payload)
        result = response.get("result")
        if not isinstance(result, dict):
            raise SetupError(f"DNS 记录写入成功, 但响应格式异常: {spec.record_type} {spec.name}")
        return result

    def _dns_record_matches(self, record: dict[str, Any], spec: DNSRecordSpec) -> bool:
        if str(record.get("type") or "").upper() != spec.record_type:
            return False
        if str(record.get("name") or "").strip().lower() != spec.name.lower():
            return False
        if str(record.get("content") or "").strip().lower() != spec.content.lower():
            return False
        if spec.priority is not None and int(record.get("priority") or 0) != spec.priority:
            return False
        return True

    def _select_updatable_dns_record(self, existing: list[dict[str, Any]], spec: DNSRecordSpec) -> dict[str, Any] | None:
        if spec.record_type == "TXT":
            spf_like = [
                record for record in existing
                if str(record.get("content") or "").strip().lower().startswith("v=spf1")
            ]
            if len(spf_like) == 1:
                return spf_like[0]
            return None
        for record in existing:
            if str(record.get("content") or "").strip().lower() == spec.content.lower():
                return record
        return None

    def get_catch_all_rule(self, zone_id: str) -> dict[str, Any] | None:
        try:
            payload = self.request("GET", f"/zones/{zone_id}/email/routing/rules/catch_all")
        except SetupError as exc:
            if "not found" in str(exc).lower():
                return None
            raise
        result = payload.get("result")
        return result if isinstance(result, dict) else None

    def ensure_catch_all_worker(self, zone_id: str, worker_name: str) -> dict[str, Any]:
        current = self.get_catch_all_rule(zone_id) or {}
        desired_actions = [{"type": "worker", "value": [worker_name]}]
        desired_matchers = [{"type": "all"}]
        if (
            bool(current.get("enabled", True))
            and self._normalize_actions(current.get("actions")) == desired_actions
            and self._normalize_matchers(current.get("matchers")) == desired_matchers
        ):
            return current
        payload = {
            "enabled": True,
            "name": str(current.get("name") or f"{worker_name} catch-all"),
            "matchers": desired_matchers,
            "actions": desired_actions,
        }
        response = self.request(
            "PUT",
            f"/zones/{zone_id}/email/routing/rules/catch_all",
            json_body=payload,
        )
        result = response.get("result")
        if not isinstance(result, dict):
            raise SetupError("Catch-all 规则更新成功, 但响应格式异常。")
        return result

    def _normalize_actions(self, actions: Any) -> list[dict[str, Any]]:
        normalized: list[dict[str, Any]] = []
        if not isinstance(actions, list):
            return normalized
        for item in actions:
            if not isinstance(item, dict):
                continue
            values = item.get("value")
            if isinstance(values, list):
                value_list = [str(v) for v in values]
            elif values is None:
                value_list = []
            else:
                value_list = [str(values)]
            normalized.append({"type": str(item.get("type") or ""), "value": value_list})
        return normalized

    def _normalize_matchers(self, matchers: Any) -> list[dict[str, Any]]:
        normalized: list[dict[str, Any]] = []
        if not isinstance(matchers, list):
            return normalized
        for item in matchers:
            if not isinstance(item, dict):
                continue
            normalized.append({"type": str(item.get("type") or "")})
        return normalized


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="一键部署 cfmail Worker 并生成 zhuce6 配置。")
    parser.add_argument("--api-token", default="", help="Cloudflare API Token")
    parser.add_argument("--auth-email", default="", help="Cloudflare 认证邮箱, 与 --auth-key 成对使用")
    parser.add_argument("--auth-key", default="", help="Cloudflare Global API Key, 与 --auth-email 成对使用")
    parser.add_argument("--zone-name", required=True, help="Cloudflare Zone 名称, 例如 example.com")
    parser.add_argument("--worker-name", default=DEFAULT_WORKER_NAME, help=f"Worker 名称, 默认 {DEFAULT_WORKER_NAME}")
    parser.add_argument("--d1-name", default=DEFAULT_D1_NAME, help=f"D1 数据库名称, 默认 {DEFAULT_D1_NAME}")
    parser.add_argument("--mail-domain", help="邮箱域名, 默认等于 --zone-name")
    parser.add_argument(
        "--skip-clone",
        action="store_true",
        help="若 vendor/cfmail-worker 已存在, 跳过 clone 并直接复用现有目录",
    )
    return parser


def print_step(number: int, total: int, title: str) -> None:
    print(f"[{number}/{total}] {title}")


def ensure_command(name: str, *, install_hint: str) -> None:
    if shutil.which(name):
        return
    raise SetupError(f"缺少必要命令: {name}", hint=install_hint)


def ensure_required_tools() -> None:
    ensure_command("git", install_hint="请先安装 git, 然后重新执行脚本。")
    ensure_command("node", install_hint="请先安装 Node.js 18+。")
    ensure_command("npm", install_hint="请先安装 npm。")
    ensure_command("npx", install_hint="请先安装 npm, 确保 npx 可用。")


def ensure_mail_domain(zone_name: str, mail_domain: str) -> str:
    zone_name = str(zone_name or "").strip().lower()
    mail_domain = str(mail_domain or zone_name).strip().lower()
    if not mail_domain:
        raise SetupError("mail_domain 不能为空。")
    if mail_domain != zone_name and not mail_domain.endswith(f".{zone_name}"):
        raise SetupError(
            f"mail_domain 必须等于 zone_name 或属于其子域: {mail_domain}",
            hint=f"当前 zone_name 为 {zone_name}, 请改用 {zone_name} 或其子域。",
        )
    return mail_domain


def clone_worker_source(target_dir: Path, *, skip_clone: bool) -> Path:
    target_dir.parent.mkdir(parents=True, exist_ok=True)
    if target_dir.exists():
        if not (target_dir / ".git").exists():
            raise SetupError(
                f"目标目录已存在但不是 git 仓库: {target_dir}",
                hint="请删除该目录后重试, 或改用干净的 vendor/cfmail-worker 路径。",
            )
        if skip_clone:
            print(f"    复用现有源码目录: {target_dir}")
            return target_dir
        print(f"    检测到现有源码目录, 直接复用: {target_dir}")
        return target_dir
    run_command(
        ["git", "clone", "--depth", "1", DEFAULT_WORKER_REPO, str(target_dir)],
        cwd=Path.cwd(),
        step="clone Worker 源码",
    )
    return target_dir


def resolve_worker_layout(repo_dir: Path) -> WorkerLayout:
    worker_dir = repo_dir / "worker"
    if not worker_dir.exists():
        raise SetupError(
            f"上游源码缺少 worker 目录: {worker_dir}",
            hint="请检查上游仓库结构是否变化。",
        )
    wrangler_template_path: Path | None = None
    for candidate in (worker_dir / "wrangler.toml.template", worker_dir / "wrangler.toml"):
        if candidate.exists():
            wrangler_template_path = candidate
            break

    schema_candidates = (
        repo_dir / "db" / "schema.sql",
        worker_dir / "schema.sql",
    )
    schema_path = next((path for path in schema_candidates if path.exists()), None)
    if schema_path is None:
        raise SetupError("未找到 schema.sql。", hint="请检查上游仓库结构是否变化。")

    migration_paths: list[Path] = []
    migration_dirs = (repo_dir / "db", worker_dir / "migrations")
    for directory in migration_dirs:
        if not directory.exists():
            continue
        for path in sorted(directory.glob("*.sql")):
            if path.resolve() == schema_path.resolve():
                continue
            migration_paths.append(path)
        if migration_paths:
            break

    return WorkerLayout(
        repo_dir=repo_dir,
        worker_dir=worker_dir,
        schema_path=schema_path,
        migration_paths=tuple(migration_paths),
        wrangler_template_path=wrangler_template_path,
    )


def read_wrangler_template_defaults(template_path: Path | None) -> dict[str, Any]:
    defaults = {
        "main": "src/worker.ts",
        "compatibility_date": DEFAULT_COMPATIBILITY_DATE,
        "compatibility_flags": ["nodejs_compat"],
        "keep_vars": True,
    }
    if template_path is None or not template_path.exists():
        return defaults
    try:
        import tomllib

        parsed = tomllib.loads(template_path.read_text(encoding="utf-8"))
    except Exception:
        return defaults
    defaults["main"] = str(parsed.get("main") or defaults["main"])
    defaults["compatibility_date"] = str(parsed.get("compatibility_date") or defaults["compatibility_date"])
    flags = parsed.get("compatibility_flags")
    if isinstance(flags, list) and flags:
        defaults["compatibility_flags"] = [str(item) for item in flags]
    defaults["keep_vars"] = bool(parsed.get("keep_vars", defaults["keep_vars"]))
    return defaults


def toml_string(value: str) -> str:
    return json.dumps(str(value), ensure_ascii=False)


def toml_array(values: list[str]) -> str:
    return json.dumps([str(value) for value in values], ensure_ascii=False)


def write_worker_wrangler(
    *,
    worker_dir: Path,
    worker_name: str,
    account_id: str,
    database_id: str,
    database_name: str,
    email_domain: str,
    admin_password: str,
    jwt_secret: str,
    compatibility_date: str = DEFAULT_COMPATIBILITY_DATE,
    main: str = "src/worker.ts",
    compatibility_flags: list[str] | None = None,
    keep_vars: bool = True,
) -> Path:
    wrangler_path = worker_dir / "wrangler.toml"
    flags = compatibility_flags or ["nodejs_compat"]
    content = "\n".join(
        [
            f"name = {toml_string(worker_name)}",
            f"account_id = {toml_string(account_id)}",
            f"main = {toml_string(main)}",
            f"compatibility_date = {toml_string(compatibility_date)}",
            f"compatibility_flags = {toml_array(flags)}",
            f"keep_vars = {'true' if keep_vars else 'false'}",
            "",
            "[vars]",
            f"PREFIX = {toml_string('tmp')}",
            f"DEFAULT_DOMAINS = {toml_array([email_domain])}",
            f"DOMAINS = {toml_array([email_domain])}",
            f"ADMIN_PASSWORDS = {toml_array([admin_password])}",
            f"JWT_SECRET = {toml_string(jwt_secret)}",
            "ENABLE_USER_CREATE_EMAIL = true",
            "ENABLE_USER_DELETE_EMAIL = true",
            "ENABLE_AUTO_REPLY = false",
            "",
            "[[d1_databases]]",
            f"binding = {toml_string('DB')}",
            f"database_name = {toml_string(database_name)}",
            f"database_id = {toml_string(database_id)}",
            "",
        ]
    )
    wrangler_path.write_text(content, encoding="utf-8")
    return wrangler_path


def render_cfmail_accounts_payload(
    *,
    worker_domain: str,
    email_domain: str,
    worker_name: str,
    admin_password: str,
) -> list[dict[str, Any]]:
    return [
        {
            "name": worker_name,
            "worker_domain": worker_domain,
            "email_domain": email_domain,
            "admin_password": admin_password,
            "enabled": True,
        }
    ]


def write_cfmail_accounts_json(
    output_path: Path,
    *,
    worker_domain: str,
    email_domain: str,
    worker_name: str,
    admin_password: str,
) -> Path:
    payload = render_cfmail_accounts_payload(
        worker_domain=worker_domain,
        email_domain=email_domain,
        worker_name=worker_name,
        admin_password=admin_password,
    )
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return output_path


def shell_quote(value: str) -> str:
    return str(value).replace("\\", "\\\\").replace('"', '\\"')


def write_cfmail_provision_env(
    output_path: Path,
    *,
    api_token: str = "",
    auth_email: str = "",
    auth_key: str = "",
    account_id: str,
    zone_id: str,
    worker_name: str,
    zone_name: str,
    d1_database_id: str = "",
) -> Path:
    content = "\n".join(
        [
            f'export ZHUCE6_CFMAIL_API_TOKEN="{shell_quote(api_token)}"',
            f'export ZHUCE6_CFMAIL_CF_AUTH_EMAIL="{shell_quote(auth_email)}"',
            f'export ZHUCE6_CFMAIL_CF_AUTH_KEY="{shell_quote(auth_key)}"',
            f'export ZHUCE6_CFMAIL_CF_ACCOUNT_ID="{shell_quote(account_id)}"',
            f'export ZHUCE6_CFMAIL_CF_ZONE_ID="{shell_quote(zone_id)}"',
            f'export ZHUCE6_CFMAIL_WORKER_NAME="{shell_quote(worker_name)}"',
            f'export ZHUCE6_CFMAIL_ZONE_NAME="{shell_quote(zone_name)}"',
            f'export ZHUCE6_D1_DATABASE_ID="{shell_quote(d1_database_id)}"',
            "",
        ]
    )
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(content, encoding="utf-8")
    return output_path


def prepare_runtime_cfmail_config(
    *,
    api_token: str,
    auth_email: str = "",
    auth_key: str = "",
    worker_domain: str | None = None,
    zone_name: str,
    worker_name: str = DEFAULT_WORKER_NAME,
    d1_name: str = DEFAULT_D1_NAME,
    mail_domain: str | None = None,
    admin_password: str | None = None,
    accounts_path: Path = DEFAULT_CFMAIL_ACCOUNTS_PATH,
    provision_env_path: Path = DEFAULT_CFMAIL_ENV_PATH,
) -> CfmailRuntimeConfig:
    normalized_zone_name = str(zone_name or "").strip().lower()
    if not normalized_zone_name:
        raise SetupError("zone_name 不能为空。")
    normalized_mail_domain = ensure_mail_domain(normalized_zone_name, mail_domain or normalized_zone_name)
    normalized_worker_name = str(worker_name or DEFAULT_WORKER_NAME).strip() or DEFAULT_WORKER_NAME
    normalized_worker_domain = str(worker_domain or "").strip().lower()
    resolved_admin_password = str(admin_password or "").strip() or secrets.token_urlsafe(24)

    with CloudflareClient(api_token, auth_email=auth_email, auth_key=auth_key) as client:
        client.verify_token()
        zone = client.resolve_zone(normalized_zone_name)
        zone_id = str(zone.get("id") or "").strip()
        account = zone.get("account") if isinstance(zone.get("account"), dict) else {}
        account_id = str(account.get("id") or "").strip()
        database = client.ensure_d1_database(account_id, d1_name)
        d1_database_id = str(database.get("uuid") or database.get("id") or "").strip()
        if not normalized_worker_domain:
            worker_subdomain = client.get_workers_subdomain(account_id)
            normalized_worker_domain = build_worker_domain(normalized_worker_name, worker_subdomain)

    write_cfmail_accounts_json(
        accounts_path,
        worker_domain=normalized_worker_domain,
        email_domain=normalized_mail_domain,
        worker_name=normalized_worker_name,
        admin_password=resolved_admin_password,
    )
    write_cfmail_provision_env(
        provision_env_path,
        api_token=api_token,
        auth_email=auth_email,
        auth_key=auth_key,
        account_id=account_id,
        zone_id=zone_id,
        worker_name=normalized_worker_name,
        zone_name=normalized_zone_name,
        d1_database_id=d1_database_id,
    )
    if normalized_worker_domain:
        provisioner = CfmailProvisioner(
            config_path=accounts_path,
            settings=ProvisioningSettings(
                auth_email=auth_email,
                auth_key=auth_key,
                account_id=account_id,
                zone_id=zone_id,
                worker_name=normalized_worker_name,
                zone_name=normalized_zone_name,
            ),
        )
        try:
            provisioner.smoke_test(normalized_worker_domain, resolved_admin_password, normalized_mail_domain)
        except Exception as exc:
            error_text = str(exc)
            if "无效的域名" not in error_text and "invalid" not in error_text.lower():
                raise
            rotation = provisioner.rotate_active_domain()
            if not rotation.success or not rotation.new_domain:
                raise SetupError(
                    "cfmail 域名校验失败, 且自动轮换未成功。",
                    hint=rotation.error or error_text or "请确认 worker_domain 与 zone_name 配置正确。",
                )
            normalized_mail_domain = rotation.new_domain

    return CfmailRuntimeConfig(
        api_token=api_token,
        account_id=account_id,
        zone_id=zone_id,
        worker_name=normalized_worker_name,
        worker_domain=normalized_worker_domain,
        zone_name=normalized_zone_name,
        email_domain=normalized_mail_domain,
        admin_password=resolved_admin_password,
        d1_name=d1_name,
        d1_database_id=d1_database_id,
    )


def run_command(
    args: list[str],
    *,
    cwd: Path,
    step: str,
    env: dict[str, str] | None = None,
    input_text: str | None = None,
) -> subprocess.CompletedProcess[str]:
    merged_env = os.environ.copy()
    if env:
        merged_env.update(env)
    process = subprocess.run(
        args,
        cwd=str(cwd),
        env=merged_env,
        input=input_text,
        text=True,
        capture_output=True,
        check=False,
    )
    if process.returncode != 0:
        detail = (process.stderr or process.stdout or "").strip()
        raise SetupError(
            f"{step} 失败: {' '.join(args)}",
            hint=detail[:1200] or "请根据命令输出检查后重试。",
        )
    return process


def is_benign_migration_error(detail: str) -> bool:
    normalized = str(detail or "").lower()
    markers = (
        "duplicate column name",
        "already exists",
        "no such table",
        "duplicate index name",
    )
    return any(marker in normalized for marker in markers)


def run_wrangler_sql_file(
    *,
    database_name: str,
    sql_path: Path,
    cwd: Path,
    env: dict[str, str],
    step: str,
    tolerate_already_applied: bool = False,
) -> subprocess.CompletedProcess[str]:
    try:
        return run_command(
            ["npx", "wrangler", "d1", "execute", database_name, "--remote", "--file", str(sql_path)],
            cwd=cwd,
            env=env,
            step=step,
        )
    except SetupError as exc:
        if tolerate_already_applied and is_benign_migration_error(exc.hint):
            print(f"    跳过已存在或历史补丁不再适用的 migration: {sql_path.name}")
            return subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr=exc.hint)
        raise


def build_wrangler_env(api_token: str) -> dict[str, str]:
    return {
        "CLOUDFLARE_API_TOKEN": api_token,
        "CF_API_TOKEN": api_token,
        "CI": "1",
        "NO_UPDATE_NOTIFIER": "1",
        "npm_config_update_notifier": "false",
    }


def ensure_email_routing_dns(client: CloudflareClient, zone_id: str, mail_domain: str) -> list[DNSRecordSpec]:
    specs = client.get_email_routing_dns_requirements(zone_id)
    filtered = [spec for spec in specs if spec.record_type in {"MX", "TXT"}]
    if not filtered:
        filtered = [
            DNSRecordSpec("MX", mail_domain, host, priority=priority)
            for host, priority in EMAIL_ROUTING_FALLBACK_MX_RECORDS
        ]
        filtered.append(DNSRecordSpec("TXT", mail_domain, EMAIL_ROUTING_FALLBACK_SPF))

    normalized: list[DNSRecordSpec] = []
    for spec in filtered:
        normalized_name = spec.name.rstrip(".")
        if normalized_name == "@":
            normalized_name = mail_domain
        if not normalized_name:
            normalized_name = mail_domain
        normalized.append(
            DNSRecordSpec(
                record_type=spec.record_type,
                name=normalized_name,
                content=spec.content.rstrip("."),
                priority=spec.priority,
                ttl=spec.ttl,
                proxied=spec.proxied,
            )
        )
    for spec in normalized:
        client.ensure_dns_record(zone_id, spec)
    return normalized


def email_routing_enabled(status: dict[str, Any]) -> bool:
    for key in ("enabled", "active"):
        value = status.get(key)
        if isinstance(value, bool):
            return value
    state = str(status.get("status") or status.get("state") or "").strip().lower()
    return state in {"active", "enabled", "ready", "verified", "success"}


def build_worker_domain(worker_name: str, account_subdomain: str) -> str:
    worker_name = str(worker_name or "").strip()
    account_subdomain = str(account_subdomain or "").strip()
    if not worker_name or not account_subdomain:
        raise SetupError(
            "无法推导 workers.dev 域名。",
            hint="请确认账号已启用 workers.dev 子域, 或在 Cloudflare Dashboard 中先完成一次 Worker 初始化。",
        )
    return f"{worker_name}.{account_subdomain}.workers.dev"


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)

    zone_name = str(args.zone_name).strip().lower()
    worker_name = str(args.worker_name).strip()
    d1_name = str(args.d1_name).strip()
    mail_domain = ensure_mail_domain(zone_name, args.mail_domain or zone_name)
    api_token = str(args.api_token).strip()
    auth_email = str(args.auth_email).strip()
    auth_key = str(args.auth_key).strip()
    if not api_token and not (auth_email and auth_key):
        raise SystemExit("ERROR: 请提供 --api-token, 或同时提供 --auth-email 和 --auth-key。")
    vendor_dir = Path.cwd() / DEFAULT_WORKER_DIR
    cfmail_accounts_path = Path.cwd() / DEFAULT_CFMAIL_ACCOUNTS_PATH
    cfmail_env_path = Path.cwd() / DEFAULT_CFMAIL_ENV_PATH

    total_steps = 13
    try:
        print_step(1, total_steps, "检查本地依赖")
        ensure_required_tools()

        with CloudflareClient(api_token, auth_email=auth_email, auth_key=auth_key) as client:
            print_step(2, total_steps, "验证 Cloudflare 凭据")
            token_info = client.verify_token()
            print(f"    token_status={token_info.get('status', 'unknown')}")

            print_step(3, total_steps, "解析 zone/account 信息")
            zone = client.resolve_zone(zone_name)
            zone_id = str(zone.get("id") or "").strip()
            account = zone.get("account") if isinstance(zone.get("account"), dict) else {}
            account_id = str(account.get("id") or "").strip()
            print(f"    account_id={account_id}")
            print(f"    zone_id={zone_id}")

            print_step(4, total_steps, "准备 Worker 源码")
            repo_dir = clone_worker_source(vendor_dir, skip_clone=bool(args.skip_clone))
            layout = resolve_worker_layout(repo_dir)
            print(f"    worker_dir={layout.worker_dir}")

            print_step(5, total_steps, "创建或复用 D1 数据库")
            database = client.ensure_d1_database(account_id, d1_name)
            database_id = str(database.get("uuid") or database.get("id") or "").strip()
            if not database_id:
                raise SetupError("D1 数据库响应缺少 database_id。")
            print(f"    database_id={database_id}")

            print_step(6, total_steps, "写入 wrangler.toml")
            defaults = read_wrangler_template_defaults(layout.wrangler_template_path)
            admin_password = secrets.token_urlsafe(24)
            jwt_secret = secrets.token_urlsafe(48)
            wrangler_path = write_worker_wrangler(
                worker_dir=layout.worker_dir,
                worker_name=worker_name,
                account_id=account_id,
                database_id=database_id,
                database_name=d1_name,
                email_domain=mail_domain,
                admin_password=admin_password,
                jwt_secret=jwt_secret,
                compatibility_date=str(defaults.get("compatibility_date") or DEFAULT_COMPATIBILITY_DATE),
                main=str(defaults.get("main") or "src/worker.ts"),
                compatibility_flags=list(defaults.get("compatibility_flags") or ["nodejs_compat"]),
                keep_vars=bool(defaults.get("keep_vars", True)),
            )
            print(f"    wrote {wrangler_path}")

            wrangler_env = build_wrangler_env(api_token)

            print_step(7, total_steps, "安装 Worker 依赖")
            run_command(["npm", "install", "--no-fund", "--no-audit"], cwd=layout.worker_dir, step="npm install")

            print_step(8, total_steps, "执行 D1 schema 与 migration")
            run_wrangler_sql_file(
                database_name=d1_name,
                sql_path=layout.schema_path,
                cwd=layout.worker_dir,
                env=wrangler_env,
                step="执行 schema.sql",
            )
            for migration_path in layout.migration_paths:
                run_wrangler_sql_file(
                    database_name=d1_name,
                    sql_path=migration_path,
                    cwd=layout.worker_dir,
                    env=wrangler_env,
                    step=f"执行 migration {migration_path.name}",
                    tolerate_already_applied=True,
                )

            print_step(9, total_steps, "部署 Worker")
            run_command(
                ["npx", "wrangler", "deploy", "--minify"],
                cwd=layout.worker_dir,
                env=wrangler_env,
                step="wrangler deploy",
            )

            print_step(10, total_steps, "配置 Email Routing 所需 DNS")
            dns_specs = ensure_email_routing_dns(client, zone_id, mail_domain)
            print(f"    ensured_records={len(dns_specs)}")

            print_step(11, total_steps, "检查 Email Routing 状态并配置 catch-all")
            routing_status = client.get_email_routing_status(zone_id)
            if not email_routing_enabled(routing_status):
                raise SetupError(
                    "当前 Zone 尚未开启 Email Routing。",
                    hint=(
                        "请先在 Cloudflare Dashboard > Email > Email Routing 中完成启用, "
                        "确认状态变为 active/enabled 后重新执行脚本。"
                    ),
                )
            client.ensure_catch_all_worker(zone_id, worker_name)

            print_step(12, total_steps, "生成 config/cfmail_accounts.json")
            account_subdomain = client.get_workers_subdomain(account_id)
            worker_domain = build_worker_domain(worker_name, account_subdomain)
            write_cfmail_accounts_json(
                cfmail_accounts_path,
                worker_domain=worker_domain,
                email_domain=mail_domain,
                worker_name=worker_name,
                admin_password=admin_password,
            )
            print(f"    wrote {cfmail_accounts_path}")

            print_step(13, total_steps, "生成 config/cfmail_provision.env")
            write_cfmail_provision_env(
                cfmail_env_path,
                api_token=api_token,
                auth_email=auth_email,
                auth_key=auth_key,
                account_id=account_id,
                zone_id=zone_id,
                worker_name=worker_name,
                zone_name=zone_name,
                d1_database_id=database_id,
            )
            print(f"    wrote {cfmail_env_path}")

        print("部署完成。")
        return 0
    except SetupError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        if exc.hint:
            print(f"HINT: {exc.hint}", file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        print("ERROR: 用户中断执行。", file=sys.stderr)
        return 130


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