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
|
# -*- coding: utf-8 -*-
# python import
import json, os
from datetime import datetime
# ds reader import
from datastore_reader.utils import _csv, config as c
def parse_meta_file(meta_path):
"""Returns an info dict from matadata content parsed.
:param meta_path: '/somewhere/here/tmp/.../store/file.metadata'
:return meta_dict: {'activity': 'paint', 'timestamp': '1324136292', ...}
>>> datastore_path = './data/tmp/demo/home/olpc/.sugar/default/datastore/store'
>>> meta_path = os.path.join(datastore_path, 'demo.metadata')
>>> parse_meta_file(meta_path)
{u'activity_id': u'ec04b5191d6a0468ff1c23d4233a2e4438517a1a', u'title_set_by_user': u'0', u'uid': u'2f84e066-39d8-4e36-b389-79e430082ca0', u'vid': 2.0, u'title': u'Activit\\xe9 M\\xe9moriser', u'timestamp': 1308067572, u'mtime': u'2011-06-14T16:06:12.760893', u'keep': 0, u'icon-color': u'#00B20D,#FF8F00', u'activity': u'org.laptop.Memorize', u'mime_type': u''}
"""
with open(meta_path, 'rb') as f:
return json.load(f, 'utf-8')
def list_meta_files(datastore_path):
"""Iters the path for the metadata files found in the given datastore
folder.
:param datastore_path: ex.: '/somewhere/here/tmp/home/.../store'
:return meta_path: ex.: ['/somewhere/here/tmp/.../store/file.metadata', ..]
>>> path = './data/tmp/demo/home/olpc/.sugar/default/datastore/store'
>>> [meta_path for meta_path in list_meta_files(path)]
['./data/tmp/demo/home/olpc/.sugar/default/datastore/store/demo.metadata']
"""
for file_name in os.listdir(datastore_path):
name, ext = os.path.splitext(file_name)
if ext == '.metadata':
yield os.path.join(datastore_path, file_name)
def extract_backup(backup_path):
"""Extracts backup archive in the tmp folder and the the corresponding
datastore path.
:param backup: a path, ex.: '/somewhere/here'
:return backup: a path, ex.: '/somewhere/here/tmp/home/.../store'
>>> extract_backup('data/demo.bz2')
'./data/tmp/demo/home/olpc/.sugar/default/datastore/store'
"""
pass
def list_backups(working_dir):
"""Iters serial nb and path tuples according the backup files of the
working dir, ex.: [('serial_1': '/somewhere/here')]
Returns None if no working dir.
:param working_dir: for testing issue otherwise use the value from
the config file.
>>> [(s, p) for s, p in list_backups('data')]
[('demo', 'data/demo.tar.bz2')]
"""
for file_name in os.listdir(working_dir):
name_tar, ext_bz2 = os.path.splitext(file_name)
name, ext_tar = os.path.splitext(name_tar)
if ext_bz2 == '.bz2' and ext_tar == '.tar':
yield (name, os.path.join(working_dir, file_name))
def read_backups():
# working dir check
if not os.path.exists(c.main.working_dir):
raise Exception('no working_dir found!')
# csv path
csv_name = datetime.now().strftime('%Y-%m-%d_%H:%M_moulinette_result.csv')
csv_path = os.path.join(c.main.working_dir, 'out', csv_name)
if not os.path.exists(csv_path):
os.mkdir(csv_path)
# init the csv writer
with open(csv_path, 'wb') as f:
# init csv writer
writer = csv.Writer(f, delimiter=';')
# header line
writer.writerow(c.moulinette.columns)
# list backups
for backup_serial, backup_path in list_backups(c.main.working_dir):
# extract backup
datastore_path = extract_backup(backup_path)
# list metadata
for meta_path in list_meta_files(datastore_path):
# parse the metadata file
meta_dict = parse_meta_file(meta_path)
# add serial do dict
meta_dict['serial'] = backup_serial
# write meta info in a new csv row according config
writer.writerow([meta_dict[k] for k in c.moulinette.columns])
if __name__ == "__main__":
import doctest
doctest.testmod()
|
