本文目录导读：
基础文件版本控制脚本（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集成
- 需要数据分析和查询：使用数据库版本
-
配置要点：
- 设置合理的监控间隔
- 配置存储空间限制
- 实现旧版本自动清理机制
-
安全考虑：
- 备份文件权限设置
- 敏感数据加密
- 访问控制
这些脚本可以根据具体需求进行调整和扩展。
