590 lines
20 KiB
Python
590 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
Git 自动同步工具
|
|
用法: python gitrun.py [命令]
|
|
|
|
命令说明:
|
|
push - 强制推送本地到远程（直接覆盖，保留历史记录）
|
|
pull - 强制拉取远程到本地（直接覆盖，保留历史记录）
|
|
sync - 智能同步（先拉取合并，再推送）
|
|
log - 查看提交历史并支持快速回退版本
|
|
history - 查看完整提交历史
|
|
config - 配置远程仓库地址
|
|
|
|
|
|
|
|
|
|
重要说明:
|
|
• push/pull 都会保留完整的 Git 历史记录，可以随时回退
|
|
• push 是纯粹的本地覆盖远程，不会先拉取远程内容
|
|
• pull 是纯粹的远程覆盖本地，不会保留本地未推送的修改
|
|
• sync 才会进行智能合并
|
|
• log 命令可以查看历史并直接回退版本（输入数字选择，0或回车跳过）
|
|
• 所有历史提交都保存在 .git 目录中，永不丢失
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import subprocess
|
|
import json
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
CONFIG_FILE = Path.home() / '.gitrun_config.json'
|
|
|
|
|
|
class GitSyncTool:
|
|
def __init__(self):
|
|
self.config = self.load_config()
|
|
self.repo_path = os.getcwd()
|
|
|
|
def load_config(self):
|
|
"""加载配置文件"""
|
|
if CONFIG_FILE.exists():
|
|
with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
return {}
|
|
|
|
def save_config(self):
|
|
"""保存配置文件"""
|
|
with open(CONFIG_FILE, 'w', encoding='utf-8') as f:
|
|
json.dump(self.config, f, indent=2, ensure_ascii=False)
|
|
|
|
def run_command(self, cmd, check=True, silent=False):
|
|
"""执行命令并返回结果"""
|
|
try:
|
|
# Windows 上使用 GBK 编码，其他系统使用 UTF-8
|
|
encoding = 'gbk' if sys.platform == 'win32' else 'utf-8'
|
|
|
|
result = subprocess.run(
|
|
cmd,
|
|
shell=True,
|
|
cwd=self.repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding=encoding,
|
|
errors='ignore' # 忽略无法解码的字符
|
|
)
|
|
if check and result.returncode != 0:
|
|
if not silent:
|
|
print(f"❌ 命令执行失败: {cmd}")
|
|
if result.stderr:
|
|
print(f"错误信息: {result.stderr}")
|
|
return None
|
|
return result
|
|
except Exception as e:
|
|
if not silent:
|
|
print(f"❌ 执行命令时出错: {e}")
|
|
return None
|
|
|
|
def is_git_repo(self):
|
|
"""检查当前目录是否是 Git 仓库"""
|
|
result = self.run_command("git rev-parse --is-inside-work-tree", check=False, silent=True)
|
|
return result and result.returncode == 0
|
|
|
|
def get_current_branch(self):
|
|
"""获取当前分支名"""
|
|
result = self.run_command("git branch --show-current", silent=True)
|
|
return result.stdout.strip() if result else None
|
|
|
|
def has_changes(self):
|
|
"""检查是否有未提交的更改"""
|
|
result = self.run_command("git status --porcelain", silent=True)
|
|
return bool(result and result.stdout.strip())
|
|
|
|
def get_remote_url(self):
|
|
"""获取远程仓库地址"""
|
|
result = self.run_command("git remote get-url origin", check=False, silent=True)
|
|
return result.stdout.strip() if result and result.returncode == 0 else None
|
|
|
|
def config_remote(self):
|
|
"""配置远程仓库地址"""
|
|
print("🔧 配置远程仓库地址")
|
|
|
|
# 检查是否是 Git 仓库，如果不是则初始化
|
|
if not self.is_git_repo():
|
|
print("⚠️ 当前目录不是 Git 仓库")
|
|
init = input("是否初始化为 Git 仓库? (y/n): ").strip().lower()
|
|
if init == 'y':
|
|
result = self.run_command("git init")
|
|
if result:
|
|
print("✅ Git 仓库初始化成功")
|
|
else:
|
|
print("❌ Git 仓库初始化失败")
|
|
return
|
|
else:
|
|
print("❌ 操作已取消")
|
|
return
|
|
|
|
current_url = self.get_remote_url()
|
|
if current_url:
|
|
print(f"当前远程仓库: {current_url}")
|
|
change = input("是否修改? (y/n): ").strip().lower()
|
|
if change != 'y':
|
|
return
|
|
|
|
remote_url = input("请输入远程仓库地址 (如: https://github.com/username/repo.git): ").strip()
|
|
|
|
if not remote_url:
|
|
print("❌ 远程仓库地址不能为空")
|
|
return
|
|
|
|
# 检查是否已有 origin
|
|
if current_url:
|
|
result = self.run_command(f"git remote set-url origin {remote_url}")
|
|
else:
|
|
result = self.run_command(f"git remote add origin {remote_url}")
|
|
|
|
if result:
|
|
print(f"✅ 远程仓库配置成功: {remote_url}")
|
|
|
|
# 保存到配置文件
|
|
repo_name = os.path.basename(self.repo_path)
|
|
if repo_name not in self.config:
|
|
self.config[repo_name] = {}
|
|
self.config[repo_name]['remote_url'] = remote_url
|
|
self.save_config()
|
|
|
|
# 询问是否设置默认分支
|
|
print("\n💡 提示: 首次推送前需要设置默认分支")
|
|
set_branch = input("是否设置默认分支为 main? (y/n): ").strip().lower()
|
|
if set_branch == 'y':
|
|
# 检查当前分支
|
|
current_branch = self.get_current_branch()
|
|
if not current_branch:
|
|
# 创建初始提交
|
|
print("\n📝 创建初始提交...")
|
|
self.run_command("git add .")
|
|
self.run_command('git commit -m "Initial commit" --allow-empty')
|
|
self.run_command("git branch -M main")
|
|
print("✅ 默认分支已设置为 main")
|
|
elif current_branch != 'main':
|
|
self.run_command("git branch -M main")
|
|
print("✅ 分支已重命名为 main")
|
|
else:
|
|
print("✅ 当前分支已是 main")
|
|
|
|
def commit_changes(self, message=None):
|
|
"""提交本地更改"""
|
|
if not self.has_changes():
|
|
print("ℹ️ 没有需要提交的更改")
|
|
return True
|
|
|
|
print("\n📝 检测到以下更改:")
|
|
self.run_command("git status --short")
|
|
|
|
if message is None:
|
|
message = input("\n请输入提交信息: ").strip()
|
|
|
|
if not message:
|
|
print("❌ 提交信息不能为空")
|
|
return False
|
|
|
|
# 添加所有更改
|
|
print("\n📦 正在暂存所有更改...")
|
|
result = self.run_command("git add -A")
|
|
if not result:
|
|
return False
|
|
|
|
# 提交更改
|
|
print(f"💾 正在提交: {message}")
|
|
result = self.run_command(f'git commit -m "{message}"')
|
|
if result:
|
|
print(f"✅ 提交成功")
|
|
return True
|
|
return False
|
|
|
|
def force_push(self):
|
|
"""纯粹的本地覆盖远程（不拉取，直接强制推送）"""
|
|
print("🚀 强制推送本地到远程")
|
|
|
|
if not self.is_git_repo():
|
|
print("❌ 当前目录不是 Git 仓库")
|
|
return
|
|
|
|
branch = self.get_current_branch()
|
|
if not branch:
|
|
print("❌ 无法获取当前分支")
|
|
return
|
|
|
|
print(f"📍 当前分支: {branch}")
|
|
|
|
# 检查远程仓库
|
|
remote_url = self.get_remote_url()
|
|
if not remote_url:
|
|
print("⚠️ 未配置远程仓库")
|
|
self.config_remote()
|
|
remote_url = self.get_remote_url()
|
|
if not remote_url:
|
|
return
|
|
|
|
print(f"🌐 远程仓库: {remote_url}")
|
|
|
|
# 提交本地更改
|
|
print()
|
|
if not self.commit_changes(): # 改为交互式提交，不自动
|
|
print("❌ 提交失败，推送中止")
|
|
return
|
|
|
|
# 强制推送（不拉取远程）
|
|
print(f"\n⬆️ 正在强制推送到远程 ({branch})...")
|
|
result = self.run_command(f"git push -f origin {branch}")
|
|
|
|
if result:
|
|
print(f"\n✅ 推送成功!")
|
|
else:
|
|
print("\n❌ 推送失败")
|
|
|
|
def force_pull(self):
|
|
"""纯粹的远程覆盖本地（不保留本地未推送的修改）"""
|
|
print("⬇️ 强制拉取远程到本地")
|
|
|
|
if not self.is_git_repo():
|
|
print("❌ 当前目录不是 Git 仓库")
|
|
return
|
|
|
|
branch = self.get_current_branch()
|
|
if not branch:
|
|
print("❌ 无法获取当前分支")
|
|
return
|
|
|
|
print(f"📍 当前分支: {branch}")
|
|
|
|
# 检查远程仓库
|
|
remote_url = self.get_remote_url()
|
|
if not remote_url:
|
|
print("⚠️ 未配置远程仓库")
|
|
self.config_remote()
|
|
return
|
|
|
|
print(f"🌐 远程仓库: {remote_url}")
|
|
|
|
# 检查本地更改
|
|
if self.has_changes():
|
|
print("\n⚠️ 检测到未提交的本地更改，这些更改将会被丢弃")
|
|
self.run_command("git status --short")
|
|
|
|
# 先获取远程最新信息
|
|
print(f"\n📡 正在获取远程仓库信息...")
|
|
self.run_command("git fetch origin")
|
|
|
|
# 强制重置到远程版本
|
|
print(f"⬇️ 正在强制拉取远程 ({branch})...")
|
|
self.run_command(f"git reset --hard origin/{branch}")
|
|
self.run_command("git clean -fd")
|
|
|
|
print(f"\n✅ 拉取成功!")
|
|
|
|
def smart_sync(self):
|
|
"""智能同步（先拉取合并，再推送）"""
|
|
print("🔄 智能同步")
|
|
|
|
if not self.is_git_repo():
|
|
print("❌ 当前目录不是 Git 仓库")
|
|
return
|
|
|
|
branch = self.get_current_branch()
|
|
if not branch:
|
|
print("❌ 无法获取当前分支")
|
|
return
|
|
|
|
print(f"📍 当前分支: {branch}")
|
|
|
|
# 检查远程仓库
|
|
remote_url = self.get_remote_url()
|
|
if not remote_url:
|
|
print("⚠️ 未配置远程仓库")
|
|
self.config_remote()
|
|
return
|
|
|
|
print(f"🌐 远程仓库: {remote_url}")
|
|
|
|
# 提交本地更改
|
|
print()
|
|
if self.has_changes():
|
|
if not self.commit_changes(): # 改为交互式提交
|
|
print("❌ 提交失败")
|
|
return
|
|
|
|
# 拉取远程更改
|
|
print(f"\n⬇️ 正在拉取远程更新 ({branch})...")
|
|
result = self.run_command(f"git pull origin {branch}", check=False)
|
|
|
|
if result and result.returncode != 0:
|
|
# 如果有冲突
|
|
if "conflict" in result.stderr.lower() or "conflict" in result.stdout.lower():
|
|
print("\n⚠️ 检测到合并冲突!")
|
|
print("\n冲突文件:")
|
|
self.run_command("git diff --name-only --diff-filter=U")
|
|
|
|
choice = input("\n选择操作:\n1. 保留本地版本\n2. 保留远程版本\n3. 手动解决后继续\n请选择 (1/2/3): ").strip()
|
|
|
|
if choice == '1':
|
|
self.run_command("git checkout --ours .")
|
|
self.run_command("git add -A")
|
|
self.run_command('git commit --no-edit')
|
|
print("✅ 已保留本地版本")
|
|
elif choice == '2':
|
|
self.run_command("git checkout --theirs .")
|
|
self.run_command("git add -A")
|
|
self.run_command('git commit --no-edit')
|
|
print("✅ 已保留远程版本")
|
|
else:
|
|
print("ℹ️ 请手动解决冲突后执行: git add . && git commit")
|
|
return
|
|
else:
|
|
print("❌ 拉取失败")
|
|
return
|
|
else:
|
|
print("✅ 拉取成功")
|
|
|
|
# 推送到远程
|
|
print(f"\n⬆️ 正在推送到远程 ({branch})...")
|
|
push_result = self.run_command(f"git push origin {branch}")
|
|
|
|
if push_result:
|
|
print(f"\n✅ 同步成功!")
|
|
else:
|
|
print("⚠️ 推送失败或无新内容需要推送")
|
|
|
|
def show_log(self, num=10):
|
|
"""查看提交历史并支持直接回退"""
|
|
print(f"📋 最近 {num} 次提交历史")
|
|
|
|
if not self.is_git_repo():
|
|
print("❌ 当前目录不是 Git 仓库")
|
|
return
|
|
|
|
branch = self.get_current_branch()
|
|
print(f"📍 当前分支: {branch}\n")
|
|
|
|
# 使用 git log 的简单格式，然后解析
|
|
result = self.run_command(f'git log -{num} --format="COMMIT_START%nHASH:%H%nSUBJECT:%s%nAUTHOR:%an%nDATE:%ar%nCOMMIT_END"')
|
|
|
|
if not result:
|
|
return
|
|
|
|
commits = result.stdout.strip().split('COMMIT_START')
|
|
commit_hashes = []
|
|
idx = 1
|
|
|
|
for commit in commits:
|
|
if not commit.strip():
|
|
continue
|
|
|
|
lines = commit.strip().split('\n')
|
|
commit_info = {}
|
|
for line in lines:
|
|
if ':' in line:
|
|
key, value = line.split(':', 1)
|
|
commit_info[key] = value.strip()
|
|
|
|
if 'HASH' in commit_info:
|
|
hash_full = commit_info['HASH']
|
|
hash_short = hash_full[:7]
|
|
subject = commit_info.get('SUBJECT', '')
|
|
date = commit_info.get('DATE', '')
|
|
|
|
commit_hashes.append(hash_full)
|
|
print(f"{idx}. [{hash_short}] {subject} -- {date}")
|
|
idx += 1
|
|
|
|
# 询问是否回退
|
|
print()
|
|
choice = input("是否回退版本? (输入数字1-10回退，直接回车或输入0跳过): ").strip()
|
|
|
|
if not choice or choice == '0':
|
|
print("✅ 未进行回退")
|
|
return
|
|
|
|
# 判断是数字还是哈希值
|
|
if choice.isdigit():
|
|
idx = int(choice) - 1
|
|
if 0 <= idx < len(commit_hashes):
|
|
commit_hash = commit_hashes[idx][:7]
|
|
else:
|
|
print(f"❌ 无效的数字，请输入 1-{len(commit_hashes)}")
|
|
return
|
|
else:
|
|
commit_hash = choice
|
|
|
|
# 验证提交是否存在
|
|
verify = self.run_command(f"git cat-file -t {commit_hash}", check=False, silent=True)
|
|
if not verify or verify.returncode != 0:
|
|
print(f"❌ 提交 {commit_hash} 不存在")
|
|
return
|
|
|
|
# 显示目标提交信息
|
|
print(f"\n📍 即将回退到: {commit_hash}")
|
|
|
|
# 选择回退方式
|
|
print("\n回退方式:")
|
|
print("1. 软回退 (保留工作区和暂存区的更改)")
|
|
print("2. 混合回退 (保留工作区更改，清空暂存区) [推荐]")
|
|
print("3. 硬回退 (完全回退，丢弃所有更改)")
|
|
|
|
mode_choice = input("\n请选择 (1/2/3，直接回车默认选2): ").strip()
|
|
|
|
if not mode_choice:
|
|
mode_choice = '2'
|
|
|
|
mode_map = {
|
|
'1': ('--soft', '软回退'),
|
|
'2': ('--mixed', '混合回退'),
|
|
'3': ('--hard', '硬回退')
|
|
}
|
|
|
|
if mode_choice not in mode_map:
|
|
print("❌ 无效选择")
|
|
return
|
|
|
|
mode, mode_name = mode_map[mode_choice]
|
|
|
|
# 执行回退
|
|
print(f"\n⏮️ 正在执行{mode_name}...")
|
|
result = self.run_command(f"git reset {mode} {commit_hash}")
|
|
|
|
if result:
|
|
print(f"\n✅ {mode_name}成功!")
|
|
print(f"📍 已回退到: {commit_hash}\n")
|
|
self.run_command("git status --short")
|
|
else:
|
|
print(f"\n❌ 回退失败")
|
|
|
|
def show_full_history(self):
|
|
"""查看完整提交历史"""
|
|
print("📚 完整提交历史")
|
|
|
|
if not self.is_git_repo():
|
|
print("❌ 当前目录不是 Git 仓库")
|
|
return
|
|
|
|
branch = self.get_current_branch()
|
|
print(f"📍 当前分支: {branch}\n")
|
|
|
|
# 显示完整历史（简化格式以兼容 Windows）
|
|
self.run_command("git log --graph --oneline --decorate --all")
|
|
|
|
def rollback(self):
|
|
"""回退到指定版本"""
|
|
print("⏮️ 版本回退")
|
|
|
|
if not self.is_git_repo():
|
|
print("❌ 当前目录不是 Git 仓库")
|
|
return
|
|
|
|
# 显示最近10次提交
|
|
print("📋 最近10次提交:\n")
|
|
result = self.run_command("git log -10 --pretty=format:%H|%s|%an|%ar")
|
|
|
|
if not result:
|
|
return
|
|
|
|
lines = result.stdout.strip().split('\n')
|
|
commits = []
|
|
|
|
for i, line in enumerate(lines, 1):
|
|
if '|' in line:
|
|
hash_full, subject, author, date = line.split('|', 3)
|
|
hash_short = hash_full[:7]
|
|
commits.append(hash_full)
|
|
print(f"{i}. [{hash_short}] {subject}")
|
|
print(f" 作者: {author} | 时间: {date}\n")
|
|
|
|
print()
|
|
|
|
# 获取用户输入
|
|
choice = input("请输入要回退到的版本 (输入数字1-10，或输入提交哈希值，q 取消): ").strip()
|
|
|
|
if not choice or choice.lower() == 'q':
|
|
print("❌ 操作已取消")
|
|
return
|
|
|
|
# 判断是数字还是哈希值
|
|
if choice.isdigit():
|
|
idx = int(choice) - 1
|
|
if 0 <= idx < len(commits):
|
|
commit_hash = commits[idx][:7]
|
|
else:
|
|
print(f"❌ 无效的数字，请输入 1-{len(commits)}")
|
|
return
|
|
else:
|
|
commit_hash = choice
|
|
|
|
# 验证提交是否存在
|
|
verify = self.run_command(f"git cat-file -t {commit_hash}", check=False, silent=True)
|
|
if not verify or verify.returncode != 0:
|
|
print(f"❌ 提交 {commit_hash} 不存在")
|
|
return
|
|
|
|
# 显示目标提交信息
|
|
print(f"\n📍 即将回退到:")
|
|
self.run_command(f"git log -1 {commit_hash} --pretty=format:[%h] %s%n作者: %an | 时间: %ar")
|
|
|
|
# 选择回退方式
|
|
print("\n\n回退方式:")
|
|
print("1. 软回退 (保留工作区和暂存区的更改)")
|
|
print("2. 混合回退 (保留工作区更改，清空暂存区) [推荐]")
|
|
print("3. 硬回退 (完全回退，丢弃所有更改)")
|
|
|
|
mode_choice = input("\n请选择 (1/2/3，输入 q 取消): ").strip()
|
|
|
|
if mode_choice.lower() == 'q':
|
|
print("❌ 操作已取消")
|
|
return
|
|
|
|
mode_map = {
|
|
'1': ('--soft', '软回退'),
|
|
'2': ('--mixed', '混合回退'),
|
|
'3': ('--hard', '硬回退')
|
|
}
|
|
|
|
if mode_choice not in mode_map:
|
|
print("❌ 无效选择")
|
|
return
|
|
|
|
mode, mode_name = mode_map[mode_choice]
|
|
|
|
# 执行回退
|
|
print(f"\n⏮️ 正在执行{mode_name}...")
|
|
result = self.run_command(f"git reset {mode} {commit_hash}")
|
|
|
|
if result:
|
|
print(f"\n✅ {mode_name}成功!")
|
|
print(f"📍 已回退到: {commit_hash}\n")
|
|
self.run_command("git status --short")
|
|
else:
|
|
print(f"\n❌ 回退失败")
|
|
|
|
|
|
def main():
|
|
tool = GitSyncTool()
|
|
|
|
if len(sys.argv) < 2:
|
|
print(__doc__)
|
|
return
|
|
|
|
command = sys.argv[1].lower()
|
|
|
|
if command == 'push':
|
|
tool.force_push()
|
|
elif command == 'pull':
|
|
tool.force_pull()
|
|
elif command == 'sync':
|
|
tool.smart_sync()
|
|
elif command == 'log':
|
|
tool.show_log(5)
|
|
elif command == 'history':
|
|
tool.show_full_history()
|
|
elif command == 'rollback':
|
|
tool.rollback()
|
|
elif command == 'config':
|
|
tool.config_remote()
|
|
else:
|
|
print(f"❌ 未知命令: {command}")
|
|
print(__doc__)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main() |