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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
|
import os
import sys
from logging import Logger, getLogger
from configparser import ConfigParser
from .utils import get_checksum
log: Logger = getLogger(__name__)
# db class that works for both html and md files
class Database:
__COLUMN_NUM: int = 5
def __init__(self, db_path: str,
config: ConfigParser):
log.debug('initializing the page db on path "%s"', db_path)
self.db_path: str = db_path
self.config: ConfigParser = config
self.e: dict[str, tuple[float, float, str, list[str]]] = dict()
# updates the tags for a specific entry (file)
# file_name only contains the entry name (without the absolute path)
def update_tags(self, file_name: str,
tags: list[str]) -> None:
if file_name in self.e:
log.debug('updating tags for entry "%s"', file_name)
cts, mts, checksum, old_tags = self.e[file_name]
log.debug('entry "%s" old content: (%s, %s, %s, (%s))',
file_name, cts, mts, checksum, ', '.join(old_tags))
self.e[file_name] = (cts, mts, checksum, tags)
log.debug('entry "%s" new content: (%s, %s, %s, (%s))',
file_name, cts, mts, checksum, ', '.join(tags))
else:
log.error('can\'t update tags for entry "%s",'
' as it is not present in db', file_name)
sys.exit(1)
# returns a bool that indicates if the entry
# was (includes new entries) or wasn't updated
def update(self, file_name: str,
remove: str='') -> bool:
log.debug('updating entry for file "%s"', file_name)
# initial default values
f: str = file_name
tags: list[str] = []
if remove != '':
f = file_name.replace(remove, '')
log.debug('removed "%s" from "%s": "%s"', remove, file_name, f)
# get current time, needs actual file name
time: float = os.stat(file_name).st_mtime
log.debug('modified time for "%s": %s', file_name, time)
# calculate current checksum, also needs actual file name
checksum: str = get_checksum(file_name)
log.debug('current checksum for "%s": "%s"', file_name, checksum)
# two cases, 1) entry didn't exist,
# 2) entry has been mod and,
# 3) entry hasn't been mod
#1)
if f not in self.e:
log.debug('entry "%s" didn\'t exist, adding with defaults', f)
self.e[f] = (time, 0.0, checksum, tags)
return True
old_time, old_mod_time, old_checksum, tags = self.e[f]
log.debug('entry "%s" old content: (%s, %s, %s, (%s))',
f, old_time, old_mod_time, old_checksum, ', '.join(tags))
# 2)
if checksum != old_checksum:
if old_mod_time == 0.0:
log.debug('entry "%s" has been modified for the first'
' time, updating', f)
else:
log.debug('entry "%s" has been modified, updating', f)
self.e[f] = (old_time, time, checksum, tags)
log.debug('entry "%s" new content: (%s, %s, %s, (%s))',
f, old_time, time, checksum, ', '.join(tags))
return True
# 3)
else:
log.debug('entry "%s" hasn\'t been modified', f)
return False
def write(self) -> None:
log.debug('writing db')
with open(self.db_path, 'w') as file:
for k, v in self.e.items():
log.debug('parsing row for page "%s"', k)
t: str
row: str
if len(v[3]) == 0:
t = '-'
else:
t = ','.join(v[3])
row = f'{k} {v[0]} {v[1]} {v[2]} {t}'
log.debug('writing row: "%s\\n"', row)
file.write(f'{row}\n')
def _db_path_exists(self) -> bool:
log.debug('checking that "%s" exists or is a file', self.db_path)
if not os.path.exists(self.db_path):
log.warning('"%s" doesn\'t exist, will be'
' created once process finishes,'
' ignore if it\'s the first run', self.db_path)
return False
if not os.path.isfile(self.db_path):
log.error('"%s" is not a file"', self.db_path)
sys.exit(1)
return True
def _read_raw(self) -> list[str]:
rows: list[str]
with open(self.db_path, 'r') as file:
rows = file.readlines()
log.debug('db contains %d rows', len(rows))
return rows
def read(self) -> None:
log.debug('reading db')
if not self._db_path_exists():
return
rows: list[str] = self._read_raw()
# l=list of values in entry
log.debug('parsing rows from db')
for it, row in enumerate(rows):
i: int = it + 1
r: str = row.strip()
log.debug('row %d content: "%s"', i, r)
# ignoring type error, as i'm doing the check later
# (file_name, ctimestamp, mtimestamp, checksum, [tags])
cols: tuple[str, float, float, str, list[str]] = tuple(r.split()) # type: ignore
col_num: int = len(cols)
if col_num != self.__COLUMN_NUM:
log.critical('row %d doesn\'t contain %s columns, contains %d'
' columns: "%s"',
i, self.__COLUMN_NUM, col_num, r)
sys.exit(1)
t: list[str]
if cols[4] == '-':
t = []
else:
# ignoring type error, the "check" is done in this whole if/else
t = cols[4].split(',') # type: ignore
log.debug('tag content: (%s)', ', '.join(t))
self.e[cols[0]] = (float(cols[1]), float(cols[2]), cols[3], t)
|
