acme-auto/backend/routers/admin.py

from fastapi import APIRouter, Depends, HTTPException, Header, Cookie, Request
from fastapi.responses import JSONResponse, PlainTextResponse
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from sqlalchemy.orm import selectinload
from pydantic import BaseModel
from datetime import datetime, timedelta
from pathlib import Path
import hashlib
import hmac
import json
import logging
import time

logger = logging.getLogger(__name__)

from backend.database import get_db, db_add, db_update, db_delete, db_flush
from backend.models import Server, Domain, DeployLog, AcmeConfig, AcmeLog
from backend.config import get_settings

router = APIRouter(tags=["admin-api"])


# ──────────────── Auth ────────────────

def generate_token(username: str, secret_key: str) -> str:
    """生成简单的认证 token"""
    expire = int(time.time()) + 86400 * 7  # 7 天过期
    payload = f"{username}:{expire}"
    signature = hmac.new(secret_key.encode(), payload.encode(), hashlib.sha256).hexdigest()[:16]
    return f"{payload}:{signature}"


def verify_token(token: str, secret_key: str) -> bool:
    """验证 token"""
    try:
        parts = token.split(":")
        if len(parts) != 3:
            return False
        username, expire_str, signature = parts
        expire = int(expire_str)
        if time.time() > expire:
            return False
        payload = f"{username}:{expire_str}"
        expected = hmac.new(secret_key.encode(), payload.encode(), hashlib.sha256).hexdigest()[:16]
        return hmac.compare_digest(signature, expected)
    except Exception:
        return False


async def require_auth(authorization: str = Header(None), admin_token: str = Cookie(None)):
    """认证依赖，检查 Header 或 Cookie 中的 token"""
    settings = get_settings()

    # 从 Header 获取
    token = None
    if authorization and authorization.startswith("Bearer "):
        token = authorization[7:]
    # 从 Cookie 获取
    elif admin_token:
        token = admin_token

    if not token or not verify_token(token, settings.secret_key):
        raise HTTPException(status_code=401, detail="未登录或登录已过期")
    return True


class LoginRequest(BaseModel):
    username: str
    password: str


@router.post("/login")
async def login(data: LoginRequest):
    settings = get_settings()
    if data.username != settings.admin_username or data.password != settings.admin_password:
        raise HTTPException(status_code=401, detail="用户名或密码错误")

    token = generate_token(data.username, settings.secret_key)
    response = JSONResponse(content={"ok": True, "token": token})
    response.set_cookie(
        key="admin_token",
        value=token,
        max_age=86400 * 7,
        httponly=True,
        samesite="lax",
    )
    return response


@router.post("/logout")
async def logout():
    response = JSONResponse(content={"ok": True})
    response.delete_cookie("admin_token")
    return response


@router.get("/me")
async def check_auth(_: bool = Depends(require_auth)):
    return {"ok": True}


# ──────────────── Schemas ────────────────

class ServerCreate(BaseModel):
    name: str
    platform: str = "linux"
    token: str
    ip: str | None = None


class ServerUpdate(BaseModel):
    name: str | None = None
    platform: str | None = None
    token: str | None = None
    ip: str | None = None


class DomainCreate(BaseModel):
    server_id: int
    domain: str
    acme_config_id: int | None = None
    cert_dir: str | None = None
    check_cmd: str = "nginx -t"
    reload_cmd: str = "systemctl reload nginx"


class DomainUpdate(BaseModel):
    server_id: int | None = None
    acme_config_id: int | None = None
    domain: str | None = None
    cert_dir: str | None = None
    check_cmd: str | None = None
    reload_cmd: str | None = None
    version: str | None = None


# ──────────────── Stats ────────────────

@router.get("/stats")
async def get_stats(db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    """仪表盘统计数据"""
    server_count = (await db.execute(select(func.count(Server.id)))).scalar() or 0
    domain_count = (await db.execute(select(func.count(Domain.id)))) .scalar() or 0

    # 即将过期（7 天内）
    now = datetime.utcnow()
    soon = now.replace(day=now.day + 7) if now.day <= 24 else now.replace(month=now.month + 1, day=now.day + 7 - 30)
    expiring = (await db.execute(
        select(func.count(Domain.id)).where(
            Domain.cert_not_after.isnot(None),
            Domain.cert_not_after <= soon,
        )
    )).scalar() or 0

    # 最近日志
    result = await db.execute(
        select(DeployLog)
        .options(selectinload(DeployLog.domain), selectinload(DeployLog.domain).selectinload(Domain.server))
        .order_by(DeployLog.created_at.desc())
        .limit(10)
    )
    logs = result.scalars().all()

    return {
        "server_count": server_count,
        "domain_count": domain_count,
        "expiring_count": expiring,
        "recent_logs": [
            {
                "id": log.id,
                "domain": log.domain.domain if log.domain else None,
                "server": log.domain.server.name if log.domain and log.domain.server else None,
                "status": log.status,
                "message": log.message,
                "created_at": log.created_at.isoformat(),
            }
            for log in logs
        ],
    }


# ──────────────── Servers ────────────────

@router.get("/servers")
async def list_servers(db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    result = await db.execute(select(Server).options(selectinload(Server.domains)).order_by(Server.id))
    servers = result.scalars().all()
    return [
        {
            "id": s.id,
            "name": s.name,
            "platform": s.platform,
            "token": s.token,
            "ip": s.ip,
            "domain_count": len(s.domains),
            "created_at": s.created_at.isoformat(),
        }
        for s in servers
    ]


@router.get("/servers/{server_id}")
async def get_server(server_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    result = await db.execute(select(Server).where(Server.id == server_id))
    s = result.scalar_one_or_none()
    if not s:
        raise HTTPException(404, "Server not found")
    return {"id": s.id, "name": s.name, "platform": s.platform, "token": s.token, "ip": s.ip}


@router.post("/servers", status_code=201)
async def create_server(data: ServerCreate, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    server = Server(**data.model_dump())
    await db_add(db, server, refresh=True)
    return {"id": server.id, "name": server.name}


@router.put("/servers/{server_id}")
async def update_server(server_id: int, data: ServerUpdate, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    result = await db.execute(select(Server).where(Server.id == server_id))
    server = result.scalar_one_or_none()
    if not server:
        raise HTTPException(404, "Server not found")
    await db_update(db, server, data.model_dump(exclude_unset=True))
    return {"id": server.id}


@router.delete("/servers/{server_id}")
async def delete_server(server_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    result = await db.execute(select(Server).where(Server.id == server_id))
    server = result.scalar_one_or_none()
    if not server:
        raise HTTPException(404, "Server not found")
    await db_delete(db, server)
    return {"ok": True}


# ──────────────── Domains ────────────────

@router.get("/domains")
async def list_domains(db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    result = await db.execute(
        select(Domain).options(selectinload(Domain.server), selectinload(Domain.acme_config)).order_by(Domain.id)
    )
    domains = result.scalars().all()

    # 查询每个域名的最新部署日志
    deploy_status = {}
    log_result = await db.execute(
        select(DeployLog).order_by(DeployLog.id.desc())
    )
    for log in log_result.scalars().all():
        if log.domain_id not in deploy_status:
            deploy_status[log.domain_id] = log.status == "success"

    return [
        {
            "id": d.id,
            "server_id": d.server_id,
            "server_name": d.server.name if d.server else None,
            "acme_config_id": d.acme_config_id,
            "acme_config_name": d.acme_config.name if d.acme_config else None,
            "domain": d.domain,
            "cert_dir": d.cert_dir,
            "check_cmd": d.check_cmd,
            "reload_cmd": d.reload_cmd,
            "version": d.version,
            "cert_not_after": d.cert_not_after.isoformat() if d.cert_not_after else None,
            "deploy_success": deploy_status.get(d.id),
        }
        for d in domains
    ]


@router.get("/domains/{domain_id}")
async def get_domain(domain_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    result = await db.execute(select(Domain).where(Domain.id == domain_id))
    d = result.scalar_one_or_none()
    if not d:
        raise HTTPException(404, "Domain not found")
    return {
        "id": d.id,
        "server_id": d.server_id,
        "acme_config_id": d.acme_config_id,
        "domain": d.domain,
        "cert_dir": d.cert_dir,
        "check_cmd": d.check_cmd,
        "reload_cmd": d.reload_cmd,
        "version": d.version,
    }


@router.post("/domains", status_code=201)
async def create_domain(data: DomainCreate, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    dump = data.model_dump()
    if not dump.get("cert_dir"):
        dump["cert_dir"] = dump["domain"].lstrip("*.")

    # 校验：同域名 + 同服务器 + 同 ACME 配置不允许重复
    conditions = [Domain.domain == dump["domain"], Domain.server_id == dump["server_id"]]
    if dump.get("acme_config_id"):
        conditions.append(Domain.acme_config_id == dump["acme_config_id"])
    else:
        conditions.append(Domain.acme_config_id.is_(None))
    existing = (await db.execute(select(Domain).where(*conditions))).scalars().first()
    if existing:
        raise HTTPException(409, "该域名在此服务器下已存在相同 ACME 配置的记录")

    domain = Domain(**dump)
    await db_add(db, domain, refresh=True)
    return {"id": domain.id, "domain": domain.domain}


@router.put("/domains/{domain_id}")
async def update_domain(domain_id: int, data: DomainUpdate, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    result = await db.execute(select(Domain).where(Domain.id == domain_id))
    domain = result.scalar_one_or_none()
    if not domain:
        raise HTTPException(404, "Domain not found")

    update_fields = data.model_dump(exclude_unset=True)

    # 如果改了 server_id 或 acme_config_id，校验是否与已有记录冲突
    new_server_id = update_fields.get("server_id", domain.server_id)
    new_acme_config_id = update_fields.get("acme_config_id", domain.acme_config_id)
    if "server_id" in update_fields or "acme_config_id" in update_fields:
        conditions = [
            Domain.domain == domain.domain,
            Domain.server_id == new_server_id,
            Domain.id != domain_id,
        ]
        if new_acme_config_id:
            conditions.append(Domain.acme_config_id == new_acme_config_id)
        else:
            conditions.append(Domain.acme_config_id.is_(None))
        dup = (await db.execute(select(Domain).where(*conditions))).scalars().first()
        if dup:
            raise HTTPException(409, "该域名在此服务器下已存在相同 ACME 配置的记录")

    await db_update(db, domain, update_fields)
    return {"id": domain.id}


@router.delete("/domains/{domain_id}")
async def delete_domain(domain_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    result = await db.execute(select(Domain).where(Domain.id == domain_id))
    domain = result.scalar_one_or_none()
    if not domain:
        raise HTTPException(404, "Domain not found")
    await db_delete(db, domain)
    return {"ok": True}


# ──────────────── Logs ────────────────

@router.get("/logs")
async def list_logs(status: str | None = None, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    query = (
        select(DeployLog)
        .options(selectinload(DeployLog.domain), selectinload(DeployLog.domain).selectinload(Domain.server))
        .order_by(DeployLog.created_at.desc())
        .limit(100)
    )
    if status:
        query = query.where(DeployLog.status == status)
    result = await db.execute(query)
    logs = result.scalars().all()
    return [
        {
            "id": log.id,
            "domain": log.domain.domain if log.domain else None,
            "server": log.domain.server.name if log.domain and log.domain.server else None,
            "hostname": log.hostname,
            "status": log.status,
            "message": log.message,
            "created_at": log.created_at.isoformat(),
        }
        for log in logs
    ]


# ──────────────── ACME 配置（多配置）────────────────

class AcmeConfigCreate(BaseModel):
    name: str = "默认配置"
    acme_server: str = "https://acme-v02.api.letsencrypt.org/directory"
    email: str = ""
    dns_provider: str = "aliyun"
    dns_credentials: str = "{}"
    renew_days: int = 30


class AcmeConfigUpdate(BaseModel):
    name: str | None = None
    acme_server: str | None = None
    email: str | None = None
    dns_provider: str | None = None
    dns_credentials: str | None = None
    renew_days: int | None = None


def _serialize_acme_config(c: AcmeConfig) -> dict:
    return {
        "id": c.id,
        "name": c.name,
        "acme_server": c.acme_server,
        "email": c.email,
        "dns_provider": c.dns_provider,
        "dns_credentials": c.dns_credentials,
        "renew_days": c.renew_days,
        "has_account_key": c.account_key is not None,
        "domain_count": len(c.domains) if c.domains else 0,
    }


@router.get("/acme/configs")
async def list_acme_configs(db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    result = await db.execute(select(AcmeConfig).options(selectinload(AcmeConfig.domains)).order_by(AcmeConfig.id))
    configs = result.scalars().all()
    return [_serialize_acme_config(c) for c in configs]


@router.get("/acme/configs/{config_id}")
async def get_acme_config(config_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    result = await db.execute(select(AcmeConfig).where(AcmeConfig.id == config_id))
    c = result.scalar_one_or_none()
    if not c:
        raise HTTPException(404, "ACME config not found")
    return {
        "id": c.id,
        "name": c.name,
        "acme_server": c.acme_server,
        "email": c.email,
        "dns_provider": c.dns_provider,
        "dns_credentials": c.dns_credentials,
        "renew_days": c.renew_days,
        "has_account_key": c.account_key is not None,
    }


@router.post("/acme/configs", status_code=201)
async def create_acme_config(data: AcmeConfigCreate, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    config = AcmeConfig(**data.model_dump())
    await db_add(db, config, refresh=True)
    return {"id": config.id, "name": config.name}


@router.put("/acme/configs/{config_id}")
async def update_acme_config(config_id: int, data: AcmeConfigUpdate, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    result = await db.execute(select(AcmeConfig).where(AcmeConfig.id == config_id))
    config = result.scalar_one_or_none()
    if not config:
        raise HTTPException(404, "ACME config not found")
    await db_update(db, config, data.model_dump(exclude_unset=True))
    return {"id": config.id}


@router.delete("/acme/configs/{config_id}")
async def delete_acme_config(config_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    result = await db.execute(select(AcmeConfig).where(AcmeConfig.id == config_id))
    config = result.scalar_one_or_none()
    if not config:
        raise HTTPException(404, "ACME config not found")
    # 解除关联的域名
    result = await db.execute(select(Domain).where(Domain.acme_config_id == config_id))
    for d in result.scalars().all():
        d.acme_config_id = None
    await db_delete(db, config)
    return {"ok": True}


# ──────────────── ACME 操作 ────────────────

async def _get_domain_acme_config(domain: Domain, db: AsyncSession) -> AcmeConfig:
    """获取域名关联的 ACME 配置，未关联则使用第一个可用配置"""
    if domain.acme_config_id:
        result = await db.execute(select(AcmeConfig).where(AcmeConfig.id == domain.acme_config_id))
        config = result.scalar_one_or_none()
        if config:
            return config
    # 回退：使用第一个有邮箱的配置
    result = await db.execute(select(AcmeConfig).where(AcmeConfig.email != "").order_by(AcmeConfig.id))
    config = result.scalars().first()
    if not config:
        raise HTTPException(400, "无可用的 ACME 配置，请先配置邮箱")
    return config


@router.post("/acme/issue/{domain_id}")
async def issue_cert(domain_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    """为指定域名申请证书"""
    from backend.acme_service import AcmeService
    from backend.config import get_settings

    # 获取域名
    result = await db.execute(select(Domain).where(Domain.id == domain_id))
    domain = result.scalar_one_or_none()
    if not domain:
        raise HTTPException(404, "Domain not found")

    # 获取该域名关联的 ACME 配置
    config = await _get_domain_acme_config(domain, db)

    settings = get_settings()
    service = AcmeService(config, settings.cert_dir)

    # 记录开始
    log = AcmeLog(domain_id=domain_id, action="issue", status="pending", message=f"Issuing certificate for {domain.domain}")
    db.add(log)
    await db_flush(db)

    try:
        success, msg, cert_info, raw_output = service.issue_certificate(domain.domain)
        log.status = "success" if success else "failed"
        log.message = msg
        log.detail = raw_output[-5000:] if raw_output else None  # 截断防止过大

        if success:
            if cert_info.get("not_after"):
                domain.cert_not_after = datetime.fromisoformat(cert_info["not_after"])
                domain.version = str(int(domain.version or "0") + 1)
                log.cert_not_after = domain.cert_not_after
                log.cert_serial = cert_info.get("serial_number", "")
                log.cert_san = ", ".join(cert_info.get("san", []))
                logger.info(f"证书信息已更新: {domain.domain}, 过期时间: {cert_info['not_after']}")
            else:
                domain.version = str(int(domain.version or "0") + 1)
                logger.warning(f"无法获取证书信息: {domain.domain}")

        await db.commit()
        return {"success": success, "message": msg}
    except Exception as e:
        log.status = "failed"
        log.message = str(e)
        await db.commit()
        raise HTTPException(500, str(e))


@router.post("/acme/renew/{domain_id}")
async def renew_cert(domain_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    """续签指定域名的证书"""
    return await issue_cert(domain_id, db)


@router.post("/acme/auto-renew")
async def auto_renew_all(db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    """自动续签所有即将过期的证书"""
    from backend.acme_service import AcmeService
    from backend.config import get_settings

    settings = get_settings()

    # 获取所有域名
    result = await db.execute(select(Domain).options(selectinload(Domain.acme_config)))
    domains = result.scalars().all()

    # 按 ACME 配置分组，为每个配置创建一个 service 实例
    services: dict[int, AcmeService] = {}

    results = []
    for d in domains:
        try:
            config = await _get_domain_acme_config(d, db)
        except HTTPException:
            results.append({"domain": d.domain, "success": False, "message": "无可用 ACME 配置", "days_left": 0})
            continue

        if config.id not in services:
            services[config.id] = AcmeService(config, settings.cert_dir)
        service = services[config.id]

        need_renew, days_left = service.check_expiry(d.domain)
        if need_renew:
            log = AcmeLog(domain_id=d.id, action="renew", status="pending", message=f"Auto-renewing {d.domain} (config: {config.name})")
            db.add(log)
            await db_flush(db)

            success, msg, cert_info, raw_output = service.renew_certificate(d.domain)
            log.status = "success" if success else "failed"
            log.message = msg
            log.detail = raw_output[-5000:] if raw_output else None

            if success and cert_info.get("not_after"):
                d.cert_not_after = datetime.fromisoformat(cert_info["not_after"])
                d.version = str(int(d.version or "0") + 1)
                log.cert_not_after = d.cert_not_after
                log.cert_serial = cert_info.get("serial_number", "")
                log.cert_san = ", ".join(cert_info.get("san", []))

            results.append({"domain": d.domain, "success": success, "message": msg, "days_left": days_left})
        else:
            results.append({"domain": d.domain, "success": True, "message": f"Skipped, {days_left} days left", "days_left": days_left})

    await db.commit()
    return {"results": results}


# ──────────────── ACME 日志 ────────────────

@router.get("/acme/logs")
async def list_acme_logs(status: str | None = None, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    query = (
        select(AcmeLog)
        .options(selectinload(AcmeLog.domain))
        .order_by(AcmeLog.created_at.desc())
        .limit(100)
    )
    if status:
        query = query.where(AcmeLog.status == status)
    result = await db.execute(query)
    logs = result.scalars().all()
    return [
        {
            "id": log.id,
            "domain": log.domain.domain if log.domain else None,
            "action": log.action,
            "status": log.status,
            "message": log.message,
            "detail": log.detail,
            "cert_not_after": log.cert_not_after.isoformat() if log.cert_not_after else None,
            "cert_serial": log.cert_serial,
            "cert_san": log.cert_san,
            "created_at": log.created_at.isoformat(),
        }
        for log in logs
    ]


# ──────────────── 部署脚本 ────────────────

async def _render_script(domain: str, server_name: str, os: str, db: AsyncSession, acme_config_id: int | None = None) -> tuple[str, Server, Domain]:
    """渲染部署脚本，返回 (script, server, domain_obj)"""
    from jinja2 import Environment, FileSystemLoader

    template_dir = Path(__file__).parent.parent / "templates_cert"
    jinja_env = Environment(loader=FileSystemLoader(str(template_dir)))

    result = await db.execute(select(Server).where(Server.name == server_name))
    server = result.scalars().first()
    if not server:
        raise HTTPException(404, "Server not found")

    # 同一域名+服务器可能有多条记录（测试/正式不同 ACME 配置），用 acme_config_id 精确匹配
    conditions = [Domain.domain == domain, Domain.server_id == server.id]
    if acme_config_id is not None:
        conditions.append(Domain.acme_config_id == acme_config_id)
    result = await db.execute(select(Domain).where(*conditions))
    d = result.scalars().first()
    if not d:
        raise HTTPException(404, "Domain not found for this server")

    settings = get_settings()
    template_name = "deploy-cert.ps1.j2" if os == "windows" else "deploy-cert.sh.j2"
    tpl = jinja_env.get_template(template_name)
    script = tpl.render(
        domain=domain,
        base_url=settings.base_url,
        token=server.token,
        cert_dir=d.cert_dir,
        check_cmd=d.check_cmd,
        reload_cmd=d.reload_cmd,
        acme_config_id=d.acme_config_id or "",
    )
    return script, server, d


@router.get("/script")
async def admin_generate_script(
    domain: str,
    server_name: str,
    os: str = "linux",
    acme_config_id: int | None = None,
    db: AsyncSession = Depends(get_db),
    _: bool = Depends(require_auth),
):
    """管理端生成部署脚本（JSON）"""
    script, server, d = await _render_script(domain, server_name, os, db, acme_config_id)
    return {
        "script": script,
        "token": server.token,
        "cert_dir": d.cert_dir,
        "check_cmd": d.check_cmd,
        "reload_cmd": d.reload_cmd,
    }


@router.get("/script/download")
async def admin_download_script(
    domain: str,
    server_name: str,
    os: str = "linux",
    acme_config_id: int | None = None,
    db: AsyncSession = Depends(get_db),
    _: bool = Depends(require_auth),
):
    """下载脚本文件（纯文本，供 curl 使用）"""
    script, _, _ = await _render_script(domain, server_name, os, db, acme_config_id)
    ext = "ps1" if os == "windows" else "sh"
    filename = f"deploy-{domain.replace('*', '_')}.{ext}"
    return PlainTextResponse(
        script,
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )


@router.get("/script/setup")
async def admin_setup_script(
    request: Request,
    domain: str,
    server_name: str,
    os: str = "linux",
    token: str = "",
    acme_config_id: int | None = None,
    db: AsyncSession = Depends(get_db),
    _: bool = Depends(require_auth),
):
    """生成一键安装脚本（纯文本下载）"""
    from jinja2 import Environment, FileSystemLoader

    template_dir = Path(__file__).parent.parent / "templates_cert"
    jinja_env = Environment(loader=FileSystemLoader(str(template_dir)))

    from urllib.parse import quote
    settings = get_settings()
    base_url = settings.base_url.rstrip("/")
    ext = "ps1" if os == "windows" else "sh"
    download_url = f"{base_url}/admin/api/script/download?domain={quote(domain)}&server_name={quote(server_name)}&os={os}"
    if acme_config_id is not None:
        download_url += f"&acme_config_id={acme_config_id}"

    template_name = "setup-cert.ps1.j2" if os == "windows" else "setup-cert.sh.j2"
    tpl = jinja_env.get_template(template_name)
    script = tpl.render(
        domain=domain,
        server_name=server_name,
        download_url=download_url,
        admin_token=token,
    )
    filename = f"setup-{domain.replace('*', '_')}.{ext}"
    return PlainTextResponse(
        script,
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )


# ──────────────── 证书信息 ────────────────

@router.get("/cert-info/{domain_id}")
async def get_cert_info(domain_id: int, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    """获取证书详细信息"""
    from backend.acme_service import AcmeService
    from backend.config import get_settings

    result = await db.execute(select(Domain).where(Domain.id == domain_id))
    domain = result.scalar_one_or_none()
    if not domain:
        raise HTTPException(404, "Domain not found")

    config = await _get_domain_acme_config(domain, db)

    settings = get_settings()
    service = AcmeService(config, settings.cert_dir)

    info = service.get_cert_info(domain.domain)
    need_renew, days_left = service.check_expiry(domain.domain)

    return {
        "domain": domain.domain,
        "cert_info": info,
        "need_renew": need_renew,
        "days_left": days_left,
    }


# ──────────────── 证书下载 ────────────────

def _cert_store_dir(domain: str) -> str:
    """泛域名用裸域名作为存储目录"""
    return domain[2:] if domain.startswith("*.") else domain


@router.get("/cert-download/{domain_id}/{file_type}")
async def download_cert_file(domain_id: int, file_type: str, db: AsyncSession = Depends(get_db), _: bool = Depends(require_auth)):
    """下载证书文件，file_type: fullchain 或 private"""
    from fastapi.responses import FileResponse

    if file_type not in ("fullchain", "private"):
        raise HTTPException(400, "file_type must be fullchain or private")

    result = await db.execute(select(Domain).where(Domain.id == domain_id))
    domain = result.scalar_one_or_none()
    if not domain:
        raise HTTPException(404, "Domain not found")

    settings = get_settings()
    store_dir = _cert_store_dir(domain.domain)

    if file_type == "fullchain":
        file_path = Path(settings.cert_dir) / store_dir / "fullchain.pem"
        filename = f"{store_dir}-fullchain.pem"
    else:
        file_path = Path(settings.cert_dir) / store_dir / "private.key"
        filename = f"{store_dir}-private.key"

    logger.debug(f"下载证书: domain={domain.domain}, store_dir={store_dir}, file_path={file_path}, exists={file_path.exists()}")

    if not file_path.exists():
        raise HTTPException(404, f"文件不存在 ({file_path})，请先申请证书")

    return FileResponse(
        str(file_path),
        media_type="application/x-pem-file",
        filename=filename,
    )