Project

General

Profile

1 53 aaronmk
#!/usr/bin/env python
2
# Maps one datasource to another, using a map spreadsheet if needed
3 986 aaronmk
# Exit status is the # of errors in the import, up to the maximum exit status
4 53 aaronmk
# For outputting an XML file to a PostgreSQL database, use the general format of
5
# http://vegbank.org/vegdocs/xml/vegbank_example_ver1.0.2.xml
6
7 1014 aaronmk
import csv
8 53 aaronmk
import os.path
9
import sys
10 299 aaronmk
import xml.dom.minidom as minidom
11 1701 aaronmk
import xml.parsers.expat as expat
12 53 aaronmk
13 266 aaronmk
sys.path.append(os.path.dirname(__file__)+"/../lib")
14 53 aaronmk
15 1389 aaronmk
import csvs
16 344 aaronmk
import exc
17 1705 aaronmk
import maps
18 64 aaronmk
import opts
19 281 aaronmk
import Parser
20 982 aaronmk
import profiling
21 131 aaronmk
import sql
22 715 aaronmk
import strings
23 828 aaronmk
import term
24 310 aaronmk
import util
25 1014 aaronmk
import xpath
26 133 aaronmk
import xml_dom
27 86 aaronmk
import xml_func
28 53 aaronmk
29 1404 aaronmk
def get_with_prefix(map_, prefixes, key):
30 1484 aaronmk
    '''Gets all entries for the given key with any of the given prefixes'''
31
    values = []
32 1681 aaronmk
    for key_ in strings.with_prefixes(['']+prefixes, key): # also with no prefix
33
        try: value = map_[key_]
34 1484 aaronmk
        except KeyError, e: continue # keep going
35
        values.append(value)
36
37
    if values != []: return values
38
    else: raise e # re-raise last KeyError
39 1404 aaronmk
40 1018 aaronmk
def metadata_value(name): return None # this feature has been removed
41 84 aaronmk
42 1360 aaronmk
def cleanup(val):
43
    if val == None: return val
44 1374 aaronmk
    return util.none_if(strings.cleanup(strings.ustr(val)), u'', u'\\N')
45 1360 aaronmk
46 847 aaronmk
def main_():
47 131 aaronmk
    env_names = []
48
    def usage_err():
49 944 aaronmk
        raise SystemExit('Usage: '+opts.env_usage(env_names, True)+' '
50 847 aaronmk
            +sys.argv[0]+' [map_path...] [<input] [>output]')
51 838 aaronmk
52 1570 aaronmk
    ## Get config from env vars
53 838 aaronmk
54 1570 aaronmk
    # Modes
55 946 aaronmk
    test = opts.env_flag('test', False, env_names)
56 947 aaronmk
    commit = opts.env_flag('commit', False, env_names) and not test
57 944 aaronmk
        # never commit in test mode
58 947 aaronmk
    redo = opts.env_flag('redo', test, env_names) and not commit
59
        # never redo in commit mode (manually run `make empty_db` instead)
60 1570 aaronmk
61
    # Ranges
62
    start = util.cast(int, opts.get_env_var('start', '0', env_names))
63 1572 aaronmk
    if test: end_default = 1
64
    else: end_default = None
65
    end = util.cast(int, util.none_if(
66
        opts.get_env_var('n', end_default, env_names), u''))
67 1570 aaronmk
    if end != None: end += start
68
69
    # Debugging
70 946 aaronmk
    debug = opts.env_flag('debug', False, env_names)
71 859 aaronmk
    sql.run_raw_query.debug = debug
72 1574 aaronmk
    verbose = debug or opts.env_flag('verbose', not test, env_names)
73 944 aaronmk
    opts.get_env_var('profile_to', None, env_names) # add to env_names
74 131 aaronmk
75 1570 aaronmk
    # DB
76
    db_config_names = ['engine', 'host', 'user', 'password', 'database']
77
    def get_db_config(prefix):
78
        return opts.get_env_vars(db_config_names, prefix, env_names)
79
    in_db_config = get_db_config('in')
80
    out_db_config = get_db_config('out')
81
    in_is_db = 'engine' in in_db_config
82
    out_is_db = 'engine' in out_db_config
83
84
    ##
85
86 838 aaronmk
    # Logging
87 662 aaronmk
    def log(msg, on=verbose):
88
        if on: sys.stderr.write(msg)
89 859 aaronmk
    def log_start(action, on=verbose): log(action+'...\n', on)
90 662 aaronmk
91 53 aaronmk
    # Parse args
92 510 aaronmk
    map_paths = sys.argv[1:]
93 512 aaronmk
    if map_paths == []:
94
        if in_is_db or not out_is_db: usage_err()
95
        else: map_paths = [None]
96 53 aaronmk
97 646 aaronmk
    def connect_db(db_config):
98 662 aaronmk
        log_start('Connecting to '+sql.db_config_str(db_config))
99 1014 aaronmk
        return sql.connect(db_config)
100 646 aaronmk
101 1573 aaronmk
    if end != None: end_str = str(end-1)
102
    else: end_str = 'end'
103
    log_start('Processing input rows '+str(start)+'-'+end_str)
104
105 1014 aaronmk
    ex_tracker = exc.ExPercentTracker(iter_text='row')
106
    profiler = profiling.ItersProfiler(start_now=True, iter_text='row')
107
108
    doc = xml_dom.create_doc()
109
    root = doc.documentElement
110 751 aaronmk
    out_is_xml_ref = [False]
111 1014 aaronmk
    in_label_ref = [None]
112
    def update_in_label():
113
        if in_label_ref[0] != None:
114
            xpath.get(root, '/_ignore/inLabel="'+in_label_ref[0]+'"', True)
115
    def prep_root():
116
        root.clear()
117
        update_in_label()
118
    prep_root()
119 751 aaronmk
120 838 aaronmk
    def process_input(root, row_ready, map_path):
121 512 aaronmk
        '''Inputs datasource to XML tree, mapping if needed'''
122
        # Load map header
123
        in_is_xpaths = True
124 751 aaronmk
        out_is_xpaths = True
125 512 aaronmk
        out_label = None
126
        if map_path != None:
127
            metadata = []
128
            mappings = []
129
            stream = open(map_path, 'rb')
130
            reader = csv.reader(stream)
131
            in_label, out_label = reader.next()[:2]
132 1402 aaronmk
133 512 aaronmk
            def split_col_name(name):
134 1402 aaronmk
                label, sep, root = name.partition(':')
135
                label, sep2, prefixes_str = label.partition('[')
136
                prefixes_str = strings.remove_suffix(']', prefixes_str)
137
                prefixes = strings.split(',', prefixes_str)
138
                return label, sep != '', root, prefixes
139 1198 aaronmk
                    # extract datasrc from "datasrc[data_format]"
140 1402 aaronmk
141 1705 aaronmk
            in_label, in_root, prefixes = maps.col_info(in_label)
142
            in_is_xpaths = in_root != None
143 1014 aaronmk
            in_label_ref[0] = in_label
144
            update_in_label()
145 1705 aaronmk
            out_label, out_root = maps.col_info(out_label)[:2]
146
            out_is_xpaths = out_root != None
147
            if out_is_xpaths: has_types = out_root.startswith('/*s/')
148
                # outer elements are types
149 1402 aaronmk
150 512 aaronmk
            for row in reader:
151
                in_, out = row[:2]
152
                if out != '':
153
                    if out_is_xpaths: out = xpath.parse(out_root+out)
154
                    mappings.append((in_, out))
155 1402 aaronmk
156 512 aaronmk
            stream.close()
157
158
            root.ownerDocument.documentElement.tagName = out_label
159
        in_is_xml = in_is_xpaths and not in_is_db
160 751 aaronmk
        out_is_xml_ref[0] = out_is_xpaths and not out_is_db
161 56 aaronmk
162 1148 aaronmk
        def process_rows(process_row, rows):
163 838 aaronmk
            '''Processes input rows
164
            @param process_row(in_row, i)
165 297 aaronmk
            '''
166 838 aaronmk
            i = -1 # in case for loop does not execute
167 314 aaronmk
            for i, row in enumerate(rows):
168 838 aaronmk
                if i < start: continue
169
                if end != None and i >= end: break
170
                process_row(row, i)
171
                row_ready(i, row)
172 978 aaronmk
            row_ct = i-start+1
173 982 aaronmk
            return row_ct
174 838 aaronmk
175
        def map_rows(get_value, rows):
176
            '''Maps input rows
177
            @param get_value(in_, row):str
178
            '''
179
            def process_row(row, i):
180 316 aaronmk
                row_id = str(i)
181
                for in_, out in mappings:
182
                    value = metadata_value(in_)
183 662 aaronmk
                    if value == None:
184 857 aaronmk
                        log_start('Getting '+str(in_), debug)
185 1360 aaronmk
                        value = cleanup(get_value(in_, row))
186 1348 aaronmk
                    if value != None:
187
                        log_start('Putting '+str(out), debug)
188 1360 aaronmk
                        xpath.put_obj(root, out, row_id, has_types, value)
189 982 aaronmk
            return process_rows(process_row, rows)
190 297 aaronmk
191 1403 aaronmk
        def map_table(col_names, rows):
192 1416 aaronmk
            col_names_ct = len(col_names)
193 1403 aaronmk
            col_idxs = util.list_flip(col_names)
194
195
            i = 0
196
            while i < len(mappings): # mappings len changes in loop
197
                in_, out = mappings[i]
198
                if metadata_value(in_) == None:
199 1404 aaronmk
                    try: mappings[i] = (
200
                        get_with_prefix(col_idxs, prefixes, in_), out)
201
                    except KeyError:
202
                        del mappings[i]
203
                        continue # keep i the same
204 1403 aaronmk
                i += 1
205
206
            def get_value(in_, row):
207 1484 aaronmk
                return util.coalesce(*util.list_subset(row.list, in_))
208 1416 aaronmk
            def wrap_row(row):
209
                return util.ListDict(util.list_as_length(row, col_names_ct),
210
                    col_names, col_idxs) # handle CSV rows of different lengths
211 1403 aaronmk
212
            return map_rows(get_value, util.WrapIter(wrap_row, rows))
213
214 1700 aaronmk
        if in_is_db:
215 130 aaronmk
            assert in_is_xpaths
216 126 aaronmk
217 1136 aaronmk
            in_db = connect_db(in_db_config)
218
            in_pkeys = {}
219
            cur = sql.select(in_db, table=in_root, fields=None, conds=None,
220
                limit=end, start=0)
221 1403 aaronmk
            row_ct = map_table(list(sql.col_names(cur)), sql.rows(cur))
222 1136 aaronmk
223 117 aaronmk
            in_db.close()
224 161 aaronmk
        elif in_is_xml:
225 1701 aaronmk
            while True:
226
                try: in_xml_root = minidom.parse(sys.stdin).documentElement
227
                except expat.ExpatError, e:
228
                    if str(e).startswith('no element found:'): break
229
                        # no more concatenated XML documents
230
                    else: raise e
231 1700 aaronmk
232 1701 aaronmk
                if map_path == None:
233
                    iter_ = xml_dom.NodeElemIter(in_xml_root)
234
                    util.skip(iter_, xml_dom.is_text) # skip metadata
235
                    row_ct = process_rows(lambda row, i: root.appendChild(row),
236
                        iter_)
237
                else:
238
                    rows = xpath.get(in_xml_root, in_root, limit=end)
239
                    if rows == []: raise SystemExit('Map error: Root "'+in_root
240
                        +'" not found in input')
241
242
                    def get_value(in_, row):
243
                        in_ = './{'+(','.join(strings.with_prefixes(
244
                            ['']+prefixes, in_)))+'}' # also with no prefix
245
                        nodes = xpath.get(row, in_, allow_rooted=False)
246
                        if nodes != []: return xml_dom.value(nodes[0])
247
                        else: return None
248
249
                    row_ct = map_rows(get_value, rows)
250 56 aaronmk
        else: # input is CSV
251 133 aaronmk
            map_ = dict(mappings)
252 1389 aaronmk
            reader, col_names = csvs.reader_and_header(sys.stdin)
253 1403 aaronmk
            row_ct = map_table(col_names, reader)
254 982 aaronmk
255
        return row_ct
256 53 aaronmk
257 838 aaronmk
    def process_inputs(root, row_ready):
258 982 aaronmk
        row_ct = 0
259
        for map_path in map_paths:
260
            row_ct += process_input(root, row_ready, map_path)
261
        return row_ct
262 512 aaronmk
263 130 aaronmk
    if out_is_db:
264 53 aaronmk
        import db_xml
265
266 646 aaronmk
        out_db = connect_db(out_db_config)
267 310 aaronmk
        out_pkeys = {}
268 53 aaronmk
        try:
269 947 aaronmk
            if redo: sql.empty_db(out_db)
270 982 aaronmk
            row_ins_ct_ref = [0]
271 449 aaronmk
272 838 aaronmk
            def row_ready(row_num, input_row):
273 452 aaronmk
                def on_error(e):
274 835 aaronmk
                    exc.add_msg(e, term.emph('row #:')+' '+str(row_num))
275 828 aaronmk
                    exc.add_msg(e, term.emph('input row:')+'\n'+str(input_row))
276
                    exc.add_msg(e, term.emph('output row:')+'\n'+str(root))
277 1617 aaronmk
                    ex_tracker.track(e, row_num)
278 452 aaronmk
279 449 aaronmk
                xml_func.process(root, on_error)
280 442 aaronmk
                if not xml_dom.is_empty(root):
281
                    assert xml_dom.has_one_child(root)
282
                    try:
283 982 aaronmk
                        sql.with_savepoint(out_db,
284
                            lambda: db_xml.put(out_db, root.firstChild,
285
                                out_pkeys, row_ins_ct_ref, on_error))
286 442 aaronmk
                        if commit: out_db.commit()
287 449 aaronmk
                    except sql.DatabaseErrors, e: on_error(e)
288 1010 aaronmk
                prep_root()
289 449 aaronmk
290 982 aaronmk
            row_ct = process_inputs(root, row_ready)
291
            sys.stdout.write('Inserted '+str(row_ins_ct_ref[0])+
292 460 aaronmk
                ' new rows into database\n')
293 53 aaronmk
        finally:
294 133 aaronmk
            out_db.rollback()
295
            out_db.close()
296 751 aaronmk
    else:
297 759 aaronmk
        def on_error(e): ex_tracker.track(e)
298 838 aaronmk
        def row_ready(row_num, input_row): pass
299 982 aaronmk
        row_ct = process_inputs(root, row_ready)
300 759 aaronmk
        xml_func.process(root, on_error)
301 751 aaronmk
        if out_is_xml_ref[0]:
302
            doc.writexml(sys.stdout, **xml_dom.prettyxml_config)
303
        else: # output is CSV
304
            raise NotImplementedError('CSV output not supported yet')
305 985 aaronmk
306 982 aaronmk
    profiler.stop(row_ct)
307
    ex_tracker.add_iters(row_ct)
308 990 aaronmk
    if verbose:
309
        sys.stderr.write('Processed '+str(row_ct)+' input rows\n')
310
        sys.stderr.write(profiler.msg()+'\n')
311
        sys.stderr.write(ex_tracker.msg()+'\n')
312 985 aaronmk
    ex_tracker.exit()
313 53 aaronmk
314 847 aaronmk
def main():
315
    try: main_()
316
    except Parser.SyntaxException, e: raise SystemExit(str(e))
317
318 846 aaronmk
if __name__ == '__main__':
319 847 aaronmk
    profile_to = opts.get_env_var('profile_to', None)
320
    if profile_to != None:
321
        import cProfile
322 1584 aaronmk
        sys.stderr.write('Profiling to '+profile_to+'\n')
323 847 aaronmk
        cProfile.run(main.func_code, profile_to)
324
    else: main()