基于Python編寫(xiě)一個(gè)簡(jiǎn)單的Gitee數(shù)據(jù)備份工具
一個(gè)用于將本地文件自動(dòng)備份到 Gitee 倉(cāng)庫(kù)的桌面應(yīng)用程序。

功能特性
- 自動(dòng)備份: 支持定時(shí)自動(dòng)備份本地文件到 Gitee 倉(cāng)庫(kù)
- 多任務(wù)管理: 可創(chuàng)建多個(gè)備份任務(wù),每個(gè)任務(wù)獨(dú)立配置
- 系統(tǒng)托盤(pán): 支持最小化到系統(tǒng)托盤(pán),后臺(tái)靜默運(yùn)行
- 命令行模式: 支持 CLI 模式運(yùn)行,可配合 Windows 任務(wù)計(jì)劃程序使用
- 實(shí)時(shí)日志: 顯示備份進(jìn)度和詳細(xì)日志信息
- 文件過(guò)濾: 自動(dòng)跳過(guò)超大文件和常見(jiàn)忽略目錄(.git, node_modules 等)
系統(tǒng)要求
- Windows 10/11
- Python 3.8+ (如需從源碼運(yùn)行)
- Git (需已安裝并添加到系統(tǒng) PATH)
安裝
方式一:使用打包好的 EXE 文件
直接運(yùn)行 GiteeBackup.exe 即可。
方式二:從源碼運(yùn)行
克隆倉(cāng)庫(kù):
git clone <repository-url>
安裝依賴:
pip install -r requirements.txt
注:安裝好git
3. 運(yùn)行程序:
python main.py
使用說(shuō)明
1. 配置 Gitee Access Token
- 登錄 Gitee,進(jìn)入 設(shè)置 -> 私有令牌 -> 生成新令牌
- 勾選所需權(quán)限(至少需要
projects權(quán)限) - 復(fù)制生成的 Token
- 在軟件中填入 Token、用戶名和郵箱,點(diǎn)擊 保存配置
2. 創(chuàng)建備份任務(wù)
- 點(diǎn)擊 添加任務(wù) 按鈕
- 填寫(xiě)任務(wù)信息:
- 任務(wù)名稱(chēng): 便于識(shí)別的名稱(chēng)
- 源路徑: 要備份的本地文件夾路徑
- 倉(cāng)庫(kù)名稱(chēng): Gitee 倉(cāng)庫(kù)名稱(chēng)(如不存在會(huì)自動(dòng)創(chuàng)建)
- 可選:設(shè)置定時(shí)備份計(jì)劃
- 點(diǎn)擊 確定 保存任務(wù)
3. 執(zhí)行備份
- 手動(dòng)備份: 選中任務(wù)后點(diǎn)擊 立即備份
- 定時(shí)備份: 勾選任務(wù)的 啟用定時(shí) 選項(xiàng),程序會(huì)在設(shè)定時(shí)間自動(dòng)執(zhí)行
4. 系統(tǒng)托盤(pán)
- 點(diǎn)擊窗口最小化按鈕,程序會(huì)隱藏到系統(tǒng)托盤(pán)
- 右鍵點(diǎn)擊托盤(pán)圖標(biāo)可以:
- 顯示窗口: 重新打開(kāi)主界面
- 退出程序: 完全關(guān)閉程序
命令行模式
程序支持命令行模式運(yùn)行,適合配合 Windows 任務(wù)計(jì)劃程序使用:
# 運(yùn)行所有啟用了定時(shí)的備份任務(wù) python main.py --cli # 運(yùn)行指定任務(wù) python main.py --cli --task "任務(wù)名稱(chēng)" # 列出所有任務(wù) python main.py --list
配合 Windows 任務(wù)計(jì)劃程序
- 打開(kāi) 任務(wù)計(jì)劃程序
- 創(chuàng)建基本任務(wù)
- 設(shè)置觸發(fā)器(如每天凌晨 2 點(diǎn))
- 操作選擇 啟動(dòng)程序
- 程序路徑填寫(xiě)
GiteeBackup.exe - 參數(shù)填寫(xiě)
--cli
注意事項(xiàng)
- 文件大小限制: 單個(gè)文件最大支持 100MB(Gitee 限制)
- Token 安全: Token 會(huì)保存在本地配置文件中,請(qǐng)妥善保管
- 網(wǎng)絡(luò)連接: 備份過(guò)程需要穩(wěn)定的網(wǎng)絡(luò)連接
- 倉(cāng)庫(kù)權(quán)限: 確保 Token 有創(chuàng)建和管理倉(cāng)庫(kù)的權(quán)限
依賴庫(kù)
requests>=2.28.0 pystray Pillow
源代碼
main.py
import tkinter as tk
from tkinter import ttk, messagebox, filedialog, scrolledtext
import threading
from datetime import datetime
import sys
import argparse
from PIL import Image, ImageDraw
import pystray
import ctypes
from config import ConfigManager, BackupTask
from backup_manager import BackupManager
class GiteeBackupApp:
def __init__(self, root: tk.Tk):
self.root = root
self.root.title("Gitee 數(shù)據(jù)備份工具")
self.root.geometry("900x700")
self.root.minsize(800, 600)
self.config_manager = ConfigManager()
self.backup_manager = BackupManager(self.config_manager)
self.backup_manager.set_progress_callback(self._on_progress)
self.backup_manager.set_status_callback(self._on_status)
self.backup_manager.set_log_callback(self._log)
self._create_ui()
self._load_config()
self.backup_manager.start_scheduler()
self.root.protocol("WM_DELETE_WINDOW", self._on_close)
self.tray_icon = None
self._is_hidden = False
self._create_tray_icon()
self.root.bind("<Unmap>", self._on_unmap)
def _create_ui(self):
self._create_menu()
main_frame = ttk.Frame(self.root, padding="10")
main_frame.pack(fill=tk.BOTH, expand=True)
self._create_settings_panel(main_frame)
self._create_tasks_panel(main_frame)
self._create_log_panel(main_frame)
def _create_menu(self):
menubar = tk.Menu(self.root)
self.root.config(menu=menubar)
file_menu = tk.Menu(menubar, tearoff=0)
menubar.add_cascade(label="文件", menu=file_menu)
file_menu.add_command(label="新建備份任務(wù)", command=self._add_task_dialog)
file_menu.add_separator()
file_menu.add_command(label="隱藏到托盤(pán)", command=self._hide_window)
file_menu.add_separator()
file_menu.add_command(label="退出", command=self._on_close)
help_menu = tk.Menu(menubar, tearoff=0)
menubar.add_cascade(label="幫助", menu=help_menu)
help_menu.add_command(label="關(guān)于", command=self._show_about)
def _create_settings_panel(self, parent):
settings_frame = ttk.LabelFrame(parent, text="Gitee 配置", padding="10")
settings_frame.pack(fill=tk.X, pady=(0, 10))
ttk.Label(settings_frame, text="Access Token:").grid(row=0, column=0, sticky=tk.W, padx=5, pady=5)
self.token_entry = ttk.Entry(settings_frame, width=50, show="*")
self.token_entry.grid(row=0, column=1, padx=5, pady=5)
ttk.Label(settings_frame, text="用戶名:").grid(row=1, column=0, sticky=tk.W, padx=5, pady=5)
self.username_entry = ttk.Entry(settings_frame, width=50)
self.username_entry.grid(row=1, column=1, padx=5, pady=5)
ttk.Label(settings_frame, text="郵箱:").grid(row=2, column=0, sticky=tk.W, padx=5, pady=5)
self.email_entry = ttk.Entry(settings_frame, width=50)
self.email_entry.grid(row=2, column=1, padx=5, pady=5)
btn_frame = ttk.Frame(settings_frame)
btn_frame.grid(row=3, column=0, columnspan=2, pady=10)
ttk.Button(btn_frame, text="保存配置", command=self._save_config).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="測(cè)試連接", command=self._test_connection).pack(side=tk.LEFT, padx=5)
self.connection_status = ttk.Label(btn_frame, text="")
self.connection_status.pack(side=tk.LEFT, padx=10)
def _create_tasks_panel(self, parent):
tasks_frame = ttk.LabelFrame(parent, text="備份任務(wù)", padding="10")
tasks_frame.pack(fill=tk.BOTH, expand=True, pady=(0, 10))
toolbar = ttk.Frame(tasks_frame)
toolbar.pack(fill=tk.X, pady=(0, 5))
ttk.Button(toolbar, text="添加任務(wù)", command=self._add_task_dialog).pack(side=tk.LEFT, padx=2)
ttk.Button(toolbar, text="編輯任務(wù)", command=self._edit_task_dialog).pack(side=tk.LEFT, padx=2)
ttk.Button(toolbar, text="刪除任務(wù)", command=self._delete_task).pack(side=tk.LEFT, padx=2)
ttk.Separator(toolbar, orient=tk.VERTICAL).pack(side=tk.LEFT, fill=tk.Y, padx=10)
ttk.Button(toolbar, text="立即備份", command=self._manual_backup).pack(side=tk.LEFT, padx=2)
columns = ("name", "source", "repo", "schedule", "last_backup", "enabled")
self.tasks_tree = ttk.Treeview(tasks_frame, columns=columns, show="headings", height=8)
self.tasks_tree.heading("name", text="任務(wù)名稱(chēng)")
self.tasks_tree.heading("source", text="源路徑")
self.tasks_tree.heading("repo", text="倉(cāng)庫(kù)名稱(chēng)")
self.tasks_tree.heading("schedule", text="定時(shí)計(jì)劃")
self.tasks_tree.heading("last_backup", text="上次備份")
self.tasks_tree.heading("enabled", text="啟用")
self.tasks_tree.column("name", width=120)
self.tasks_tree.column("source", width=200)
self.tasks_tree.column("repo", width=120)
self.tasks_tree.column("schedule", width=100)
self.tasks_tree.column("last_backup", width=150)
self.tasks_tree.column("enabled", width=60)
scrollbar = ttk.Scrollbar(tasks_frame, orient=tk.VERTICAL, command=self.tasks_tree.yview)
self.tasks_tree.configure(yscrollcommand=scrollbar.set)
self.tasks_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
self._refresh_tasks()
def _create_log_panel(self, parent):
log_frame = ttk.LabelFrame(parent, text="日志", padding="10")
log_frame.pack(fill=tk.BOTH, expand=True)
self.log_text = scrolledtext.ScrolledText(log_frame, height=8, state=tk.DISABLED)
self.log_text.pack(fill=tk.BOTH, expand=True)
self.progress_var = tk.DoubleVar()
self.progress_bar = ttk.Progressbar(log_frame, variable=self.progress_var, maximum=100)
self.progress_bar.pack(fill=tk.X, pady=(5, 0))
self.status_label = ttk.Label(log_frame, text="就緒")
self.status_label.pack(fill=tk.X, pady=(5, 0))
def _load_config(self):
config = self.config_manager.config
self.token_entry.insert(0, config.gitee_token)
self.username_entry.insert(0, config.gitee_username)
self.email_entry.insert(0, config.gitee_email)
def _save_config(self):
token = self.token_entry.get().strip()
username = self.username_entry.get().strip()
email = self.email_entry.get().strip()
if not token:
messagebox.showwarning("警告", "請(qǐng)輸入Access Token")
return
self.backup_manager.set_gitee_token(token, username, email)
self._log("配置已保存")
messagebox.showinfo("成功", "配置已保存")
def _test_connection(self):
token = self.token_entry.get().strip()
if not token:
messagebox.showwarning("警告", "請(qǐng)先輸入Access Token")
return
self.backup_manager.gitee_api = type('GiteeAPI', (), {'get_user_info': lambda: None})()
from gitee_api import GiteeAPI
self.backup_manager.gitee_api = GiteeAPI(token)
def test():
success, result = self.backup_manager.test_connection()
self.root.after(0, lambda: self._on_test_result(success, result))
threading.Thread(target=test, daemon=True).start()
self.connection_status.config(text="測(cè)試中...")
def _on_test_result(self, success: bool, result: str):
if success:
self.connection_status.config(text=f"連接成功: {result}", foreground="green")
self._log(f"連接測(cè)試成功,用戶: {result}")
else:
self.connection_status.config(text=f"連接失敗: {result}", foreground="red")
self._log(f"連接測(cè)試失敗: {result}")
def _refresh_tasks(self):
for item in self.tasks_tree.get_children():
self.tasks_tree.delete(item)
tasks = self.config_manager.get_backup_tasks()
for task in tasks:
schedule = f"{task.schedule_time}" if task.schedule_enabled else "未啟用"
enabled = "是" if task.schedule_enabled else "否"
last_backup = task.last_backup or "從未備份"
self.tasks_tree.insert("", tk.END, values=(
task.name,
task.source_path,
task.repo_name,
schedule,
last_backup,
enabled
))
def _add_task_dialog(self):
self._task_dialog(None)
def _edit_task_dialog(self):
selection = self.tasks_tree.selection()
if not selection:
messagebox.showwarning("警告", "請(qǐng)先選擇一個(gè)任務(wù)")
return
index = self.tasks_tree.index(selection[0])
tasks = self.config_manager.get_backup_tasks()
if 0 <= index < len(tasks):
self._task_dialog(index, tasks[index])
def _task_dialog(self, edit_index: int = None, task: BackupTask = None):
dialog = tk.Toplevel(self.root)
dialog.title("編輯任務(wù)" if edit_index is not None else "添加任務(wù)")
dialog.geometry("500x400")
dialog.transient(self.root)
dialog.grab_set()
frame = ttk.Frame(dialog, padding="20")
frame.pack(fill=tk.BOTH, expand=True)
ttk.Label(frame, text="任務(wù)名稱(chēng):").grid(row=0, column=0, sticky=tk.W, pady=5)
name_entry = ttk.Entry(frame, width=40)
name_entry.grid(row=0, column=1, pady=5)
ttk.Label(frame, text="源路徑:").grid(row=1, column=0, sticky=tk.W, pady=5)
source_frame = ttk.Frame(frame)
source_frame.grid(row=1, column=1, pady=5, sticky=tk.EW)
source_entry = ttk.Entry(source_frame, width=32)
source_entry.pack(side=tk.LEFT)
ttk.Button(source_frame, text="瀏覽...", command=lambda: self._browse_folder(source_entry)).pack(side=tk.LEFT, padx=5)
ttk.Label(frame, text="倉(cāng)庫(kù)名稱(chēng):").grid(row=2, column=0, sticky=tk.W, pady=5)
repo_entry = ttk.Entry(frame, width=40)
repo_entry.grid(row=2, column=1, pady=5)
ttk.Label(frame, text="定時(shí)備份:").grid(row=3, column=0, sticky=tk.W, pady=5)
schedule_frame = ttk.Frame(frame)
schedule_frame.grid(row=3, column=1, pady=5, sticky=tk.W)
enabled_var = tk.BooleanVar()
enabled_check = ttk.Checkbutton(schedule_frame, text="啟用", variable=enabled_var)
enabled_check.pack(side=tk.LEFT)
ttk.Label(schedule_frame, text="時(shí)間:").pack(side=tk.LEFT, padx=(10, 5))
time_entry = ttk.Entry(schedule_frame, width=8)
time_entry.pack(side=tk.LEFT)
time_entry.insert(0, "00:00")
ttk.Label(frame, text="重復(fù)日期:").grid(row=4, column=0, sticky=tk.W, pady=5)
days_frame = ttk.Frame(frame)
days_frame.grid(row=4, column=1, pady=5, sticky=tk.W)
days_vars = []
day_names = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
for i, day in enumerate(day_names):
var = tk.BooleanVar(value=True)
days_vars.append(var)
ttk.Checkbutton(days_frame, text=day, variable=var).pack(side=tk.LEFT)
if task:
name_entry.insert(0, task.name)
source_entry.insert(0, task.source_path)
repo_entry.insert(0, task.repo_name)
enabled_var.set(task.schedule_enabled)
time_entry.delete(0, tk.END)
time_entry.insert(0, task.schedule_time)
for i, var in enumerate(days_vars):
var.set(i in (task.schedule_days or []))
def save():
name = name_entry.get().strip()
source = source_entry.get().strip()
repo = repo_entry.get().strip()
if not name or not source or not repo:
messagebox.showwarning("警告", "請(qǐng)?zhí)顚?xiě)所有必填項(xiàng)")
return
if not repo.replace('-', '').replace('_', '').isalnum():
messagebox.showwarning("警告", "倉(cāng)庫(kù)名稱(chēng)只能包含字母、數(shù)字、-和_")
return
days = [i for i, var in enumerate(days_vars) if var.get()]
new_task = BackupTask(
name=name,
source_path=source,
repo_name=repo,
schedule_enabled=enabled_var.get(),
schedule_time=time_entry.get(),
schedule_days=days,
last_backup=task.last_backup if task else None
)
if edit_index is not None:
self.config_manager.update_backup_task(edit_index, new_task)
else:
self.config_manager.add_backup_task(new_task)
self._refresh_tasks()
self._log(f"任務(wù)已{'更新' if edit_index is not None else '添加'}: {name}")
dialog.destroy()
btn_frame = ttk.Frame(frame)
btn_frame.grid(row=5, column=0, columnspan=2, pady=20)
ttk.Button(btn_frame, text="保存", command=save).pack(side=tk.LEFT, padx=10)
ttk.Button(btn_frame, text="取消", command=dialog.destroy).pack(side=tk.LEFT, padx=10)
def _browse_folder(self, entry: ttk.Entry):
folder = filedialog.askdirectory()
if folder:
entry.delete(0, tk.END)
entry.insert(0, folder)
def _delete_task(self):
selection = self.tasks_tree.selection()
if not selection:
messagebox.showwarning("警告", "請(qǐng)先選擇一個(gè)任務(wù)")
return
if messagebox.askyesno("確認(rèn)", "確定要?jiǎng)h除選中的任務(wù)嗎?"):
index = self.tasks_tree.index(selection[0])
self.config_manager.remove_backup_task(index)
self._refresh_tasks()
self._log("任務(wù)已刪除")
def _manual_backup(self):
selection = self.tasks_tree.selection()
if not selection:
messagebox.showwarning("警告", "請(qǐng)先選擇一個(gè)任務(wù)")
return
index = self.tasks_tree.index(selection[0])
def backup():
self.backup_manager.manual_backup(index)
threading.Thread(target=backup, daemon=True).start()
def _on_progress(self, current: int, total: int, message: str):
if total > 0:
progress = (current / total) * 100
self.root.after(0, lambda: self.progress_var.set(progress))
self.root.after(0, lambda: self.status_label.config(text=message))
def _on_status(self, status: str):
self._log(status)
self.root.after(0, lambda: self.status_label.config(text=status))
self.root.after(0, self._refresh_tasks)
def _log(self, message: str):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_message = f"[{timestamp}] {message}\n"
def append_log():
self.log_text.config(state=tk.NORMAL)
self.log_text.insert(tk.END, log_message)
self.log_text.see(tk.END)
self.log_text.config(state=tk.DISABLED)
self.root.after(0, append_log)
def _show_about(self):
messagebox.showinfo("關(guān)于",
"Gitee 數(shù)據(jù)備份工具\(yùn)n\n"
"版本: 1.0.0\n\n"
"功能:\n"
"- 支持手動(dòng)備份\n"
"- 支持定時(shí)備份\n"
"- 多任務(wù)管理\n"
"- 最小化到系統(tǒng)托盤(pán)\n\n"
"使用Gitee API進(jìn)行數(shù)據(jù)備份"
)
def _create_tray_icon(self):
def create_image():
width = 64
height = 64
color = (70, 130, 180)
image = Image.new('RGBA', (width, height), (0, 0, 0, 0))
dc = ImageDraw.Draw(image)
dc.ellipse([8, 8, 56, 56], fill=color, outline=(50, 100, 150), width=2)
dc.text((14, 18), 'Gitee', fill='white')
dc.text((14, 36), 'Backup', fill='white')
return image
menu = pystray.Menu(
pystray.MenuItem('顯示窗口', self._show_window, default=True),
pystray.Menu.SEPARATOR,
pystray.MenuItem('退出程序', self._on_tray_exit)
)
self.tray_icon = pystray.Icon('gitee-backup', create_image(), 'Gitee 備份工具', menu)
def run_tray():
self.tray_icon.run()
tray_thread = threading.Thread(target=run_tray, daemon=True)
tray_thread.start()
def _on_unmap(self, event):
if event.widget == self.root:
if self.root.state() == 'iconic':
self.root.after(100, self._hide_window)
def _hide_window(self):
self.root.withdraw()
self._is_hidden = True
def _show_window(self, icon=None, item=None):
self.root.after(0, self._do_show_window)
def _do_show_window(self):
self.root.deiconify()
self.root.state('normal')
self.root.lift()
self.root.focus_force()
self._is_hidden = False
def _on_tray_exit(self, icon=None, item=None):
if self.tray_icon:
self.tray_icon.stop()
self.root.after(0, self._do_close)
def _on_close(self):
self._hide_window()
def _do_close(self):
self.backup_manager.stop_scheduler()
self.root.destroy()
def run_cli_backup(task_name: str = None):
print("=" * 50)
print("Gitee 備份工具 - 命令行模式")
print("=" * 50)
config_manager = ConfigManager()
backup_manager = BackupManager(config_manager)
def log_callback(msg):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"[{timestamp}] {msg}")
backup_manager.set_log_callback(log_callback)
tasks = config_manager.get_backup_tasks()
if not tasks:
print("錯(cuò)誤: 沒(méi)有配置任何備份任務(wù)")
return 1
if task_name:
for i, task in enumerate(tasks):
if task.name == task_name:
print(f"執(zhí)行任務(wù): {task.name}")
success = backup_manager.backup_task(task, i)
return 0 if success else 1
print(f"錯(cuò)誤: 未找到任務(wù) '{task_name}'")
return 1
else:
print(f"共有 {len(tasks)} 個(gè)任務(wù)")
success_count = 0
for i, task in enumerate(tasks):
print(f"\n執(zhí)行任務(wù) [{i+1}/{len(tasks)}]: {task.name}")
if backup_manager.backup_task(task, i):
success_count += 1
print(f"\n完成: {success_count}/{len(tasks)} 個(gè)任務(wù)成功")
return 0 if success_count == len(tasks) else 1
def main():
parser = argparse.ArgumentParser(description="Gitee 數(shù)據(jù)備份工具")
parser.add_argument('--cli', action='store_true', help='命令行模式運(yùn)行')
parser.add_argument('--task', type=str, help='指定要執(zhí)行的任務(wù)名稱(chēng)')
parser.add_argument('--all', action='store_true', help='執(zhí)行所有任務(wù)')
parser.add_argument('--list', action='store_true', help='列出所有任務(wù)')
args = parser.parse_args()
if args.list:
config_manager = ConfigManager()
tasks = config_manager.get_backup_tasks()
print("備份任務(wù)列表:")
print("-" * 40)
for i, task in enumerate(tasks):
status = "啟用" if task.schedule_enabled else "禁用"
print(f"{i+1}. {task.name}")
print(f" 源路徑: {task.source_path}")
print(f" 倉(cāng)庫(kù): {task.repo_name}")
print(f" 定時(shí): {status} - {task.schedule_time}")
print(f" 上次備份: {task.last_backup or '從未'}")
print()
return
if args.cli or args.all or args.task:
sys.exit(run_cli_backup(args.task))
root = tk.Tk()
app = GiteeBackupApp(root)
root.mainloop()
if __name__ == "__main__":
main()
config.py
import json
import os
from dataclasses import dataclass, asdict
from typing import Optional, List
from datetime import datetime
CONFIG_FILE = "config.json"
@dataclass
class BackupTask:
name: str
source_path: str
repo_name: str
schedule_enabled: bool = False
schedule_time: str = "00:00"
schedule_days: List[int] = None
last_backup: Optional[str] = None
def __post_init__(self):
if self.schedule_days is None:
self.schedule_days = [0, 1, 2, 3, 4, 5, 6]
@dataclass
class Config:
gitee_token: str = ""
gitee_username: str = ""
gitee_email: str = ""
backup_tasks: List[dict] = None
def __post_init__(self):
if self.backup_tasks is None:
self.backup_tasks = []
class ConfigManager:
def __init__(self, config_dir: str = None):
if config_dir:
self.config_file = os.path.join(config_dir, CONFIG_FILE)
else:
self.config_file = CONFIG_FILE
self.config = self._load_config()
def _load_config(self) -> Config:
if os.path.exists(self.config_file):
try:
with open(self.config_file, 'r', encoding='utf-8') as f:
data = json.load(f)
return Config(**data)
except Exception as e:
print(f"加載配置失敗: {e}")
return Config()
return Config()
def save_config(self):
with open(self.config_file, 'w', encoding='utf-8') as f:
json.dump(asdict(self.config), f, ensure_ascii=False, indent=2)
def set_gitee_credentials(self, token: str, username: str, email: str):
self.config.gitee_token = token
self.config.gitee_username = username
self.config.gitee_email = email
self.save_config()
def add_backup_task(self, task: BackupTask):
self.config.backup_tasks.append(asdict(task))
self.save_config()
def update_backup_task(self, index: int, task: BackupTask):
if 0 <= index < len(self.config.backup_tasks):
self.config.backup_tasks[index] = asdict(task)
self.save_config()
def remove_backup_task(self, index: int):
if 0 <= index < len(self.config.backup_tasks):
self.config.backup_tasks.pop(index)
self.save_config()
def get_backup_tasks(self) -> List[BackupTask]:
return [BackupTask(**task) for task in self.config.backup_tasks]
def update_task_last_backup(self, index: int):
if 0 <= index < len(self.config.backup_tasks):
self.config.backup_tasks[index]['last_backup'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.save_config()
backup_manager.py
import os
import shutil
import subprocess
import tempfile
import threading
import time
from datetime import datetime
from typing import Callable, Optional, List
from config import ConfigManager, BackupTask
from gitee_api import GiteeAPI
class BackupManager:
MAX_FILE_SIZE = 100 * 1024 * 1024
def __init__(self, config_manager: ConfigManager):
self.config_manager = config_manager
self.gitee_api: Optional[GiteeAPI] = None
self.scheduler_thread: Optional[threading.Thread] = None
self.running = False
self.progress_callback: Optional[Callable] = None
self.status_callback: Optional[Callable] = None
self.log_callback: Optional[Callable] = None
self._init_gitee_api()
def _init_gitee_api(self):
token = self.config_manager.config.gitee_token
if token:
self.gitee_api = GiteeAPI(token)
def set_gitee_token(self, token: str, username: str, email: str):
self.config_manager.set_gitee_credentials(token, username, email)
self.gitee_api = GiteeAPI(token)
def set_progress_callback(self, callback: Callable):
self.progress_callback = callback
def set_status_callback(self, callback: Callable):
self.status_callback = callback
def set_log_callback(self, callback: Callable):
self.log_callback = callback
def _report_log(self, message: str):
if self.log_callback:
self.log_callback(message)
def _report_progress(self, current: int, total: int, message: str = ""):
if self.progress_callback:
self.progress_callback(current, total, message)
def _report_status(self, status: str):
if self.status_callback:
self.status_callback(status)
def _run_git_command(self, cmd: List[str], cwd: str) -> tuple:
try:
env = os.environ.copy()
env['GIT_TERMINAL_PROMPT'] = '0'
env['GIT_ASKPASS'] = ''
env['GCM_INTERACTIVE'] = 'never'
result = subprocess.run(
cmd,
cwd=cwd,
capture_output=True,
text=True,
encoding='utf-8',
errors='replace',
env=env
)
return result.returncode == 0, result.stdout, result.stderr
except Exception as e:
return False, "", str(e)
def _get_repo_url(self, owner: str, repo: str) -> str:
token = self.config_manager.config.gitee_token
return f"https://oauth2:{token}@gitee.com/{owner}/{repo}.git"
def _collect_files(self, source_path: str, exclude_patterns: List[str] = None) -> tuple:
files = []
skipped_files = []
exclude_patterns = exclude_patterns or ['.git', '__pycache__', 'node_modules', '.idea', '.vscode']
for root, dirs, filenames in os.walk(source_path):
dirs[:] = [d for d in dirs if d not in exclude_patterns]
for filename in filenames:
filepath = os.path.join(root, filename)
file_size = os.path.getsize(filepath)
if file_size <= self.MAX_FILE_SIZE:
files.append(filepath)
else:
skipped_files.append((filepath, file_size))
return files, skipped_files
def backup_task(self, task: BackupTask, task_index: int = -1) -> bool:
if not self.gitee_api:
self._report_status("錯(cuò)誤: 未配置Gitee Token")
return False
if not os.path.exists(task.source_path):
self._report_status(f"錯(cuò)誤: 源路徑不存在 - {task.source_path}")
return False
owner = self.config_manager.config.gitee_username
if not owner:
self._report_status("錯(cuò)誤: 未配置Gitee用戶名")
return False
temp_dir = None
try:
self._report_status(f"開(kāi)始備份: {task.name}")
self._report_log(f"開(kāi)始備份: {task.name}")
if not self.gitee_api.repo_exists(owner, task.repo_name):
self._report_log(f"創(chuàng)建倉(cāng)庫(kù): {task.repo_name}")
self.gitee_api.create_repo(task.repo_name, f"備份: {task.name}")
temp_dir = tempfile.mkdtemp(prefix="gitee_backup_")
self._report_log(f"工作目錄: {temp_dir}")
repo_url = self._get_repo_url(owner, task.repo_name)
self._report_progress(0, 5, "克隆倉(cāng)庫(kù)...")
self._report_log("克隆倉(cāng)庫(kù)...")
success, stdout, stderr = self._run_git_command(
["git", "clone", repo_url, temp_dir],
temp_dir
)
if not success and not os.path.exists(os.path.join(temp_dir, ".git")):
self._report_log(f"克隆失敗,嘗試初始化新倉(cāng)庫(kù): {stderr}")
self._run_git_command(["git", "init"], temp_dir)
self._run_git_command(["git", "remote", "add", "origin", repo_url], temp_dir)
git_dir = os.path.join(temp_dir, ".git")
if not os.path.exists(git_dir):
self._run_git_command(["git", "init"], temp_dir)
self._run_git_command(["git", "remote", "add", "origin", repo_url], temp_dir)
self._report_progress(1, 5, "清理舊文件...")
self._report_log("清理舊文件...")
for item in os.listdir(temp_dir):
if item == ".git":
continue
item_path = os.path.join(temp_dir, item)
if os.path.isdir(item_path):
shutil.rmtree(item_path)
else:
os.remove(item_path)
self._report_progress(2, 5, "復(fù)制文件...")
files, skipped_files = self._collect_files(task.source_path)
if skipped_files:
self._report_log(f"跳過(guò) {len(skipped_files)} 個(gè)超大文件 (>100MB):")
for filepath, size in skipped_files[:5]:
size_mb = size / (1024 * 1024)
self._report_log(f" - {filepath} ({size_mb:.1f}MB)")
if len(skipped_files) > 5:
self._report_log(f" ... 還有 {len(skipped_files) - 5} 個(gè)文件")
self._report_log(f"復(fù)制 {len(files)} 個(gè)文件...")
copied = 0
for filepath in files:
relative_path = os.path.relpath(filepath, task.source_path)
dest_path = os.path.join(temp_dir, relative_path)
os.makedirs(os.path.dirname(dest_path), exist_ok=True)
shutil.copy2(filepath, dest_path)
copied += 1
if copied % 50 == 0:
self._report_progress(2, 5, f"復(fù)制文件 {copied}/{len(files)}")
self._report_progress(3, 5, "配置Git...")
username = self.config_manager.config.gitee_username or "backup"
email = self.config_manager.config.gitee_email or "backup@local"
self._run_git_command(["git", "config", "user.name", username], temp_dir)
self._run_git_command(["git", "config", "user.email", email], temp_dir)
self._run_git_command(["git", "branch", "-M", "master"], temp_dir)
self._report_progress(4, 5, "提交更改...")
self._report_log("添加文件到暫存區(qū)...")
self._run_git_command(["git", "add", "-A"], temp_dir)
commit_msg = f"備份: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
success, stdout, stderr = self._run_git_command(
["git", "commit", "-m", commit_msg],
temp_dir
)
if not success:
if "nothing to commit" in stderr or "nothing to commit" in stdout:
self._report_log("沒(méi)有變更,跳過(guò)推送")
self._report_status("備份完成: 無(wú)變更")
return True
self._report_log(f"提交警告: {stderr}")
self._report_progress(5, 5, "推送到遠(yuǎn)程...")
self._report_log("推送到 Gitee...")
success, stdout, stderr = self._run_git_command(
["git", "branch", "--show-current"],
temp_dir
)
current_branch = stdout.strip() if success else "master"
if not current_branch:
current_branch = "master"
self._report_log(f"當(dāng)前分支: {current_branch}")
success, stdout, stderr = self._run_git_command(
["git", "push", "-u", "origin", current_branch, "--force"],
temp_dir
)
if not success:
self._report_log(f"推送 {current_branch} 失敗,嘗試 main...")
success, stdout, stderr = self._run_git_command(
["git", "push", "-u", "origin", "main", "--force"],
temp_dir
)
if not success:
self._report_log(f"推送失敗: {stderr}")
self._report_status(f"備份失敗: {stderr}")
return False
if task_index >= 0:
self.config_manager.update_task_last_backup(task_index)
self._report_log(f"備份完成: {task.name} ({len(files)} 個(gè)文件)")
self._report_status(f"備份完成: {task.name}")
return True
except Exception as e:
error_msg = f"備份失敗: {str(e)}"
self._report_log(error_msg)
self._report_status(error_msg)
return False
finally:
if temp_dir and os.path.exists(temp_dir):
try:
shutil.rmtree(temp_dir)
except:
pass
def start_scheduler(self):
if self.scheduler_thread and self.scheduler_thread.is_alive():
return
self.running = True
self.scheduler_thread = threading.Thread(target=self._scheduler_loop, daemon=True)
self.scheduler_thread.start()
def stop_scheduler(self):
self.running = False
if self.scheduler_thread:
self.scheduler_thread.join(timeout=5)
def _scheduler_loop(self):
while self.running:
now = datetime.now()
tasks = self.config_manager.get_backup_tasks()
for i, task in enumerate(tasks):
if not task.schedule_enabled:
continue
try:
schedule_hour, schedule_minute = map(int, task.schedule_time.split(':'))
if (now.weekday() in task.schedule_days and
now.hour == schedule_hour and
now.minute == schedule_minute):
last_backup = task.last_backup
if last_backup:
last_dt = datetime.strptime(last_backup, "%Y-%m-%d %H:%M:%S")
if (now - last_dt).total_seconds() < 3600:
continue
self.backup_task(task, i)
except Exception as e:
print(f"定時(shí)任務(wù)執(zhí)行失敗: {e}")
time.sleep(60)
def manual_backup(self, task_index: int) -> bool:
tasks = self.config_manager.get_backup_tasks()
if 0 <= task_index < len(tasks):
return self.backup_task(tasks[task_index], task_index)
return False
def test_connection(self) -> tuple:
if not self.gitee_api:
return False, "未配置Gitee API"
try:
user_info = self.gitee_api.get_user_info()
return True, user_info.get('login', 'Unknown')
except Exception as e:
return False, str(e)
gitee_api.py
import requests
import base64
import os
from typing import Optional, List, Dict
class GiteeAPI:
BASE_URL = "https://gitee.com/api/v5"
def __init__(self, token: str):
self.token = token
self.session = requests.Session()
self.session.params = {'access_token': token}
def _request(self, method: str, endpoint: str, **kwargs) -> Dict:
url = f"{self.BASE_URL}/{endpoint}"
if 'params' not in kwargs:
kwargs['params'] = {}
kwargs['params']['access_token'] = self.token
response = self.session.request(method, url, **kwargs)
if response.status_code >= 400:
try:
error_data = response.json()
error_msg = error_data.get('message', error_data.get('error', response.text))
except:
error_msg = response.text or f"HTTP {response.status_code}"
raise Exception(f"HTTP {response.status_code}: {error_msg}")
response.raise_for_status()
return response.json()
def get_user_info(self) -> Dict:
return self._request('GET', 'user')
def get_repos(self, username: str = None) -> List[Dict]:
if username:
return self._request('GET', f'users/{username}/repos')
return self._request('GET', 'user/repos')
def create_repo(self, name: str, description: str = "", private: bool = True) -> Dict:
data = {
'name': name,
'description': description,
'private': private,
'has_issues': False,
'has_wiki': False,
'auto_init': True
}
return self._request('POST', 'user/repos', json=data)
def get_repo(self, owner: str, repo: str) -> Dict:
return self._request('GET', f'repos/{owner}/{repo}')
def delete_repo(self, owner: str, repo: str) -> bool:
try:
self._request('DELETE', f'repos/{owner}/{repo}')
return True
except Exception:
return False
def list_contents(self, owner: str, repo: str, path: str = '') -> List[Dict]:
return self._request('GET', f'repos/{owner}/{repo}/contents/{path}')
def get_file_content(self, owner: str, repo: str, path: str) -> Dict:
return self._request('GET', f'repos/{owner}/{repo}/contents/{path}')
def create_file(self, owner: str, repo: str, path: str, content: str,
message: str = "Add file") -> Dict:
encoded_content = base64.b64encode(content.encode('utf-8')).decode('utf-8')
data = {
'content': encoded_content,
'message': message
}
return self._request('POST', f'repos/{owner}/{repo}/contents/{path}', json=data)
def update_file(self, owner: str, repo: str, path: str, content: str,
sha: str, message: str = "Update file") -> Dict:
encoded_content = base64.b64encode(content.encode('utf-8')).decode('utf-8')
data = {
'content': encoded_content,
'message': message,
'sha': sha
}
return self._request('PUT', f'repos/{owner}/{repo}/contents/{path}', json=data)
def delete_file(self, owner: str, repo: str, path: str, sha: str,
message: str = "Delete file") -> Dict:
data = {
'message': message,
'sha': sha
}
return self._request('DELETE', f'repos/{owner}/{repo}/contents/{path}', json=data)
def upload_file(self, owner: str, repo: str, local_path: str,
remote_path: str, message: str = None) -> Dict:
with open(local_path, 'rb') as f:
content = f.read()
encoded_content = base64.b64encode(content).decode('utf-8')
if message is None:
message = f"Upload {os.path.basename(local_path)}"
try:
existing = self.get_file_content(owner, repo, remote_path)
sha = existing['sha']
data = {
'content': encoded_content,
'message': message,
'sha': sha
}
return self._request('PUT', f'repos/{owner}/{repo}/contents/{remote_path}', json=data)
except Exception:
data = {
'content': encoded_content,
'message': message
}
return self._request('POST', f'repos/{owner}/{repo}/contents/{remote_path}', json=data)
def create_or_update_file(self, owner: str, repo: str, path: str,
content: bytes, message: str) -> Dict:
encoded_content = base64.b64encode(content).decode('utf-8')
try:
existing = self.get_file_content(owner, repo, path)
sha = existing['sha']
data = {
'content': encoded_content,
'message': message,
'sha': sha
}
return self._request('PUT', f'repos/{owner}/{repo}/contents/{path}', json=data)
except Exception:
data = {
'content': encoded_content,
'message': message
}
return self._request('POST', f'repos/{owner}/{repo}/contents/{path}', json=data)
def repo_exists(self, owner: str, repo: str) -> bool:
try:
self.get_repo(owner, repo)
return True
except Exception:
return False
requirements.txt
requests>=2.28.0 pystray Pillow
build.bat
@echo off
chcp 65001 >nul
echo 正在打包 Gitee 備份工具...
echo.
pyinstaller --onefile ^
--windowed ^
--name "GiteeBackup" ^
--icon=NONE ^
--add-data "config.json;." ^
--hidden-import=requests ^
--hidden-import=urllib3 ^
--hidden-import=charset_normalizer ^
--hidden-import=certifi ^
--collect-all requests ^
main.py
echo.
if exist "dist\GiteeBackup.exe" (
echo 打包成功!
echo 輸出文件: dist\GiteeBackup.exe
) else (
echo 打包失敗!
)
pause到此這篇關(guān)于基于Python編寫(xiě)一個(gè)簡(jiǎn)單的Gitee數(shù)據(jù)備份工具的文章就介紹到這了,更多相關(guān)Python Gitee數(shù)據(jù)備份內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Scrapy-Redis結(jié)合POST請(qǐng)求獲取數(shù)據(jù)的方法示例
這篇文章主要給大家介紹了關(guān)于Scrapy-Redis結(jié)合POST請(qǐng)求獲取數(shù)據(jù)的相關(guān)資料,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家學(xué)習(xí)或者使用Scrapy-Redis具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2019-05-05
Python包管理工具pip警告WARNING:?Ignoring?invalid?distribution問(wèn)題解
pip警告通常是由于包安裝不完整、損壞或拼寫(xiě)錯(cuò)誤引起的,這篇文章主要介紹了Python包管理工具pip警告WARNING:?Ignoring?invalid?distribution問(wèn)題解決的相關(guān)資料,文中通過(guò)代碼介紹的非常詳細(xì),需要的朋友可以參考下2025-11-11
Python multiprocessing 進(jìn)程間通信方式實(shí)現(xiàn)
本文主要介紹了Python multiprocessing 進(jìn)程間通信方式實(shí)現(xiàn),文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2023-02-02
使用Python進(jìn)行數(shù)據(jù)庫(kù)交互的實(shí)踐指南(從SQL查詢到ORM操作)
在現(xiàn)代應(yīng)用開(kāi)發(fā)中,數(shù)據(jù)庫(kù)交互是核心技能之一,無(wú)論是用戶數(shù)據(jù)存儲(chǔ)、內(nèi)容管理還是交易記錄,幾乎每個(gè)應(yīng)用都需要與數(shù)據(jù)庫(kù)打交道,本文將帶你從零開(kāi)始,系統(tǒng)學(xué)習(xí)SQL查詢、ORM操作,并教你如何識(shí)別潛在錯(cuò)誤和防范SQL注入攻擊,需要的朋友可以參考下2025-07-07
python數(shù)據(jù)類(lèi)型可變與不可變深入分析
這篇文章主要為大家介紹了python數(shù)據(jù)類(lèi)型可變與不可變深入分析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2022-06-06
淺談PySpark SQL 相關(guān)知識(shí)介紹
這篇文章主要介紹了淺談PySpark SQL 相關(guān)知識(shí)介紹,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2019-06-06
使用Python腳本來(lái)控制Windows Azure的簡(jiǎn)單教程
這篇文章主要介紹了使用Python腳本來(lái)控制Windows Azure的簡(jiǎn)單教程,由于微軟官方提供了Python SDK,使得用戶自己用Python控制Azure成為了可能,需要的朋友可以參考下2015-04-04
python正則表達(dá)式re.match()匹配多個(gè)字符方法的實(shí)現(xiàn)
這篇文章主要介紹了python正則表達(dá)式re.match()匹配多個(gè)字符方法的實(shí)現(xiàn),文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2021-01-01

