1#!/usr/bin/env python3
2
3"""Implementation of Manifest class."""
4
5import logging
6import zipfile
7from enum import Enum
8from datetime import datetime
9
10from chart.common.xml import XMLElement
11from chart.common.path import Path
12from chart.backend.activity import Activity
13from chart.project import SID
14from chart.common.args import ArgumentParser
15import chart.alg.settings
16
17ELEM_MANIFEST = 'manifest'
18ELEM_ACTIVITY = 'activity'
19ELEM_SENSING_START = 'sensing-start'
20ELEM_SENSING_STOP = 'sensing-stop'
21ELEM_JOB_ID = 'job-id'
22ELEM_REVISION = 'revision'
23ELEM_USER = 'user'
24ELEM_FILE = 'file'
25ELEM_HISTORY = 'history'
26ELEM_CREATION = 'creation'
27ELEM_PUBLISH = 'publish'
28ELEM_TIME = 'time'
29
30SCHED_USER = 'scheduler'
31EDITED_REPORT_NAME = '{base}.{{ver}}'.format(base=chart.alg.settings.REPORT_FILENAME)
32
33REV_ELEM = 'elem'
34REV_USER = 'user'
35REV_TIMESTAMP = 'timestamp'
36
37logger = logging.getLogger()
38
39
40class Manifest:
41 """Read and write manifest.xml files."""
42
43 Mode = Enum('Mode', 'READ WRITE UPDATE')
44
45 def __init__(self, path=None, mode=Mode.READ, buff=None, report=None, report_path=None):
46 """Args:
47 `path` (Path): Filename of manifest file to open. Can be a directory containing
48 manifest.xml or a full path. Must already exist unless mode is Mode.WRITE in which
49 case any existing file will be overwritten.
50 `buff` (str): Read from buffer
51 `report` (zipfile): Read manifest from an open report zipfile
52 `report_name` (Path): Open report zipfile and read manifest from it
53
54 If `path`, `buff` and `report` are all None we look for manifest.xml in the current
55 directory (creating it if mode is WRITE).
56 """
57 self.path = None
58
59 if path is not None:
60 if path.is_dir():
61 self.path = path.joinpath(chart.alg.settings.MANIFEST_FILENAME)
62
63 else:
64 self.path = path
65
66 self.elem = XMLElement(filename=self.path)
67
68 elif buff is not None:
69 self.elem = XMLElement(from_text=buff)
70
71 elif report_path is not None:
72 buff = zipfile.ZipFile(str(report)).read(str(chart.alg.settings.MANIFEST_FILENAME))
73 self.elem = XMLElement(from_text=buff)
74
75 elif report is not None:
76 buff = report.read(str(chart.alg.settings.MANIFEST_FILENAME))
77 self.elem = XMLElement(from_text=buff)
78
79 elif mode is Manifest.Mode.WRITE:
80 self.elem = XMLElement(tag=ELEM_MANIFEST)
81 self.path = chart.alg.settings.MANIFEST_FILENAME
82
83 else:
84 raise ValueError('Attempting to create readable manifest with no data source')
85
86 self.mode = mode
87
88 def __del__(self):
89 self.close()
90
91 def close(self):
92 """Save file if writeable."""
93 if self.elem is None:
94 return
95
96 # we could be called in shutdown after class attrs have been destroyed
97 #(!?) weird but it does happen in test_manifest.py when run thru py.test
98 try:
99 write_needed = self.mode in (Manifest.Mode.WRITE, Manifest.Mode.UPDATE)
100 except AttributeError:
101 write_needed = False
102
103 if write_needed:
104 # elif hasattr(self, 'mode') and
105 self.elem.write(self.path, pretty_print=True)
106 self.elem = None
107
108 def get_sid(self):
109 """Retrieve SID from xml element(s)."""
110 return SID.from_xml(self.elem.elem)
111
112 def set_sid(self, sid):
113 """Write SID to xml."""
114 if sid is not None:
115 sid.to_xml(self.elem.elem)
116
117 sid = property(get_sid, set_sid)
118
119 def get_activity(self):
120 """Read activity as Activity object."""
121 name = self.elem.parse_child(ELEM_ACTIVITY, None)
122 if name is None:
123 # legacy report
124 res = None
125
126 else:
127 res = Activity(name)
128
129 return res
130
131 def set_activity(self, activity):
132 """Write the activity name to our node."""
133 if activity is not None:
134 self.elem.add(tag=ELEM_ACTIVITY, text=activity.name)
135
136 activity = property(get_activity, set_activity)
137
138 def get_sensing_start(self):
139 """Retrieve sensing start time."""
140 return self.elem.parse_datetime(ELEM_SENSING_START)
141
142 def set_sensing_start(self, sensing_start):
143 """Set sensing start."""
144 self.elem.add(tag=ELEM_SENSING_START, text=sensing_start)
145
146 sensing_start = property(get_sensing_start, set_sensing_start)
147
148 def get_sensing_stop(self):
149 """Retrieve sensing stop time."""
150 return self.elem.parse_datetime(ELEM_SENSING_STOP)
151
152 def set_sensing_stop(self, sensing_stop):
153 """Set sensing stop."""
154 self.elem.add(tag=ELEM_SENSING_STOP, text=sensing_stop)
155
156 sensing_stop = property(get_sensing_stop, set_sensing_stop)
157
158 def get_job_id(self):
159 """Read report id."""
160 return self.elem.parse_int(ELEM_JOB_ID)
161
162 def set_job_id(self, job_id):
163 """Write job id to our node."""
164 self.elem.add(tag=ELEM_JOB_ID, text=job_id)
165
166 job_id = property(get_job_id, set_job_id)
167
168 def set_creation_info(self, user=None):
169 """Add creation information to the Manifest."""
170 # we are going to create the first revision in the manifest
171 generation = self.get_elem_creation()
172
173 # check if we already have these elements to avoid creating duplicates
174 if self.get_creation_user() is None:
175 generation.add(tag=ELEM_USER, text=user if user is not None else SCHED_USER)
176
177 if self.get_creation_filename() is None:
178 generation.add(tag=ELEM_FILE, text=chart.alg.settings.REPORT_FILENAME)
179
180 if self.get_creation_time() is None:
181 generation.add(tag=ELEM_TIME, text=datetime.now())
182
183 def _get_creation_info(self):
184 """Get creation information from the Manifest.
185 Only used by get_generation_time.
186 """
187 return {ELEM_USER: self.get_creation_user(),
188 ELEM_FILE: self.get_creation_filename(),
189 ELEM_TIME: self.get_creation_time()}
190
191 def get_elem_history(self):
192 """Get history element from Manifest elem tree.
193 Note: If it does not exist, create one (legacy reports do not have it)."""
194 res = self.elem.find(ELEM_HISTORY)
195 if res is not None:
196 return res
197
198 # if no ELEM_HISTORY exists we are dealing with a legacy report, and we update it by
199 # adding a history element tree
200 history = self.elem.add(tag=ELEM_HISTORY)
201 # creation = SubElement(history, ELEM_CREATION)
202 return history
203
204 def get_elem_creation(self):
205 """Get creation element from Manifest elem tree."""
206 history_elem = self.get_elem_history()
207 res = history_elem.find(ELEM_CREATION)
208 if res is not None:
209 return res
210
211 # if no ELEM_CREATION exists we are dealing with a legacy report, and we update it by
212 # adding a creation element
213 creation = history_elem.add(tag=ELEM_CREATION)
214 return creation
215
216 def get_creation_user(self):
217 """Get creation user from the Manifest creation section."""
218 user = self.get_elem_creation().parse_str(ELEM_USER, None)
219 return user
220
221 def get_creation_time(self):
222 """Get creation time from the Manifest creation section."""
223 creation_time = self.get_elem_creation().parse_datetime(ELEM_TIME, None)
224 return creation_time
225
226 def get_creation_filename(self):
227 """Get filename from the Manifest creation section."""
228 filename = self.get_elem_creation().parse_str(ELEM_FILE, None)
229 return filename
230
231 def add_revision(self, username):
232 """Add a new revision to an existing manifest.
233
234 Args:
235 `revision_id` (int): id of the revision to add to the manifest
236 """
237 # we do not have any revisions yet, copy report.html.0 as creation filename
238 latest_rev = self.get_latest_revision()
239 if latest_rev is None:
240 # file_elem = self.get_elem_creation().find(ELEM_FILE)
241 # file_elem.text = EDITED_REPORT_NAME.format(ver=self.revision_count())
242 self.get_elem_creation().set(
243 tag=ELEM_FILE,
244 text=EDITED_REPORT_NAME.format(ver=self.revision_count()))
245
246 else:
247 # if there are existing revisions, the last one will have report filename
248 # set to "report.html". So we patch it to "report.html.x" where
249 # x is the previous max revison number
250 # file_elem = latest_rev[REV_ELEM].find(ELEM_FILE)
251 # file_elem.text = EDITED_REPORT_NAME.format(ver=self.revision_count())
252 latest_rev[REV_ELEM].set(
253 tag=ELEM_FILE,
254 text=EDITED_REPORT_NAME.format(ver=self.revision_count()))
255
256 # now we add the new revision
257 revision = self.get_elem_history().add(tag=ELEM_REVISION)
258 revision.add(tag=ELEM_USER, text=username)
259 revision.add(tag=ELEM_FILE, text=chart.alg.settings.REPORT_FILENAME)
260 revision.add(tag=ELEM_TIME, text=datetime.now())
261
262 # def decode_revisions(self, type=History.REV):
263 def get_revisions(self):
264 """Get a list of all revisions in the Manifest."""
265 revs = []
266 for rev_elem in self.get_elem_history().findall(ELEM_REVISION):
267 revs.append({
268 REV_ELEM: rev_elem,
269 REV_USER: rev_elem.parse_str(ELEM_USER, None),
270 REV_TIMESTAMP: rev_elem.parse_datetime(ELEM_TIME)})
271
272 return sorted(revs, key=lambda k: k[REV_TIMESTAMP])
273
274 def get_latest_revision(self):
275 """Get the latest revision group in the Manifest."""
276 revs = self.get_revisions()
277 if len(revs) > 0:
278 return revs[-1]
279
280 return None
281
282 def revision_count(self):
283 """Get the number of revisions of the current report."""
284 return len(self.get_revisions())
285
286 def add_publish_time(self, publish_time, user):
287 """Add a new publish_time tag to the Manifest.
288
289 Args:
290 `publish_time` (datetime): publishing time
291 """
292 publish = self.get_elem_history().add(tag=ELEM_PUBLISH)
293 publish.add(tag=ELEM_TIME, text=publish_time)
294 publish.add(tag=ELEM_USER, text=user.user_name)
295
296 def get_publish_events(self):
297 """Get a list of all publishing times in the Manifest."""
298 publish_events = []
299 for pub_elem in self.get_elem_history().findall(ELEM_PUBLISH):
300 publish_events.append(
301 {'time': pub_elem.parse_datetime(ELEM_TIME),
302 'user': pub_elem.parse_str(ELEM_USER)})
303
304 return publish_events
305
306 def get_latest_publish_time(self):
307 """Get the most recent publish time in the Manifest."""
308 publish_events = self.get_publish_events()
309 if len(publish_events) > 0:
310 times = [e['time'] for e in publish_events]
311 return max(times)
312
313 return None
314
315 def get_history(self):
316 """Get a sorted list of events (publish, editing) for this report."""
317 history = []
318 # add creation
319 entry = {'action': 'Created', 'time': self.get_creation_time()}
320 history.append(entry)
321
322 # add edit events
323 for rev in self.get_revisions():
324 entry = {'action': 'Edited by {usr}'.format(usr=rev[REV_USER]),
325 'time': rev['timestamp']}
326 history.append(entry)
327
328 # add publish events
329 for pub_event in self.get_publish_events():
330 entry = {'action': 'Published by {usr}'.format(usr=pub_event['user']),
331 'time': pub_event['time']}
332 history.append(entry)
333
334 # sort by time
335 history = sorted(history, key=lambda k: k['time'])
336 return history
337
338 def tostring(self):
339 """Write Manifest contents as a string."""
340 return self.elem.to_str()
341
342
343def main():
344 """Command line entry point."""
345 parser = ArgumentParser(__doc__)
346 parser.add_argument('--filename',
347 help='Manifest file to test')
348
349 args = parser.parse_args()
350
351 from chart.common.log import init_log
352 init_log()
353 if args.filename:
354 manifest = Manifest(path=Path(args.filename), mode=Manifest.Mode.UPDATE)
355 # creation
356 print(manifest.get_creation_time())
357 print(manifest._get_creation_info())
358 print(manifest.get_history())
359 manifest.set_creation_info()
360 print(manifest._get_creation_info())
361 print(manifest.get_history())
362
363 # revisions
364 manifest.add_revision('joseg')
365 print(manifest.revision_count())
366 print(manifest.get_revisions())
367 print(manifest.get_latest_revision())
368
369 # publish times
370 manifest.add_publish_time(datetime.now(), 'testuser')
371 print(manifest.get_publish_events())
372 print(manifest.get_latest_publish_time())
373
374 # print(manifest.tostring())
375 del manifest
376
377 parser.exit()
378
379 parser.error('No actions specified')
380
381
382if __name__ == '__main__':
383 main()