366 lines
15 KiB
Python
366 lines
15 KiB
Python
import re
|
|
import os
|
|
import ssh
|
|
|
|
|
|
class Git(ssh.SSH):
|
|
projects = {}
|
|
projects_list = []
|
|
remotes = {}
|
|
if_set_local = False
|
|
if_global_init = False
|
|
if_global_init_local = False
|
|
if_base_init = False
|
|
if_fix_project = False
|
|
if_fix_local = False
|
|
if_get_branches = False
|
|
if_get_remote = False
|
|
if_fetch_remote = False
|
|
if_get_project = False
|
|
if_connected = False
|
|
|
|
def __init__(self, hostname, path, user, passwd):
|
|
super().__init__(hostname, path, user, passwd)
|
|
self.base_path = [False]
|
|
self.connect(5)
|
|
self.if_connected = True
|
|
self.local_path = None
|
|
self.fix_cmd = None
|
|
self.fix_name = None
|
|
self.git_user = None
|
|
self.git_email = None
|
|
self.ssh_user = user
|
|
self.ssh_hostname = hostname
|
|
self.passwd = passwd
|
|
self.path = path
|
|
self.branches_server = []
|
|
self.current_branch_server = ""
|
|
|
|
def global_init(self, git_user, git_email):
|
|
self.git_user = git_user
|
|
self.git_email = git_email
|
|
cmd = "git config --global user.email \"{0}\";".format(git_email)
|
|
cmd += "git config --global user.name \"{0}\";".format(git_user)
|
|
stdout, stderr = self.run(cmd)
|
|
if self.check_error(stderr):
|
|
print("Global Setting Successful.")
|
|
self.if_global_init = True
|
|
else:
|
|
print(stderr)
|
|
raise ValueError("Args Error")
|
|
|
|
def base_init(self):
|
|
if self.if_connected:
|
|
stdout, stderr = self.run("ls {}".format(self.path))
|
|
if self.check_error(stderr):
|
|
if "giteasy\n" in stdout:
|
|
print("Directory Already Found.")
|
|
else:
|
|
stdout, stderr = self.run("mkdir {0}/{1}".format(self.path, "giteasy"))
|
|
if self.check_error(stderr):
|
|
print("Create directory.")
|
|
else:
|
|
raise ValueError(stderr)
|
|
self.base_path = [True, "{0}{1}".format(self.path, "giteasy")]
|
|
self.if_base_init = True
|
|
else:
|
|
raise ValueError("Path Error")
|
|
else:
|
|
raise AttributeError("Connect First.")
|
|
|
|
def create_project(self, name):
|
|
if self.if_base_init:
|
|
if os.path.exists("{0}/{1}.git".format(self.base_path[1], name)):
|
|
print("Server Project Already Exist.")
|
|
else:
|
|
stdout, stderr = self.run("mkdir {0}/{1}.git".format(self.base_path[1], name))
|
|
if self.check_error(stderr):
|
|
print("Succeed In Making Directory.")
|
|
self.projects[name+".git"] = {"path": "{0}/{1}.git".format(self.base_path[1], name), "branch": []}
|
|
cmd = "cd {0};".format(self.projects[name+".git"]["path"])
|
|
cmd += "git init --bare;"
|
|
stdout, stderr = self.run(cmd)
|
|
if self.check_error(stderr):
|
|
print("Succeed In Getting Repository.")
|
|
else:
|
|
print(stderr)
|
|
raise ValueError("Target Path Abnormal")
|
|
else:
|
|
raise ValueError("Path Error.")
|
|
else:
|
|
raise AttributeError("Set Base Path First.")
|
|
|
|
def clone_project(self):
|
|
if self.if_set_local and self.if_connected:
|
|
os.chdir(self.local_path)
|
|
os.popen("git clone ssh://{0}@{1}:{2}/{3}".format(self.user, self.hostname, self.base_path[1],
|
|
self.fix_name))
|
|
else:
|
|
raise AttributeError("Connect & Set Local First.")
|
|
|
|
def fix_project_local(self):
|
|
if self.if_set_local and self.if_fix_project:
|
|
os.chdir(os.path.join(self.local_path, self.fix_name))
|
|
self.if_fix_local = True
|
|
else:
|
|
raise AttributeError("Set Local & Fix Project First.")
|
|
|
|
def init_project_local(self, name):
|
|
if self.if_set_local and self.if_fix_project:
|
|
os.chdir(self.local_path)
|
|
try:
|
|
os.mkdir(name)
|
|
except FileExistsError:
|
|
os.chdir(os.path.join(self.local_path, name))
|
|
print("Local Project Already Exist.")
|
|
os.chdir(os.path.join(self.local_path, name))
|
|
cmd = "git init"
|
|
os.popen(cmd)
|
|
readme = open(os.path.join(self.local_path, name,"ReadME.md"),"w")
|
|
readme.write("# Default Project Introduction.\n")
|
|
readme.close()
|
|
cmd = "git add *"
|
|
os.popen(cmd)
|
|
# self.add_remote("origin")
|
|
else:
|
|
raise AttributeError("Set Local & Fix Project First.")
|
|
|
|
def global_init_local(self):
|
|
cmd = "git config --global user.email \"{0}\"".format(self.git_email)
|
|
print(os.popen(cmd).read())
|
|
cmd = "git config --global user.name \"{0}\"".format(self.git_user)
|
|
print(os.popen(cmd).read())
|
|
self.if_global_init = True
|
|
|
|
def update_projects(self):
|
|
if self.base_path[0]:
|
|
self.projects = {}
|
|
stdout, stderr = self.run("ls {0}".format(self.base_path[1]))
|
|
if self.check_error(stderr):
|
|
reform = re.compile("\w+")
|
|
for project in stdout:
|
|
project_name = reform.match(project).string.strip('\n')
|
|
self.projects[project_name] = {"path": "{0}/{1}".format(self.base_path[1], project_name),
|
|
"branch": []}
|
|
self.if_get_project = True
|
|
if self.if_fix_project and self.if_fix_local:
|
|
self.get_branch()
|
|
else:
|
|
raise ValueError("Base Path Abnormal")
|
|
|
|
self.list_projects()
|
|
|
|
else:
|
|
raise UnboundLocalError("Init Base First")
|
|
|
|
def list_projects(self):
|
|
if self.if_get_project:
|
|
self.projects_list = []
|
|
for project in self.projects.keys():
|
|
self.projects_list.append(project)
|
|
else:
|
|
raise AttributeError("Get Project First.")
|
|
|
|
def set_local(self, path):
|
|
if self.if_connected:
|
|
self.remotes = {}
|
|
self.local_path = path
|
|
self.if_set_local = True
|
|
os.chdir(self.local_path)
|
|
else:
|
|
raise AttributeError("Connect First.")
|
|
|
|
def add_remote(self, name="origin"):
|
|
if self.if_base_init and self.if_get_project and self.if_fix_project:
|
|
stdout = os.popen("git remote add {3} ssh://{0}@{1}:{2}/{4}.git".format(self.user, self.hostname,
|
|
self.base_path[1],
|
|
name, self.fix_name)).read()
|
|
self.remotes[name] = {"name": name,
|
|
"url": "ssh://{0}@{1}:{2}/{3}.git".format(self.user, self.hostname, self.base_path[1],
|
|
self.fix_name)}
|
|
|
|
self.if_get_remote = True
|
|
return stdout
|
|
else:
|
|
raise AttributeError("Get Project & Base Init & Fix Project First.")
|
|
|
|
def fetch_remote(self, name):
|
|
if self.if_fix_project and self.if_base_init and self.if_set_local:
|
|
if name in self.remotes.keys():
|
|
os.popen("git fetch {0}".format(name))
|
|
print("git fetch {0}".format(name))
|
|
self.if_fetch_remote = True
|
|
else:
|
|
raise AttributeError("Set Local & Fix Project & Base Init First.")
|
|
|
|
def get_remote(self):
|
|
if self.if_fix_project and self.if_base_init and self.if_set_local:
|
|
ret_code = re.compile(r"[\t, ]")
|
|
for remote in os.popen("git remote -v").readlines():
|
|
results = ret_code.split(remote)
|
|
results = list(results)
|
|
if len(results) >= 1:
|
|
self.remotes = {}
|
|
for item in results:
|
|
if results[0] not in self.remotes.keys():
|
|
self.remotes[results[0]] = {"name": results[0], "url": results[1]}
|
|
self.if_get_remote = True
|
|
else:
|
|
raise AttributeError("Set Local & Fix Project & Base Init First.")
|
|
|
|
def push_remote(self, name, branch):
|
|
if self.if_fix_project and self.if_base_init and self.if_set_local \
|
|
and self.if_get_remote:
|
|
self.get_branch_server()
|
|
if name in self.remotes.keys():
|
|
if branch not in self.branches_server:
|
|
return os.popen("git push -u {0} {1}".format(name, branch)).read()
|
|
else:
|
|
return os.popen("git push {0} {1}".format(name, branch)).read()
|
|
else:
|
|
raise ValueError("Name Abnormal")
|
|
else:
|
|
raise AttributeError("Set Local & Fix Project & Base Init & Get Remote First.")
|
|
|
|
def pull_remote(self, name, branch):
|
|
if self.if_fix_project and self.if_base_init and self.if_set_local \
|
|
and self.if_get_remote:
|
|
self.get_branch_server()
|
|
if name in self.remotes.keys():
|
|
if branch in self.branches_server:
|
|
return os.popen("git pull {0} {1}".format(name, branch)).read()
|
|
else: return "Current Branch '{0}' Not Exist In Server.".format(branch)
|
|
else:
|
|
raise ValueError("Remote Error")
|
|
else:
|
|
raise AttributeError("Set Local & Fix Project & Base Init & Get Remote & Get Branches First.")
|
|
|
|
def get_changes(self):
|
|
if self.if_fix_project and self.if_base_init and self.if_set_local \
|
|
and self.if_get_remote:
|
|
stdout = os.popen("git status")
|
|
modified_ret_code = re.compile("modified:[ ]*[\w,/,.]+")
|
|
new_ret_code = re.compile("new file:[ ]*[\w,/,.]+")
|
|
changed_files = []
|
|
|
|
for line in stdout:
|
|
modified_tmp_ret = modified_ret_code.search(line)
|
|
new_tmp_ret = new_ret_code.search(line)
|
|
if modified_tmp_ret is not None:
|
|
modified_file = modified_tmp_ret.group()
|
|
changed_files.append(modified_file)
|
|
elif new_tmp_ret is not None:
|
|
new_file = new_tmp_ret.group()
|
|
changed_files.append(new_file)
|
|
return changed_files
|
|
|
|
else:
|
|
raise AttributeError("Set Local & Fix Project & Base Init & Get Remote & Get Branches First.")
|
|
|
|
def fix_project(self, name):
|
|
if self.if_get_project and self.if_base_init:
|
|
if name+".git" in self.projects_list:
|
|
self.fix_name = name
|
|
stdout, stderr = self.run("cd {0}".format(self.projects[name+".git"]["path"]))
|
|
if self.check_error(stderr):
|
|
self.fix_cmd = "cd {0}".format(self.projects[name+".git"]["path"]) + ";"
|
|
self.if_fix_project = True
|
|
else:
|
|
raise ValueError("Project Path Abnormal")
|
|
else:
|
|
raise AttributeError("Get Project & Base Init First.")
|
|
|
|
def commit_local(self, message):
|
|
if self.if_set_local and self.if_fix_local and self.if_base_init:
|
|
if self.base_path[0]:
|
|
stdout = os.popen("git commit --message=\"{0}\"".format(message)).read()
|
|
return stdout
|
|
else:
|
|
raise UnboundLocalError("Init Base First")
|
|
else:
|
|
raise AttributeError("Set Local & Fix Local & Base Init First.")
|
|
|
|
def add(self):
|
|
if self.if_set_local and self.if_fix_local and self.if_base_init:
|
|
stdout = os.popen("git add *").read()
|
|
return stdout
|
|
else:
|
|
raise AttributeError("Set Local & Fix Local & Base Init First.")
|
|
|
|
def get_branch(self):
|
|
if self.if_get_project and self.if_base_init and self.if_fix_local:
|
|
stdout = os.popen("git branch").read()
|
|
self.projects[self.fix_name+".git"]["branch"] = []
|
|
reform = re.compile("[ ]+")
|
|
self.projects[self.fix_name + ".git"]["active_branch"] = ""
|
|
lines = []
|
|
str = ""
|
|
for char in stdout:
|
|
if char is '\n':
|
|
lines.append(str)
|
|
str = ""
|
|
else:
|
|
str += char
|
|
|
|
for branch in lines:
|
|
print("Branch:",branch)
|
|
branch_name = reform.split(branch)
|
|
if branch_name[0] == '*':
|
|
self.projects[self.fix_name+".git"]["active_branch"] = branch_name[1]
|
|
self.projects[self.fix_name + ".git"]["branch"].append(branch_name[1])
|
|
else:
|
|
self.projects[self.fix_name + ".git"]["branch"].append(branch_name[1])
|
|
self.if_get_branches = True
|
|
else:
|
|
raise AttributeError("Get Project & Base Init & Fix Local First.")
|
|
|
|
def get_branch_server(self):
|
|
if self.if_connected and self.if_get_project and self.if_fix_project:
|
|
self.branches_server = []
|
|
self.current_branch_server = ""
|
|
stdout, stderr = self.run(self.fix_cmd+"git branch -v")
|
|
ret_code = re.compile(r"[ ]+")
|
|
for branch in stdout:
|
|
branches_info = ret_code.split(branch)
|
|
if branches_info[0] == '*':
|
|
self.current_branch_server = branches_info[1]
|
|
self.branches_server.append(branches_info[1])
|
|
else:
|
|
self.branches_server.append(branches_info[1])
|
|
|
|
else:
|
|
raise AttributeError("Get Project & Base Init & Fix Project First.")
|
|
|
|
def status(self):
|
|
if self.fix_name and self.if_fix_local and self.if_base_init:
|
|
stdout = os.popen("git status").read()
|
|
return stdout
|
|
else:
|
|
raise AttributeError("Fix Project & Base Init & Fix Local First.")
|
|
|
|
def list_branch(self):
|
|
if self.if_get_branches:
|
|
for project in self.projects.items():
|
|
for branch in project[1]["branch"]:
|
|
if branch == project[1]["active_branch"]:
|
|
print("(*)", branch)
|
|
else:
|
|
print(" ", branch)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
compute = Git("compute.bktus.com", "/home/git", "git", "123456")
|
|
compute.global_init(git_user="saturneric", git_email="eric.bktu@gmail.com")
|
|
compute.global_init_local()
|
|
compute.base_init()
|
|
# compute.create_project("lcs")
|
|
compute.update_projects()
|
|
compute.list_projects()
|
|
compute.fix_project("lcs")
|
|
compute.get_branch()
|
|
compute.set_local("C:\\Users\\Saturneric\\Documents\\Code\\")
|
|
# compute.init_project_local("test")
|
|
compute.add_remote()
|
|
compute.get_remote()
|
|
compute.list_branch()
|