-
Notifications
You must be signed in to change notification settings - Fork 0
/
059.py
144 lines (114 loc) Β· 4.13 KB
/
059.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
136
137
138
139
140
141
142
143
144
"""
Problem:
Implement a file syncing algorithm for two computers over a low-bandwidth network.
What if we know the files in the two computers are mostly the same?
"""
from __future__ import annotations
from hashlib import sha256
hash_func = sha256
class MerkleNode:
def __init__(self, name: str) -> None:
self.parent = None
self.node_hash = None
self.name = name
def __hash__(self) -> int:
return hash(self.name)
def __eq__(self, other: MerkleNode) -> bool:
if type(other) != MerkleNode:
return False
return self.node_hash == other.node_hash
class MerkleDirectory(MerkleNode):
def __init__(self, name: str) -> None:
MerkleNode.__init__(self, name)
self.children = set()
self.is_dir = True
# creating a file on directory initialize and calculating hash
new_file = MerkleFile("dir_init")
new_file.add_to_directory(self)
def __repr__(self) -> str:
return f"Name: {self.name}, Children: {self.children}, Hash: {self.node_hash}"
def add_to_directory(self, directory: MerkleDirectory) -> None:
# adding the node
self.parent = directory
directory.children.add(self)
# recalculating hash for all anscestors
while directory is not None:
directory.recalculate_hash()
directory = directory.parent
def recalculate_hash(self) -> None:
# concatinating all hashes and recalculating on the cumulative hash
cumulative_hash = ""
for child in self.children:
cumulative_hash += child.node_hash
self.node_hash = hash_func(cumulative_hash.encode()).hexdigest()
def synchronize(self, other: MerkleDirectory) -> None:
# if the directories have the same hash, they are already synchronized
if self.node_hash == other.node_hash:
return
# updating other using self
for node in self.children:
if not node in other.children:
type(node)(node.add_to_directory(other))
# updating self using other
for node in other.children:
if not node in self.children:
type(node)(node.add_to_directory(self))
class MerkleFile(MerkleNode):
def __init__(self, name: str) -> None:
MerkleNode.__init__(self, name)
self.node_contents = ""
self.is_dir = False
self.node_hash = hash_func(self.node_contents.encode()).hexdigest()
def __repr__(self) -> str:
return (
f"[ Name: {self.name}, Content: "
+ f"{self.node_contents if self.node_contents else 'null'}, "
+ f"Hash: {self.node_hash} ]"
)
def add_to_directory(self, directory: MerkleDirectory) -> None:
# adding the node
self.parent = directory
directory.children.add(self)
# recalculating hash for all anscestors
while directory is not None:
directory.recalculate_hash()
directory = directory.parent
def update_contents(self, new_contents: str) -> None:
# contents updated and hash recaculated
self.node_hash = hash_func(new_contents.encode()).hexdigest()
self.node_contents = new_contents
if self.parent:
self.parent.recalculate_hash()
class Computer:
def __init__(self):
self.root = MerkleDirectory("root")
def __repr__(self) -> str:
return str(self.root)
def synchronize(self, new_comp: Computer) -> None:
print("Syncing computers...")
self.root.synchronize(new_comp.root)
print("Sync successful!\n")
if __name__ == "__main__":
c1 = Computer()
c2 = Computer()
print("COMPUTER 1:")
print(c1)
print("COMPUTER 2:")
print(c2)
print()
new_file = MerkleFile("check_file")
new_file.update_contents("Check")
new_file.add_to_directory(c1.root)
new_dir = MerkleDirectory("check_dir")
new_dir.add_to_directory(c2.root)
print("COMPUTER 1:")
print(c1)
print("COMPUTER 2:")
print(c2)
print()
c1.synchronize(c2)
print("COMPUTER 1:")
print(c1)
print("COMPUTER 2:")
print(c2)
print()