document-management/backend/dms/api/v1/backup.py

"""备份管理接口（契约1.21），对齐 Java 版 BackupController。

配置读写 + 手动触发 + 进度查询 + 历史列表 + 目标连通性测试（契约1.23）。
全部仅系统管理权限可访问；手动触发在任务终态记 BACKUP_CREATE（成功/失败
含原因）、目标测试记 BACKUP_TEST_TARGET 审计。密码不回显（仅回是否已设置），
PUT 空值保持不变。
"""

from __future__ import annotations

import logging

from flask import Flask, current_app, request

from dms.api.v1.blueprint import api_v1
from dms.common.enums import AuditAction, AuditResult, AuditTarget
from dms.common.errors import InvalidArgumentError
from dms.common.response import success_response
from dms.extensions import db
from dms.models import AuditLog
from dms.security.decorators import bearer_auth_required, system_management_required
from dms.services.audit_service import business_audit
from dms.services.backup_service import (
    TRIGGER_MANUAL,
    TYPE_FULL,
    TYPE_INCREMENTAL,
    BackupConfig,
    BackupEngine,
    BackupOutcome,
)

logger = logging.getLogger(__name__)

_CONFIG_FIELDS = (
    "target_mode",
    "local_path",
    "rsync_url",
    "rsync_password",
    "schedule_enabled",
    "schedule_time",
    "full_interval_days",
    "retain_full_count",
)


def _engine() -> BackupEngine:
    engine = current_app.extensions["dms_backup_engine"]
    assert isinstance(engine, BackupEngine)
    return engine


def _json_body() -> dict:
    if not request.is_json:
        raise InvalidArgumentError("请求体必须是JSON对象")
    body = request.get_json(silent=True)
    if not isinstance(body, dict):
        raise InvalidArgumentError("请求体必须是JSON对象")
    return body


def _exact_payload(fields: tuple[str, ...]) -> dict:
    body = _json_body()
    if set(body) != set(fields):
        raise InvalidArgumentError(
            "请求体字段必须且只能包含：" + "、".join(fields)
        )
    return body


def _required_text(value, field: str, max_length: int) -> str:
    if not isinstance(value, str) or not value.strip():
        raise InvalidArgumentError(f"{field}不能为空")
    normalized = value.strip()
    if len(normalized) > max_length:
        raise InvalidArgumentError(f"{field}长度不能超过{max_length}")
    return normalized


def _nullable_text(value, field: str, max_length: int) -> str | None:
    if value is None:
        return None
    if not isinstance(value, str):
        raise InvalidArgumentError(f"{field}必须是字符串或null")
    normalized = value.strip()
    if len(normalized) > max_length:
        raise InvalidArgumentError(f"{field}长度不能超过{max_length}")
    return normalized or None


def _boolean_value(value, field: str) -> bool:
    if not isinstance(value, bool):
        raise InvalidArgumentError(f"{field}必须是布尔值")
    return value


def _int_value(value, field: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int):
        raise InvalidArgumentError(f"{field}必须是整数")
    return value


@api_v1.get("/backup/config")
@bearer_auth_required
@system_management_required
def get_backup_config():
    return success_response(_engine().load_config().public_view())


@api_v1.put("/backup/config")
@bearer_auth_required
@system_management_required
def update_backup_config():
    engine = _engine()
    payload = _exact_payload(_CONFIG_FIELDS)
    current = engine.load_config()
    current.target_mode = _required_text(payload["target_mode"], "target_mode", 16)
    current.local_path = _nullable_text(payload["local_path"], "local_path", 1024) or ""
    current.rsync_url = _nullable_text(payload["rsync_url"], "rsync_url", 1024) or ""
    # 密码空值 = 保持不变（回显永远是掩码，前端不回传明文）
    password = _nullable_text(payload["rsync_password"], "rsync_password", 256)
    if password is not None:
        current.rsync_password = password
    current.schedule_enabled = _boolean_value(
        payload["schedule_enabled"], "schedule_enabled"
    )
    current.schedule_time = _required_text(payload["schedule_time"], "schedule_time", 5)
    current.full_interval_days = _int_value(
        payload["full_interval_days"], "full_interval_days"
    )
    current.retain_full_count = _int_value(payload["retain_full_count"], "retain_full_count")
    current.validate()
    current.save(engine.backup_config_path)
    return success_response(current.public_view())


@api_v1.post("/backup/jobs")
@bearer_auth_required
@system_management_required
def start_backup_job():
    engine = _engine()
    payload = _exact_payload(("type",))
    backup_type = _required_text(payload["type"], "type", 16)
    if backup_type not in (TYPE_FULL, TYPE_INCREMENTAL):
        raise InvalidArgumentError("type 必须为 FULL 或 INCREMENTAL")
    # 审计在任务终态落库（反映真实结果）：此处先按请求上下文快照用户/IP/UA
    record = business_audit(
        action=AuditAction.BACKUP_CREATE,
        target=AuditTarget.BACKUP,
        target_id=0,
        target_name=backup_type,
    )
    app = current_app._get_current_object()
    engine.start_async(
        backup_type,
        TRIGGER_MANUAL,
        on_complete=lambda outcome: _write_completion_audit(app, record, outcome),
    )
    return success_response(engine.progress_snapshot(), status=202)


def _write_completion_audit(
    app: Flask, record: AuditLog, outcome: BackupOutcome
) -> None:
    """任务终态审计落库：成功/失败（含失败原因）；写入失败不否定备份结果。"""
    try:
        with app.app_context():
            record.target_name = outcome.type
            record.operation_result = (
                AuditResult.SUCCESS.value
                if outcome.success
                else AuditResult.FAILURE.value
            )
            record.failure_reason = (
                None
                if outcome.success or outcome.error is None
                else outcome.error[:1000]
            )
            record.operation_detail = {
                "type": outcome.type,
                "trigger": outcome.trigger,
                "backupId": outcome.backup_id,
            }
            db.session.add(record)
            db.session.commit()
    except Exception:  # noqa: BLE001 - 审计写入失败不否定已完成的备份
        logger.warning("备份完成审计写入失败", exc_info=True)
        with app.app_context():
            db.session.rollback()


@api_v1.get("/backup/jobs/current")
@bearer_auth_required
@system_management_required
def backup_job_current():
    engine = _engine()
    snapshot = engine.progress_snapshot()
    snapshot["running"] = engine.is_running
    return success_response(snapshot)


@api_v1.post("/backup/jobs/current/cancel")
@bearer_auth_required
@system_management_required
def cancel_backup_job():
    engine = _engine()
    had_running = engine.is_running
    engine.cancel()
    return success_response({"canceled": had_running})


@api_v1.get("/backup/jobs")
@bearer_auth_required
@system_management_required
def list_backup_history():
    engine = _engine()
    config: BackupConfig = engine.load_config()
    return success_response({"entries": engine.list_history(config)})


@api_v1.post("/backup/test-target")
@bearer_auth_required
@system_management_required
def test_backup_target():
    """目标连通性测试（契约1.23）：探测备份目标可写性并记审计。"""
    engine = _engine()
    detail: dict = {}
    try:
        target = engine.build_target(engine.load_config())
        target.check_writable()
        detail = {"ok": True, "target": target.describe()}
    except Exception as exc:  # noqa: BLE001 - 探测失败返回结果而非报错
        detail = {"ok": False, "error": str(exc)[:200]}
    db.session.add(
        business_audit(
            action=AuditAction.BACKUP_TEST_TARGET,
            target=AuditTarget.BACKUP,
            target_id=0,
            target_name="",
            detail=detail,
        )
    )
    db.session.commit()
    return success_response(detail)