#!/usr/bin/env python3 """ MySQL 数据定时备份到 Cloudflare D1 - 定时：每小时 XX:30 - 模式：按游标列增量同步（timestamp/created_at/updated_at/id） """ import json import os import threading import time from datetime import datetime from decimal import Decimal from typing import Any import httpx import schedule from dotenv import load_dotenv from api.models import get_db_connection from utils.project_logger import get_project_logger load_dotenv() logger = get_project_logger("mysql_d1_backup") class MySQLD1BackupService: """MySQL -> D1 备份服务。""" def __init__(self): self.enabled = os.getenv("D1_BACKUP_ENABLED", "false").lower() == "true" self.api_token = os.getenv("CF_D1_API_TOKEN", "") self.account_id = os.getenv("CF_D1_ACCOUNT_ID", "") self.database_id = os.getenv("CF_D1_DATABASE_ID", "") self.batch_size = int(os.getenv("D1_BACKUP_BATCH_SIZE", "500")) self.max_sql_params = int(os.getenv("D1_MAX_SQL_PARAMS", "90")) self.tables_config = os.getenv("D1_BACKUP_TABLES", "").strip() self._job_lock = threading.Lock() def is_config_ready(self) -> bool: """检查 D1 配置是否完整。""" if not self.enabled: return False return bool(self.api_token and self.account_id and self.database_id) def _d1_endpoint(self) -> str: return ( f"https://api.cloudflare.com/client/v4/accounts/" f"{self.account_id}/d1/database/{self.database_id}/query" ) def _query_d1(self, sql: str, params: list[Any] | None = None) -> dict: """执行 D1 SQL。""" headers = { "Authorization": f"Bearer {self.api_token}", "Content-Type": "application/json", } statement: dict[str, Any] = {"sql": sql} if params is not None: statement["params"] = params with httpx.Client(timeout=30) as client: # Cloudflare D1 REST API 对 /query 更稳定的格式是语句数组 resp = client.post(self._d1_endpoint(), headers=headers, json=[statement]) if resp.status_code >= 400: # 兼容部分环境下使用对象格式的情况，失败时回退重试一次 fallback_resp = client.post(self._d1_endpoint(), headers=headers, json=statement) if fallback_resp.status_code >= 400: sql_preview = " ".join(sql.strip().split())[:180] raise RuntimeError( f"D1 HTTP {fallback_resp.status_code}，sql={sql_preview}，响应={fallback_resp.text[:500]}" ) data = fallback_resp.json() else: data = resp.json() if not data.get("success", False): raise RuntimeError(f"D1 SQL 执行失败: {data}") return data def _d1_results(self, response: dict) -> list[dict]: """解析 D1 响应结果。""" result = response.get("result", []) if not result: return [] first = result[0] if isinstance(result, list) else result return first.get("results", []) if isinstance(first, dict) else [] def _normalize_value(self, value: Any) -> Any: """把 MySQL 字段值转换为可写入 D1 的类型。""" if value is None: return None if isinstance(value, (datetime,)): return value.strftime("%Y-%m-%d %H:%M:%S") if isinstance(value, Decimal): return float(value) if isinstance(value, bytes): return value.hex() if isinstance(value, (dict, list)): return json.dumps(value, ensure_ascii=False) return value def _sqlite_type_from_mysql(self, mysql_type: str) -> str: t = mysql_type.lower() if any(x in t for x in ["int", "bit", "bool"]): return "INTEGER" if any(x in t for x in ["double", "float", "decimal", "numeric", "real"]): return "REAL" if any(x in t for x in ["blob", "binary", "varbinary"]): return "BLOB" return "TEXT" def _get_mysql_tables(self) -> list[str]: """获取待备份表。""" if self.tables_config: return [x.strip() for x in self.tables_config.split(",") if x.strip()] with get_db_connection() as conn: cursor = conn.cursor() cursor.execute("SHOW TABLES") rows = cursor.fetchall() tables: list[str] = [] for row in rows: # DictCursor 下 SHOW TABLES 返回单键字典 table_name = list(row.values())[0] tables.append(str(table_name)) return tables def _get_mysql_columns(self, table_name: str) -> list[dict]: with get_db_connection() as conn: cursor = conn.cursor() cursor.execute(f"SHOW COLUMNS FROM `{table_name}`") return list(cursor.fetchall()) def _guess_cursor_column(self, columns: list[dict]) -> str | None: """选择增量同步游标列。""" column_names = {c["Field"] for c in columns} for candidate in ["timestamp", "created_at", "updated_at", "id"]: if candidate in column_names: return candidate return None def _ensure_d1_meta_tables(self): """创建 D1 元数据表。""" self._query_d1( """ CREATE TABLE IF NOT EXISTS backup_sync_state ( table_name TEXT PRIMARY KEY, cursor_col TEXT, last_cursor TEXT, updated_at TEXT ) """ ) self._query_d1( """ CREATE TABLE IF NOT EXISTS backup_run_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_at TEXT, table_name TEXT, synced_rows INTEGER, status TEXT, message TEXT ) """ ) def _ensure_d1_table(self, table_name: str, columns: list[dict]): """创建 D1 备份表。""" table_sql_name = f"backup_{table_name}" col_defs: list[str] = [] primary_keys: list[str] = [] for col in columns: field = col["Field"] col_type = self._sqlite_type_from_mysql(col["Type"]) col_defs.append(f'"{field}" {col_type}') if col.get("Key") == "PRI": primary_keys.append(field) # SQLite/D1 对复合主键需要表级 PRIMARY KEY 约束，不能多列各自 PRIMARY KEY if primary_keys: pk_cols = ", ".join([f'"{k}"' for k in primary_keys]) col_defs.append(f"PRIMARY KEY ({pk_cols})") create_sql = f"CREATE TABLE IF NOT EXISTS \"{table_sql_name}\" ({', '.join(col_defs)})" self._query_d1(create_sql) def _get_last_cursor(self, table_name: str) -> str | None: """读取 D1 已同步游标。""" resp = self._query_d1( "SELECT last_cursor FROM backup_sync_state WHERE table_name = ?", [table_name], ) rows = self._d1_results(resp) if not rows: return None return rows[0].get("last_cursor") def _update_last_cursor(self, table_name: str, cursor_col: str, cursor_value: Any): """更新 D1 已同步游标。""" now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") self._query_d1( """ INSERT INTO backup_sync_state(table_name, cursor_col, last_cursor, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(table_name) DO UPDATE SET cursor_col = excluded.cursor_col, last_cursor = excluded.last_cursor, updated_at = excluded.updated_at """, [table_name, cursor_col, str(cursor_value), now_str], ) def _log_run(self, table_name: str, synced_rows: int, status: str, message: str): """记录备份执行日志。""" run_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") self._query_d1( "INSERT INTO backup_run_log(run_at, table_name, synced_rows, status, message) VALUES (?, ?, ?, ?, ?)", [run_at, table_name, synced_rows, status, message[:1000]], ) def _fetch_mysql_rows( self, table_name: str, cursor_col: str | None, last_cursor: str | None, ) -> list[dict]: """按游标从 MySQL 抽取增量数据。""" with get_db_connection() as conn: cursor = conn.cursor() if cursor_col and last_cursor: sql = ( f"SELECT * FROM `{table_name}` WHERE `{cursor_col}` > %s " f"ORDER BY `{cursor_col}` ASC LIMIT %s" ) cursor.execute(sql, (last_cursor, self.batch_size)) elif cursor_col: sql = f"SELECT * FROM `{table_name}` ORDER BY `{cursor_col}` ASC LIMIT %s" cursor.execute(sql, (self.batch_size,)) else: sql = f"SELECT * FROM `{table_name}` LIMIT %s" cursor.execute(sql, (self.batch_size,)) return list(cursor.fetchall()) def _upsert_rows_to_d1(self, table_name: str, columns: list[dict], rows: list[dict]): """把数据批量 UPSERT 到 D1。""" if not rows: return table_sql_name = f"backup_{table_name}" col_names = [c["Field"] for c in columns] col_sql = ", ".join([f'"{c}"' for c in col_names]) # 避免 D1 的 SQL 变量数量限制触发 400 # 通过环境变量 D1_MAX_SQL_PARAMS 控制，默认保守值 90 chunk_size = max(1, self.max_sql_params // max(1, len(col_names))) for i in range(0, len(rows), chunk_size): chunk_rows = rows[i:i + chunk_size] values_clause: list[str] = [] params: list[Any] = [] for row in chunk_rows: placeholders = ", ".join(["?" for _ in col_names]) values_clause.append(f"({placeholders})") for col in col_names: params.append(self._normalize_value(row.get(col))) sql = ( f"INSERT OR REPLACE INTO \"{table_sql_name}\" ({col_sql}) " f"VALUES {', '.join(values_clause)}" ) self._query_d1(sql, params) def backup_table_once(self, table_name: str) -> int: """单表执行一次增量备份，返回同步行数。""" columns = self._get_mysql_columns(table_name) if not columns: return 0 cursor_col = self._guess_cursor_column(columns) self._ensure_d1_table(table_name, columns) total_synced = 0 last_cursor = self._get_last_cursor(table_name) while True: rows = self._fetch_mysql_rows(table_name, cursor_col, last_cursor) if not rows: break self._upsert_rows_to_d1(table_name, columns, rows) total_synced += len(rows) if cursor_col: last_cursor = self._normalize_value(rows[-1].get(cursor_col)) self._update_last_cursor(table_name, cursor_col, last_cursor) else: # 无游标列时仅做一批，避免重复全量 break if len(rows) < self.batch_size: break return total_synced def run_backup_once(self): """执行一次全表备份任务。""" if not self.is_config_ready(): logger.warning("D1 备份未启用或配置不完整，跳过本次任务") return if not self._job_lock.acquire(blocking=False): logger.warning("上一次 D1 备份尚未完成，跳过本次任务") return try: self._ensure_d1_meta_tables() tables = self._get_mysql_tables() logger.info("D1 备份任务开始，待同步表: %s", ", ".join(tables)) for table_name in tables: try: synced = self.backup_table_once(table_name) logger.info("表 %s 同步完成，本次同步 %d 行", table_name, synced) self._log_run(table_name, synced, "success", "ok") except Exception as table_error: logger.exception("表 %s 同步失败", table_name) self._log_run(table_name, 0, "failed", str(table_error)) except Exception: logger.exception("D1 备份任务执行失败") finally: self._job_lock.release() def schedule_mysql_d1_backup(): """启动 D1 备份定时任务（每小时 :30）。""" service = MySQLD1BackupService() if not service.is_config_ready(): logger.warning("D1 备份调度未启动：请检查 D1_BACKUP_ENABLED 与 Cloudflare 参数") return # 启动后先执行一次，避免首次备份必须等待到下一个 :30 service.run_backup_once() # 使用独立调度器，避免与项目内其他定时任务共享全局调度状态 local_scheduler = schedule.Scheduler() local_scheduler.every().hour.at(":30").do(service.run_backup_once) logger.info("MySQL -> D1 定时备份已启动（每小时 :30 执行）") while True: local_scheduler.run_pending() time.sleep(1) def run_mysql_d1_backup_scheduler_background(): """后台线程启动 D1 备份调度器。""" service = MySQLD1BackupService() if not service.is_config_ready(): logger.warning("D1 备份配置不完整，后台线程不启动") return None scheduler_thread = threading.Thread(target=schedule_mysql_d1_backup, daemon=True) scheduler_thread.start() logger.info("D1 备份调度器已在后台启动") return scheduler_thread if __name__ == "__main__": schedule_mysql_d1_backup()