/
sync.py
135 lines (112 loc) · 4.85 KB
/
sync.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
# -*- encoding: utf-8 -*-
# kb v0.1.6
# A knowledge base organizer
# Copyright © 2020, gnc.
# See /LICENSE for licensing information.
"""
kb export command module
:Copyright: © 2020, gnc.
:License: GPLv3 (see /LICENSE).
"""
import time
from typing import Dict
import kb.filesystem as fs
import git
def sync(args: Dict[str, str], config: Dict[str, str]):
"""
Synchronize the knowledge base with a remote repository.
Arguments:
args: - a dictionary containing the following fields:
file -> a string representing the wished output
filename
config: - a configuration dictionary containing at least
the following keys:
PATH_KB - the main path of KB
"""
operation = args["operation"]
if operation == "init":
if is_local_git_repo_initialized(config["PATH_KB_GIT"]):
print("Warning: a git repository already exists...")
cancel_repo = input("Do you want to re-init your repository ? [Type YES in that case] ")
if cancel_repo == "YES":
fs.remove_directory(config["PATH_KB_GIT"])
git_init(config["PATH_KB"])
else:
print("Maybe you wanted to type \"kb sync push\" or \"kb sync pull\"!")
else:
git_init(config["PATH_KB"])
elif operation == "push":
if not is_local_git_repo_initialized(config["PATH_KB_GIT"]):
print("Warning: You did not initialize any repository...")
print("try: kb sync init")
return
git_push(config["PATH_KB"])
elif operation == "pull":
if not is_local_git_repo_initialized(config["PATH_KB_GIT"]):
print("Warning: no local git repository has been instantiated!")
cancel_repo = input("Do you want to remove possible local kb files and sync from a remote repository? [Type YES in that case] ")
if cancel_repo == "YES":
try:
fs.remove_directory(config["PATH_KB"])
except:
pass
git_clone(config["PATH_KB"])
else:
git_pull(config["PATH_KB_GIT"])
elif operation == "info":
if not is_local_git_repo_initialized(config["PATH_KB_GIT"]):
print("No local git repository has been instantiated!")
return
show_info(config["PATH_KB"])
else:
print("Error: No valid operation has been specified")
return
def is_local_git_repo_initialized(git_path):
if fs.is_directory(git_path):
return True
def git_push(repo_path):
try:
kb_repo = git.Repo(repo_path)
kb_repo.git.add('--all')
timestamp = time.strftime("%d/%m/%Y-%H:%M:%S")
kb_repo.index.commit("kb synchronization {ts}".format(ts=timestamp))
kb_repo.remote(name='origin').push(refspec='{}:{}'.format("main", "main"))
print("Repository correctly synchronized to remote!")
except:
print('Some error occured while pushing the code')
print('Check your internet connection or the existence of the remote repository')
def git_pull(repo_path):
try:
kb_repo = git.Repo(repo_path)
origin = kb_repo.remotes.origin
origin.pull(origin.refs[0].remote_head)
print("Repository correctly synchronized from remote!")
except:
print('Some error occured while pulling the code')
print('Check your internet connection or the existence of the remote repository')
def git_clone(repo_path):
remote_repo_url = input("Insert the URL of the remote repo (e.g., https://github/user/mykb): ")
if remote_repo_url:
git.Repo.clone_from(remote_repo_url, repo_path)
print("Knowledge base correctly pulled from remote!")
else:
print("Error: Check your internet connection or provide a valid repository")
def git_init(repo_path):
print("Create a remote empty repository on github/gitlab or other git provider...")
remote_repo_url = input("Insert the URL of the created empty remote repo (e.g., https://github/user/mykb): ")
if remote_repo_url:
local_repo = git.Repo.init(repo_path)
remote = local_repo.create_remote("origin", url=remote_repo_url)
local_repo.git.add('--all')
timestamp = time.strftime("%d/%m/%Y-%H:%M:%S")
local_repo.index.commit("kb synchronization {ts}".format(ts=timestamp))
print("Initialization with remote may take time...")
print("Please provide remote credentials and wait...")
remote.push(refspec='{}:{}'.format("main", "main"))
print("Remote repository correctly initialized!")
else:
print("Error: Provide a valid remote URL")
def show_info(repo_path):
repo = git.cmd.Git(repo_path)
print("Remote repository:")
print(repo.execute(["git", "remote", "-v"]))