109 lines
2.8 KiB
Python
109 lines
2.8 KiB
Python
from yaml import safe_load as yaml_load
|
|
from yaml import dump as yaml_dump
|
|
from sys import argv
|
|
|
|
def main():
|
|
path = "./testdata/hello_world.yaml"
|
|
if len(argv) > 1:
|
|
path = argv[1]
|
|
doc = Doc.from_file(path)
|
|
print(doc.render())
|
|
print(doc.to_dict())
|
|
|
|
# https://blog.kevinjahns.de/are-crdts-suitable-for-shared-editing/
|
|
class Doc():
|
|
def __init__(self):
|
|
self._nodes = []
|
|
|
|
def from_file(path):
|
|
doc = Doc()
|
|
with open(path, "r") as f:
|
|
d = yaml_load(f)
|
|
doc._nodes = [Node.from_dict(i) for i in d["nodes"]]
|
|
return doc
|
|
|
|
def to_dict(self):
|
|
return yaml_dump(slim_dict({
|
|
"nodes": [node.to_dict() for node in self._nodes],
|
|
}))
|
|
|
|
def render(self):
|
|
result = []
|
|
nodes = [node for node in self._nodes]
|
|
while nodes:
|
|
first = self._first_node(nodes)
|
|
result.append(first)
|
|
nodes = [node for node in nodes if node != first]
|
|
return "".join([i.content() for i in result])
|
|
|
|
def _first_node(self, nodes):
|
|
for potential_root_node in sorted([node for node in nodes if not node.from_id()], key=lambda x: x.id()):
|
|
return potential_root_node
|
|
node_ids = [node.id() for node in nodes]
|
|
for potential_rootest_node in sorted([node for node in nodes if not node.from_id() in node_ids], key=lambda x: x.id()):
|
|
return potential_rootest_node
|
|
raise Exception(nodes)
|
|
|
|
class Node():
|
|
def __init__(self, client_name, ts, content, is_delete, from_id, to_id):
|
|
self._ts = ts
|
|
self._client_name = client_name
|
|
self._content = content
|
|
self._is_delete = is_delete
|
|
self._from_id = from_id
|
|
self._to_id = to_id
|
|
|
|
def __str__(self):
|
|
return str(self.to_dict())
|
|
|
|
def from_dict(d):
|
|
return Node(
|
|
d.get("client_name", ""),
|
|
d.get("ts", 0),
|
|
d.get("content", ""),
|
|
d.get("is_delete", False),
|
|
d.get("from", ""),
|
|
d.get("to", ""),
|
|
)
|
|
|
|
def to_dict(self):
|
|
return slim_dict({
|
|
"client_name": self._client_name,
|
|
"ts": self._ts,
|
|
"content": self._content,
|
|
"is_delete": self._is_delete,
|
|
"from": self._from_id,
|
|
"to": self._to_id,
|
|
})
|
|
|
|
def clone(self):
|
|
return Node(
|
|
self._ts,
|
|
self._client_name,
|
|
self._content,
|
|
self._is_delete,
|
|
self._from_id,
|
|
self._to_id,
|
|
)
|
|
|
|
def id(self):
|
|
return f'{self._ts}/{self._client_name}'
|
|
|
|
def from_id(self):
|
|
return self._from_id
|
|
|
|
def to_id(self):
|
|
return self._to_id
|
|
|
|
def content(self):
|
|
if self._is_delete:
|
|
self._content = ""
|
|
return self._content
|
|
|
|
def slim_dict(d):
|
|
return {k:v for k,v in d.items() if v or (isinstance(v, int) and not isinstance(v, bool))}
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|