本文目录导读:

在Python中恢复备份数据的方法取决于你的备份格式,以下是几种常见场景的完整解决方案:
JSON格式备份恢复(最常用)
import json
import os
def restore_from_json(backup_file="backup.json"):
"""
从JSON文件恢复数据
"""
if not os.path.exists(backup_file):
print(f"备份文件 {backup_file} 不存在")
return None
try:
with open(backup_file, 'r', encoding='utf-8') as f:
data = json.load(f)
print(f"成功从 {backup_file} 恢复数据")
return data
except json.JSONDecodeError:
print("JSON文件格式错误")
return None
except Exception as e:
print(f"恢复失败: {e}")
return None
# 使用示例
recovered_data = restore_from_json("my_backup.json")
if recovered_data:
print(recovered_data)
pickle序列化恢复
import pickle
import os
def restore_from_pickle(backup_file="backup.pkl"):
"""
从pickle文件恢复数据(可恢复Python对象)
"""
if not os.path.exists(backup_file):
print(f"备份文件 {backup_file} 不存在")
return None
try:
with open(backup_file, 'rb') as f:
data = pickle.load(f)
print(f"成功从 {backup_file} 恢复数据")
return data
except pickle.UnpicklingError:
print("Pickle文件损坏")
return None
except Exception as e:
print(f"恢复失败: {e}")
return None
# 使用示例
recovered_data = restore_from_pickle("data_backup.pkl")
SQLite数据库恢复
import sqlite3
import os
def restore_sqlite_backup(db_path="mydb_backup.db"):
"""
从SQLite备份文件恢复
"""
if not os.path.exists(db_path):
print(f"数据库文件 {db_path} 不存在")
return False
try:
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# 获取所有表
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = cursor.fetchall()
print(f"成功恢复数据库,包含 {len(tables)} 个表:")
for table in tables:
cursor.execute(f"SELECT COUNT(*) FROM {table[0]};")
count = cursor.fetchone()[0]
print(f" - {table[0]}: {count} 条记录")
conn.close()
return True
except Exception as e:
print(f"恢复失败: {e}")
return False
# 使用示例
restore_sqlite_backup("backup.db")
CSV文件恢复
import pandas as pd
import os
def restore_from_csv(backup_file="backup.csv"):
"""
从CSV文件恢复数据
"""
if not os.path.exists(backup_file):
print(f"备份文件 {backup_file} 不存在")
return None
try:
df = pd.read_csv(backup_file)
print(f"成功从 {backup_file} 恢复数据")
print(f"数据维度: {df.shape}")
return df
except Exception as e:
print(f"恢复失败: {e}")
return None
# 使用示例
df = restore_from_csv("data_backup.csv")
完整备份恢复系统示例
import json
import pickle
import os
from datetime import datetime
class BackupRestoreSystem:
def __init__(self, backup_dir="backups"):
self.backup_dir = backup_dir
os.makedirs(backup_dir, exist_ok=True)
def create_backup(self, data, name="data"):
"""
创建备份
"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{name}_backup_{timestamp}.json"
filepath = os.path.join(self.backup_dir, filename)
try:
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"备份已创建: {filepath}")
return filepath
except Exception as e:
print(f"创建备份失败: {e}")
return None
def restore_latest(self, name="data"):
"""
恢复最新的备份
"""
backups = [f for f in os.listdir(self.backup_dir)
if f.startswith(name) and f.endswith('.json')]
if not backups:
print(f"未找到 {name} 的备份文件")
return None
# 按修改时间排序
backups.sort(key=lambda x: os.path.getmtime(
os.path.join(self.backup_dir, x)), reverse=True)
latest_backup = backups[0]
filepath = os.path.join(self.backup_dir, latest_backup)
try:
with open(filepath, 'r', encoding='utf-8') as f:
data = json.load(f)
print(f"成功从最新备份恢复: {latest_backup}")
return data
except Exception as e:
print(f"恢复失败: {e}")
return None
def list_backups(self, name="data"):
"""
列出所有备份
"""
backups = [f for f in os.listdir(self.backup_dir)
if f.startswith(name) and f.endswith('.json')]
if not backups:
print("没有找到备份文件")
return []
print(f"找到 {len(backups)} 个备份文件:")
for i, backup in enumerate(backups, 1):
filepath = os.path.join(self.backup_dir, backup)
size = os.path.getsize(filepath)
mtime = datetime.fromtimestamp(os.path.getmtime(filepath))
print(f"{i}. {backup} ({size/1024:.2f} KB, {mtime})")
return backups
def restore_specific(self, backup_name):
"""
恢复指定备份
"""
filepath = os.path.join(self.backup_dir, backup_name)
if not os.path.exists(filepath):
print(f"备份文件 {backup_name} 不存在")
return None
try:
with open(filepath, 'r', encoding='utf-8') as f:
data = json.load(f)
print(f"成功从 {backup_name} 恢复数据")
return data
except Exception as e:
print(f"恢复失败: {e}")
return None
# 使用示例
def main():
# 创建备份恢复系统
system = BackupRestoreSystem()
# 示例数据
data = {
"users": [
{"id": 1, "name": "张三"},
{"id": 2, "name": "李四"}
],
"settings": {
"theme": "dark",
"language": "zh-CN"
}
}
# 创建备份
system.create_backup(data, "my_app_data")
# 恢复最新备份
recovered = system.restore_latest("my_app_data")
if recovered:
print("恢复的数据:", recovered)
# 列出所有备份
system.list_backups("my_app_data")
if __name__ == "__main__":
main()
关键注意事项:
- 验证备份文件完整性:恢复前检查文件是否存在且完整
- 处理异常:捕获可能的错误(文件损坏、格式错误等)
- 编码问题:JSON文件使用
utf-8编码 - 版本兼容性:确保恢复的数据格式与当前程序兼容
- 错误处理:总是提供友好的错误信息
选择哪种方式取决于你的数据结构和需求:
- JSON:人类可读,适合配置文件
- pickle:可保存Python对象,但可能不安全
- SQLite:适合数据库应用
- CSV:适合表格数据
记得备份文件要定期更新,并保存在安全的位置!