"""备份管理接口（契约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)