本文目录导读：
我来帮你设计一个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
功能特点
- 多场景支持：数据库、文件系统、API等多种场景
- 完整记录：包含时间戳、变更类型、变更详情
- 持久化存储：JSON文件或数据库存储
- 查询功能：按时间、类型等条件查询
- 版本追踪：支持版本回退和追溯
- 自动监控：实时监控结构变更
这个方案提供了完整的结构变更日志记录功能,可以根据具体需求进行扩展和定制。
