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
145
146
147
148
149
|
from pathlib import Path
import json
from enum import Enum
import hashlib
import os
from .writer import Writer
def add_or_create(dictio, key, value):
if key in dictio:
if value not in dictio[key]:
dictio[key].append(value)
else:
dictio[key] = [value]
class FileState(Enum):
Unused = 1
Owned = 2
Used = 3
links_to_path: Path = Path()
def can_write(self) -> bool:
return self in [FileState.Unused, FileState.Owned]
def links_to(self) -> str:
if self is not FileState.Owned:
raise Exception(f"Cannot call location on {self}")
return self.links_to_path
def link_intact(self) -> bool:
return Path(self.links_to()).exists()
def check_string(self, dotdir: Path) -> str:
if self == FileState.Unused:
return "Free"
elif self == FileState.Owned:
status = "Intact" if self.link_intact() else "Broken"
return f"Owned by {self.owner(dotdir)} ({status})"
else:
return "Occupied"
def owner(self, dotdir: Path) -> str:
to = Path(self.links_to())
return to.relative_to(dotdir).parts[0]
@staticmethod
def create_owned(links_to: Path) -> "FileState":
s = FileState.Owned
s.links_to_path = links_to
return s
@staticmethod
def check_location(path: Path) -> "FileState":
if path.is_symlink():
dest = Path(os.readlink(str(path)))
if Path.cwd() in dest.parents:
return FileState.create_owned(dest)
if not path.exists():
return FileState.Unused
return FileState.Used
class StateFile:
links = {}
dirs = {}
saved = []
attr_to_save = ["links", "applydir", "dirs", "saved"]
def __init__(self, applydir):
# Generate unique string for each possible applydir
ustr = hashlib.md5(applydir.encode("utf-8")).hexdigest()[10:]
self.applydir = str(applydir)
self.statefile = Path(f"state_{ustr}.json")
if self.statefile.exists():
with self.statefile.open("r") as f:
self.set_from_dict(json.load(f))
else:
self.set_from_dict({})
self.stateclean = True
def set_from_dict(self, state):
self.links = state.get("links", {})
self.dirs = state.get("dirs", {})
self.applydir = state.get("applydir", self.applydir)
self.saved = state.get("saved", [])
def save_to_dict(self):
all_attr = self.__dict__
res = {}
for key in self.attr_to_save:
res[key] = all_attr[key]
return res
def dump_state(self, dry_run):
if dry_run:
return
with self.statefile.open("w") as f:
json.dump(self.save_to_dict(), f)
def add_dir(self, path: Path, modulename: str):
# Add to state
add_or_create(self.dirs, str(path), modulename)
def add_link(self, dest, modulename):
self.links[str(dest)] = modulename
def add_saved_module(self, modulename):
if modulename not in self.saved:
self.saved.append(modulename)
def is_saved(self, modulename):
return modulename in self.saved
def remove_saved_module(self, modulename):
if modulename in self.saved:
self.saved.remove(modulename)
def __delete_dir(self, writer: Writer, path: str, module):
self.dirs[path].remove(module)
if len(self.dirs[path]) == 0:
writer.remove(Path(path))
def __delete_link(self, writer: Writer, path: str):
del self.links[path]
if FileState.check_location(Path(path)) == FileState.Owned:
writer.delete(Path(path))
def remove_by_condition(self, cond, writer: Writer):
writer.attach_hook(self.dump_state)
links = list(self.links.keys())
for link in links:
if cond(self.links[link]):
self.__delete_link(writer, link)
dirs = list(self.dirs.keys())
for directory in dirs:
modules = self.dirs[directory]
for module in modules:
if cond(module):
self.__delete_dir(writer, directory, module)
|
