监控数据版本控制的脚本如何编写

wen 实用脚本 2026-07-05 201

本文目录导读：

监控数据版本控制的脚本如何编写

基础文件版本控制脚本（Bash）
Python版本的监控脚本
带Git集成的版本控制脚本
高级版本：带数据库的版本控制（Python）
使用建议

基础文件版本控制脚本（Bash）

#!/bin/bash
# 监控文件变化并保存版本
MONITOR_DIR="/path/to/monitor"
BACKUP_DIR="/path/to/backup"
VERSION_FILE="version.txt"
# 初始化版本号
if [ ! -f "$VERSION_FILE" ]; then
    echo "1" > "$VERSION_FILE"
fi
# 监控文件变化函数
monitor_changes() {
    inotifywait -m -r -e modify,create,delete,move "$MONITOR_DIR" |
    while read path action file; do
        # 获取当前版本号
        current_version=$(cat "$VERSION_FILE")
        new_version=$((current_version + 1))
        # 创建备份
        timestamp=$(date +"%Y%m%d_%H%M%S")
        backup_file="$BACKUP_DIR/${file}_v${new_version}_${timestamp}"
        cp "$path$file" "$backup_file"
        # 更新版本号
        echo "$new_version" > "$VERSION_FILE"
        # 记录变更日志
        echo "[$(date +"%Y-%m-%d %H:%M:%S")] $action: $file (Version: $new_version)" >> change_log.txt
    done
}
# 启动监控
monitor_changes

Python版本的监控脚本

#!/usr/bin/env python3
import os
import hashlib
import json
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import shutil
from datetime import datetime
class VersionControlHandler(FileSystemEventHandler):
    def __init__(self, monitor_dir, version_dir):
        self.monitor_dir = monitor_dir
        self.version_dir = version_dir
        self.version_db = os.path.join(version_dir, "version_db.json")
        self.load_version_db()
    def load_version_db(self):
        if os.path.exists(self.version_db):
            with open(self.version_db, 'r') as f:
                self.db = json.load(f)
        else:
            self.db = {}
    def save_version_db(self):
        with open(self.version_db, 'w') as f:
            json.dump(self.db, f, indent=2)
    def get_file_hash(self, filepath):
        """计算文件哈希值"""
        hasher = hashlib.md5()
        with open(filepath, 'rb') as f:
            buf = f.read()
            hasher.update(buf)
        return hasher.hexdigest()
    def create_version(self, filepath):
        """创建文件版本"""
        if not os.path.exists(filepath):
            return
        filename = os.path.basename(filepath)
        rel_path = os.path.relpath(filepath, self.monitor_dir)
        # 获取或初始化版本信息
        if rel_path not in self.db:
            self.db[rel_path] = {
                'versions': [],
                'current_version': 0
            }
        file_hash = self.get_file_hash(filepath)
        # 检查是否发生变化
        if self.db[rel_path]['versions']:
            last_version = self.db[rel_path]['versions'][-1]
            if last_version['hash'] == file_hash:
                return  # 文件未变化，跳过
        # 创建新版本
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        version_num = self.db[rel_path]['current_version'] + 1
        # 备份文件
        version_file = f"{filename}_v{version_num}_{timestamp}"
        version_path = os.path.join(self.version_dir, version_file)
        shutil.copy2(filepath, version_path)
        # 更新数据库
        version_info = {
            'version': version_num,
            'timestamp': timestamp,
            'hash': file_hash,
            'backup_file': version_file
        }
        self.db[rel_path]['versions'].append(version_info)
        self.db[rel_path]['current_version'] = version_num
        self.save_version_db()
        print(f"[{timestamp}] 新版本 {version_num} 已创建: {rel_path}")
    def on_modified(self, event):
        if not event.is_directory:
            self.create_version(event.src_path)
    def on_created(self, event):
        if not event.is_directory:
            self.create_version(event.src_path)
class VersionMonitor:
    def __init__(self, monitor_dir, version_dir):
        self.monitor_dir = monitor_dir
        self.version_dir = version_dir
        self.ensure_directories()
    def ensure_directories(self):
        os.makedirs(self.monitor_dir, exist_ok=True)
        os.makedirs(self.version_dir, exist_ok=True)
    def start_monitoring(self):
        event_handler = VersionControlHandler(self.monitor_dir, self.version_dir)
        observer = Observer()
        observer.schedule(event_handler, self.monitor_dir, recursive=True)
        observer.start()
        try:
            print(f"开始监控: {self.monitor_dir}")
            print(f"版本存储: {self.version_dir}")
            print("按 Ctrl+C 停止监控...")
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            observer.stop()
            print("\n监控已停止")
        observer.join()
# 使用示例
if __name__ == "__main__":
    monitor = VersionMonitor(
        monitor_dir="/path/to/monitor",
        version_dir="/path/to/versions"
    )
    monitor.start_monitoring()

带Git集成的版本控制脚本

#!/bin/bash
# 使用Git进行版本控制
REPO_DIR="/path/to/monitor/repo"
GIT_REMOTE="origin"
BRANCH="main"
# 初始化Git仓库
init_git_repo() {
    if [ ! -d "$REPO_DIR/.git" ]; then
        cd "$REPO_DIR"
        git init
        git add .
        git commit -m "Initial commit"
        echo "Git仓库初始化完成"
    fi
}
# 检查文件变化
check_changes() {
    cd "$REPO_DIR"
    # 添加所有变化
    git add -A
    # 检查是否有变化
    if ! git diff --cached --quiet; then
        timestamp=$(date +"%Y-%m-%d %H:%M:%S")
        file_count=$(git diff --cached --name-only | wc -l)
        # 创建commit
        git commit -m "Auto-save: $timestamp - $file_count files changed"
        # 推送远程（可选）
        if [ -n "$GIT_REMOTE" ]; then
            git push "$GIT_REMOTE" "$BRANCH" 2>/dev/null || echo "推送失败，检查远程配置"
        fi
        echo "[$timestamp] 创建了新版本，包含 $file_count 个文件的更改"
    fi
}
# 监控主循环
monitor_loop() {
    init_git_repo
    echo "开始监控文件变化..."
    echo "按 Ctrl+C 停止监控"
    while true; do
        check_changes
        sleep 60  # 每60秒检查一次
    done
}
# 版本回滚函数
rollback_version() {
    cd "$REPO_DIR"
    if [ -z "$1" ]; then
        echo "请指定要回滚的commit哈希"
        echo "可用版本列表："
        git log --oneline -10
        return 1
    fi
    git revert --no-commit "$1"
    git commit -m "Rollback to version $1"
    echo "已回滚到版本: $1"
}
# 主函数
main() {
    case "$1" in
        start)
            monitor_loop
            ;;
        rollback)
            rollback_version "$2"
            ;;
        history)
            cd "$REPO_DIR"
            git log --oneline --graph
            ;;
        *)
            echo "用法: $0 {start|rollback <commit>|history}"
            echo ""
            echo "选项:"
            echo "  start     启动文件监控"
            echo "  rollback  回滚到指定版本"
            echo "  history   查看版本历史"
            ;;
    esac
}
main "$@"

高级版本：带数据库的版本控制（Python）

#!/usr/bin/env python3
import os
import json
import sqlite3
import hashlib
from datetime import datetime
import difflib
import gzip
class DatabaseVersionControl:
    def __init__(self, db_path, monitor_dir):
        self.db_path = db_path
        self.monitor_dir = monitor_dir
        self.init_database()
    def init_database(self):
        """初始化SQLite数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        # 创建文件表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS files (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                file_path TEXT UNIQUE,
                current_version INTEGER DEFAULT 0,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        # 创建版本表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS versions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                file_id INTEGER,
                version_number INTEGER,
                file_hash TEXT,
                file_size INTEGER,
                content BLOB,
                diff_content TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (file_id) REFERENCES files (id)
            )
        ''')
        # 创建变更日志表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS change_log (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                file_id INTEGER,
                version_number INTEGER,
                change_type TEXT,
                description TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (file_id) REFERENCES files (id)
            )
        ''')
        conn.commit()
        conn.close()
    def get_file_hash(self, filepath):
        """计算文件哈希值"""
        hasher = hashlib.sha256()
        with open(filepath, 'rb') as f:
            for chunk in iter(lambda: f.read(4096), b''):
                hasher.update(chunk)
        return hasher.hexdigest()
    def compress_content(self, content):
        """压缩文件内容"""
        return gzip.compress(content.encode('utf-8'))
    def save_version(self, filepath):
        """保存文件版本"""
        if not os.path.exists(filepath):
            return
        rel_path = os.path.relpath(filepath, self.monitor_dir)
        file_hash = self.get_file_hash(filepath)
        file_size = os.path.getsize(filepath)
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        # 获取或创建文件记录
        cursor.execute('SELECT id, current_version FROM files WHERE file_path = ?', (rel_path,))
        result = cursor.fetchone()
        if result:
            file_id, current_version = result
            new_version = current_version + 1
        else:
            cursor.execute('INSERT INTO files (file_path) VALUES (?)', (rel_path,))
            file_id = cursor.lastrowid
            new_version = 1
        # 读取文件内容
        with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
        # 压缩内容
        compressed_content = self.compress_content(content)
        # 生成diff（如果有前一个版本）
        diff_content = None
        if new_version > 1:
            cursor.execute('''
                SELECT content FROM versions 
                WHERE file_id = ? AND version_number = ?
            ''', (file_id, new_version - 1))
            last_version = cursor.fetchone()
            if last_version:
                last_content = gzip.decompress(last_version[0]).decode('utf-8')
                diff = difflib.unified_diff(
                    last_content.splitlines(keepends=True),
                    content.splitlines(keepends=True),
                    fromfile=f'{rel_path} (v{new_version-1})',
                    tofile=f'{rel_path} (v{new_version})'
                )
                diff_content = ''.join(diff)
        # 保存版本
        cursor.execute('''
            INSERT INTO versions (file_id, version_number, file_hash, file_size, content, diff_content)
            VALUES (?, ?, ?, ?, ?, ?)
        ''', (file_id, new_version, file_hash, file_size, compressed_content, diff_content))
        # 更新文件信息
        cursor.execute('''
            UPDATE files SET current_version = ? WHERE id = ?
        ''', (new_version, file_id))
        # 记录变更日志
        change_type = 'create' if new_version == 1 else 'modify'
        description = f'文件 {rel_path} 创建了新版本 {new_version}'
        cursor.execute('''
            INSERT INTO change_log (file_id, version_number, change_type, description)
            VALUES (?, ?, ?, ?)
        ''', (file_id, new_version, change_type, description))
        conn.commit()
        conn.close()
        print(f"[{datetime.now()}] 版本 {new_version} 已保存: {rel_path}")
    def get_version_history(self, filepath, limit=10):
        """获取文件版本历史"""
        rel_path = os.path.relpath(filepath, self.monitor_dir)
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            SELECT v.version_number, v.file_size, v.file_hash, v.created_at
            FROM versions v
            JOIN files f ON v.file_id = f.id
            WHERE f.file_path = ?
            ORDER BY v.version_number DESC
            LIMIT ?
        ''', (rel_path, limit))
        history = cursor.fetchall()
        conn.close()
        return history
    def compare_versions(self, filepath, version1, version2):
        """比较两个版本"""
        rel_path = os.path.relpath(filepath, self.monitor_dir)
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        # 获取两个版本的内容
        cursor.execute('''
            SELECT v.content, v.version_number
            FROM versions v
            JOIN files f ON v.file_id = f.id
            WHERE f.file_path = ? AND v.version_number IN (?, ?)
            ORDER BY v.version_number
        ''', (rel_path, version1, version2))
        versions = cursor.fetchall()
        conn.close()
        if len(versions) != 2:
            print("无法获取指定版本")
            return
        content1 = gzip.decompress(versions[0][0]).decode('utf-8')
        content2 = gzip.decompress(versions[1][0]).decode('utf-8')
        # 生成差异
        diff = difflib.unified_diff(
            content1.splitlines(keepends=True),
            content2.splitlines(keepends=True),
            fromfile=f'{rel_path} (v{version1})',
            tofile=f'{rel_path} (v{version2})'
        )
        print(f"比较 {rel_path} 版本 {version1} 和 {version2}:")
        print(''.join(diff))
# 使用示例
if __name__ == "__main__":
    vc = DatabaseVersionControl(
        db_path="version_control.db",
        monitor_dir="/path/to/monitor"
    )
    # 保存版本
    vc.save_version("/path/to/monitor/example.txt")
    # 查看历史
    history = vc.get_version_history("/path/to/monitor/example.txt")
    print("版本历史:")
    for v in history:
        print(f"  版本 {v[0]}: 大小={v[1]}字节, 哈希={v[2][:16]}..., 时间={v[3]}")
    # 比较版本
    vc.compare_versions("/path/to/monitor/example.txt", 1, 2)

使用建议

选择合适的方法：
- 简单场景：使用Bash脚本
- 中等复杂度：使用Python脚本
- 需要版本回滚：使用Git集成
- 需要数据分析和查询：使用数据库版本
配置要点：
- 设置合理的监控间隔
- 配置存储空间限制
- 实现旧版本自动清理机制
安全考虑：
- 备份文件权限设置
- 敏感数据加密
- 访问控制

这些脚本可以根据具体需求进行调整和扩展。

本文地址： https://www.bmjup.com/post/16935.html