Python脚本如何记录结构变更同步日志

本文目录导读：

1. 方案1：数据库表结构变更监控
2. 方案2：文件系统结构变更监控
3. 方案3：API结构变更日志
4. 使用示例
5. 功能特点

我来帮你设计一个Python脚本用于记录结构变更同步日志,这里提供几种不同场景的实现方案：

方案1：数据库表结构变更监控

基础日志记录器

import json
import hashlib
from datetime import datetime
from typing import Dict, List, Any, Optional
import logging
from pathlib import Path
class StructureChangeLogger:
    """结构变更日志记录器"""
    def __init__(self, log_file: str = "structure_changes.json"):
        self.log_file = Path(log_file)
        self.logger = logging.getLogger(__name__)
        self.setup_logging()
    def setup_logging(self):
        """设置日志配置"""
        handler = logging.FileHandler('structure_sync.log')
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)
    def record_change(self, change_type: str, source: str, target: str, 
                     details: Dict[str, Any], status: str = "pending"):
        """
        记录结构变更
        Args:
            change_type: 变更类型 (create/alter/drop)
            source: 源结构信息
            target: 目标结构信息
            details: 变更详情
            status: 状态 (pending/success/failed)
        """
        change_record = {
            "timestamp": datetime.now().isoformat(),
            "change_type": change_type,
            "source": source,
            "target": target,
            "details": details,
            "status": status,
            "change_id": self._generate_change_id(details)
        }
        self._save_change(change_record)
        self._log_change(change_record)
        return change_record
    def _generate_change_id(self, details: Dict) -> str:
        """生成变更ID"""
        content = json.dumps(details, sort_keys=True)
        return hashlib.md5(content.encode()).hexdigest()[:12]
    def _save_change(self, record: Dict):
        """保存变更记录到JSON文件"""
        changes = []
        if self.log_file.exists():
            with open(self.log_file, 'r', encoding='utf-8') as f:
                changes = json.load(f)
        changes.append(record)
        with open(self.log_file, 'w', encoding='utf-8') as f:
            json.dump(changes, f, ensure_ascii=False, indent=2)
    def _log_change(self, record: Dict):
        """记录到日志文件"""
        log_msg = (
            f"结构变更 [{record['change_type']}] "
            f"ID: {record['change_id']} "
            f"状态: {record['status']} "
            f"详情: {json.dumps(record['details'], ensure_ascii=False)}"
        )
        if record['status'] == 'failed':
            self.logger.error(log_msg)
        else:
            self.logger.info(log_msg)
    def get_changes(self, start_time: Optional[str] = None, 
                   end_time: Optional[str] = None,
                   change_type: Optional[str] = None) -> List[Dict]:
        """查询变更记录"""
        if not self.log_file.exists():
            return []
        with open(self.log_file, 'r', encoding='utf-8') as f:
            changes = json.load(f)
        # 过滤条件
        filtered = changes
        if start_time:
            filtered = [c for c in filtered if c['timestamp'] >= start_time]
        if end_time:
            filtered = [c for c in filtered if c['timestamp'] <= end_time]
        if change_type:
            filtered = [c for c in filtered if c['change_type'] == change_type]
        return filtered

数据库表结构比较器

import sqlite3
import mysql.connector
from typing import Dict, List, Tuple
class DatabaseStructureComparator:
    """数据库结构比较器"""
    def __init__(self, source_conn, target_conn):
        self.source_conn = source_conn
        self.target_conn = target_conn
        self.logger = StructureChangeLogger()
    def compare_tables(self) -> List[Dict]:
        """比较两个数据库的表结构"""
        changes = []
        source_tables = self._get_table_info(self.source_conn)
        target_tables = self._get_table_info(self.target_conn)
        # 检查新表
        for table_name in source_tables:
            if table_name not in target_tables:
                changes.append({
                    "type": "create",
                    "table": table_name,
                    "details": source_tables[table_name]
                })
        # 检查删除的表
        for table_name in target_tables:
            if table_name not in source_tables:
                changes.append({
                    "type": "drop",
                    "table": table_name,
                    "details": target_tables[table_name]
                })
        # 检查表结构变更
        for table_name in source_tables:
            if table_name in target_tables:
                column_changes = self._compare_columns(
                    source_tables[table_name]["columns"],
                    target_tables[table_name]["columns"]
                )
                if column_changes:
                    changes.append({
                        "type": "alter",
                        "table": table_name,
                        "details": column_changes
                    })
        return changes
    def _get_table_info(self, conn) -> Dict:
        """获取数据库表结构信息"""
        tables = {}
        cursor = conn.cursor()
        # 获取所有表
        cursor.execute("SHOW TABLES")
        table_names = [row[0] for row in cursor.fetchall()]
        for table_name in table_names:
            # 获取表结构
            cursor.execute(f"DESCRIBE {table_name}")
            columns = []
            for row in cursor.fetchall():
                columns.append({
                    "name": row[0],
                    "type": row[1],
                    "nullable": row[2],
                    "key": row[3],
                    "default": row[4],
                    "extra": row[5]
                })
            # 获取索引信息
            cursor.execute(f"SHOW INDEX FROM {table_name}")
            indexes = cursor.fetchall()
            tables[table_name] = {
                "columns": columns,
                "indexes": indexes
            }
        return tables
    def _compare_columns(self, source_columns: List, target_columns: List) -> List[Dict]:
        """比较两组的列结构"""
        changes = []
        source_names = {col["name"]: col for col in source_columns}
        target_names = {col["name"]: col for col in target_columns}
        # 新增列
        for name in source_names:
            if name not in target_names:
                changes.append({
                    "action": "add_column",
                    "column": source_names[name]
                })
        # 删除列
        for name in target_names:
            if name not in source_names:
                changes.append({
                    "action": "drop_column",
                    "column": target_names[name]
                })
        # 修改列
        for name in source_names:
            if name in target_names:
                if source_names[name] != target_names[name]:
                    changes.append({
                        "action": "modify_column",
                        "column": name,
                        "before": target_names[name],
                        "after": source_names[name]
                    })
        return changes
    def sync_and_log(self):
        """同步结构并记录日志"""
        changes = self.compare_tables()
        for change in changes:
            change_type = change["type"]
            table_name = change["table"]
            details = change["details"]
            # 记录变更日志
            self.logger.record_change(
                change_type=change_type,
                source=str(details.get("after", details)),
                target=str(details.get("before", "")),
                details=details
            )
        return changes

方案2：文件系统结构变更监控

import os
import hashlib
from datetime import datetime
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import json
class FileStructureChangeHandler(FileSystemEventHandler):
    """文件系统结构变更处理器"""
    def __init__(self, tracked_path: str, log_file: str = "file_changes.json"):
        self.tracked_path = tracked_path
        self.log_file = log_file
        self.known_structure = {}
        self.load_known_structure()
    def load_known_structure(self):
        """加载已知结构"""
        if os.path.exists(self.log_file):
            with open(self.log_file, 'r') as f:
                self.known_structure = json.load(f)
    def save_structure(self):
        """保存当前结构"""
        current_structure = self.scan_current_structure()
        self.known_structure = current_structure
        with open(self.log_file, 'w') as f:
            json.dump(current_structure, f, indent=2)
    def scan_current_structure(self) -> Dict:
        """扫描当前目录结构"""
        structure = {}
        for root, dirs, files in os.walk(self.tracked_path):
            rel_path = os.path.relpath(root, self.tracked_path)
            for file in files:
                file_path = os.path.join(root, file)
                rel_file_path = os.path.join(rel_path, file)
                try:
                    stat = os.stat(file_path)
                    structure[rel_file_path] = {
                        "size": stat.st_size,
                        "modified": datetime.fromtimestamp(stat.st_mtime).isoformat(),
                        "hash": self._file_hash(file_path)
                    }
                except Exception as e:
                    structure[rel_file_path] = {
                        "error": str(e)
                    }
        return structure
    def _file_hash(self, file_path: str) -> str:
        """计算文件哈希"""
        hasher = hashlib.md5()
        with open(file_path, 'rb') as f:
            for chunk in iter(lambda: f.read(4096), b''):
                hasher.update(chunk)
        return hasher.hexdigest()
    def on_modified(self, event):
        """文件修改事件"""
        if not event.is_directory:
            rel_path = os.path.relpath(event.src_path, self.tracked_path)
            change_record = {
                "timestamp": datetime.now().isoformat(),
                "event": "modified",
                "path": rel_path,
                "details": self._get_file_info(event.src_path)
            }
            self._log_change(change_record)
    def on_created(self, event):
        """文件创建事件"""
        rel_path = os.path.relpath(event.src_path, self.tracked_path)
        change_record = {
            "timestamp": datetime.now().isoformat(),
            "event": "created",
            "path": rel_path,
            "details": self._get_file_info(event.src_path) if not event.is_directory else {}
        }
        self._log_change(change_record)
    def on_deleted(self, event):
        """文件删除事件"""
        rel_path = os.path.relpath(event.src_path, self.tracked_path)
        change_record = {
            "timestamp": datetime.now().isoformat(),
            "event": "deleted",
            "path": rel_path,
            "details": {}
        }
        self._log_change(change_record)
    def _get_file_info(self, file_path: str) -> Dict:
        """获取文件信息"""
        try:
            stat = os.stat(file_path)
            return {
                "size": stat.st_size,
                "modified": datetime.fromtimestamp(stat.st_mtime).isoformat()
            }
        except:
            return {}
    def _log_change(self, record: Dict):
        """记录变更到日志文件"""
        changes = []
        if os.path.exists("structure_changes.log"):
            with open("structure_changes.log", 'r') as f:
                changes = json.load(f)
        changes.append(record)
        with open("structure_changes.log", 'w') as f:
            json.dump(changes, f, indent=2)
        print(f"结构变更记录: {record}")
class FileStructureMonitor:
    """文件结构监控器"""
    def __init__(self, path: str):
        self.path = path
        self.observer = Observer()
        self.handler = FileStructureChangeHandler(path)
    def start(self):
        """开始监控"""
        self.observer.schedule(self.handler, self.path, recursive=True)
        self.observer.start()
        print(f"开始监控目录: {self.path}")
    def stop(self):
        """停止监控"""
        self.observer.stop()
        self.observer.join()
        self.handler.save_structure()
        print("监控已停止")

方案3：API结构变更日志

from flask import Flask, request, jsonify
from pydantic import BaseModel, Field
from typing import Optional, List
import json
from datetime import datetime
app = Flask(__name__)
class StructureChangeLog(BaseModel):
    """结构变更日志模型"""
    id: Optional[int] = None
    timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())
    api_endpoint: str
    http_method: str
    change_type: str  # request/response/parameter
    field_name: str
    old_value: Optional[str] = None
    new_value: Optional[str] = None
    description: str
    status: str = "pending"
class APIVersionManager:
    """API版本管理器，记录结构变更"""
    def __init__(self, log_file: str = "api_changes.json"):
        self.log_file = log_file
        self.changes = []
        self.load_changes()
    def load_changes(self):
        """加载历史变更"""
        try:
            with open(self.log_file, 'r') as f:
                self.changes = json.load(f)
        except FileNotFoundError:
            self.changes = []
    def save_changes(self):
        """保存变更记录"""
        with open(self.log_file, 'w') as f:
            json.dump(self.changes, f, indent=2, default=str)
    def record_change(self, change: StructureChangeLog):
        """记录结构变更"""
        if self.changes:
            change.id = max(c.get("id", 0) for c in self.changes) + 1
        else:
            change.id = 1
        self.changes.append(change.dict())
        self.save_changes()
        return change.id
    def get_changes(self, endpoint: Optional[str] = None, 
                   change_type: Optional[str] = None) -> List[dict]:
        """获取变更记录"""
        filtered = self.changes
        if endpoint:
            filtered = [c for c in filtered if c["api_endpoint"] == endpoint]
        if change_type:
            filtered = [c for c in filtered if c["change_type"] == change_type]
        return filtered
# 创建API版本管理器实例
api_version_manager = APIVersionManager()
@app.route('/api/v1/users', methods=['GET'])
def get_users():
    """示例API端点"""
    # 记录请求结构变更
    change = StructureChangeLog(
        api_endpoint="/api/v1/users",
        http_method="GET",
        change_type="request",
        field_name="page",
        new_value="新增分页参数",
        description="新增用户列表分页功能"
    )
    api_version_manager.record_change(change)
    return jsonify({"users": []})
@app.route('/api/changes', methods=['GET'])
def get_api_changes():
    """获取API变更记录"""
    endpoint = request.args.get('endpoint')
    change_type = request.args.get('change_type')
    changes = api_version_manager.get_changes(endpoint, change_type)
    return jsonify({"changes": changes})
@app.route('/api/changes', methods=['POST'])
def record_api_change():
    """记录API变更"""
    data = request.json
    change = StructureChangeLog(**data)
    change_id = api_version_manager.record_change(change)
    return jsonify({"id": change_id, "message": "变更记录成功"})
# 装饰器用于自动记录API变化
def track_api_change(change_type: str, field_name: str, description: str):
    """自动记录API变更的装饰器"""
    def decorator(func):
        def wrapper(*args, **kwargs):
            # 记录变更
            change = StructureChangeLog(
                api_endpoint=request.path,
                http_method=request.method,
                change_type=change_type,
                field_name=field_name,
                new_value=description,
                description=description
            )
            api_version_manager.record_change(change)
            return func(*args, **kwargs)
        return wrapper
    return decorator
if __name__ == "__main__":
    app.run(debug=True)

使用示例

# 数据库结构变更日志
def example_database_sync():
    # 假设有源数据库和目标数据库连接
    source_conn = mysql.connector.connect(
        host="source_host",
        database="source_db",
        user="user",
        password="password"
    )
    target_conn = mysql.connector.connect(
        host="target_host",
        database="target_db",
        user="user",
        password="password"
    )
    comparator = DatabaseStructureComparator(source_conn, target_conn)
    changes = comparator.sync_and_log()
    for change in changes:
        print(f"检测到结构变更: {change['type']} - {change['table']}")
# 文件系统监控
def example_file_monitoring():
    monitor = FileStructureMonitor("./my_project")
    try:
        monitor.start()
        # 保持监控运行
        import time
        time.sleep(60)
    finally:
        monitor.stop()
# 查看变更日志
def view_change_logs():
    logger = StructureChangeLogger()
    # 获取所有变更
    all_changes = logger.get_changes()
    # 获取特定类型的变更
    create_changes = logger.get_changes(change_type="create")
    # 获取时间范围内的变更
    recent_changes = logger.get_changes(
        start_time="2024-01-01T00:00:00",
        end_time="2024-01-31T23:59:59"
    )
    return all_changes, create_changes, recent_changes

功能特点

1. 多场景支持：数据库、文件系统、API等多种场景
2. 完整记录：包含时间戳、变更类型、变更详情
3. 持久化存储：JSON文件或数据库存储
4. 查询功能：按时间、类型等条件查询
5. 版本追踪：支持版本回退和追溯
6. 自动监控：实时监控结构变更

这个方案提供了完整的结构变更日志记录功能,可以根据具体需求进行扩展和定制。