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 1714 aaronmk
import itertools
9 53 aaronmk
import os.path
10
import sys
11 299 aaronmk
import xml.dom.minidom as minidom
12 53 aaronmk
13 266 aaronmk
sys.path.append(os.path.dirname(__file__)+"/../lib")
14 53 aaronmk
15 1389 aaronmk
import csvs
16 2040 aaronmk
import db_xml
17 344 aaronmk
import exc
18 1714 aaronmk
import iters
19 1705 aaronmk
import maps
20 64 aaronmk
import opts
21 2035 aaronmk
import parallelproc
22 281 aaronmk
import Parser
23 982 aaronmk
import profiling
24 131 aaronmk
import sql
25 2288 aaronmk
import sql_gen
26 1758 aaronmk
import streams
27 715 aaronmk
import strings
28 828 aaronmk
import term
29 310 aaronmk
import util
30 1014 aaronmk
import xpath
31 133 aaronmk
import xml_dom
32 86 aaronmk
import xml_func
33 1713 aaronmk
import xml_parse
34 53 aaronmk
35 1404 aaronmk
def get_with_prefix(map_, prefixes, key):
36 2040 aaronmk
    '''Gets all entries for the given key with any of the given prefixes
37
    @return tuple(found_key, found_value)
38
    '''
39 1484 aaronmk
    values = []
40 1681 aaronmk
    for key_ in strings.with_prefixes(['']+prefixes, key): # also with no prefix
41
        try: value = map_[key_]
42 1484 aaronmk
        except KeyError, e: continue # keep going
43 2040 aaronmk
        values.append((key_, value))
44 1484 aaronmk
45
    if values != []: return values
46
    else: raise e # re-raise last KeyError
47 1404 aaronmk
48 1018 aaronmk
def metadata_value(name): return None # this feature has been removed
49 84 aaronmk
50 1360 aaronmk
def cleanup(val):
51
    if val == None: return val
52 1374 aaronmk
    return util.none_if(strings.cleanup(strings.ustr(val)), u'', u'\\N')
53 1360 aaronmk
54 847 aaronmk
def main_():
55 131 aaronmk
    env_names = []
56
    def usage_err():
57 944 aaronmk
        raise SystemExit('Usage: '+opts.env_usage(env_names, True)+' '
58 1945 aaronmk
            +sys.argv[0]+' [map_path...] [<input] [>output]\n'
59 1946 aaronmk
            'Note: Row #s start with 1')
60 838 aaronmk
61 1570 aaronmk
    ## Get config from env vars
62 838 aaronmk
63 1570 aaronmk
    # Modes
64 946 aaronmk
    test = opts.env_flag('test', False, env_names)
65 947 aaronmk
    commit = opts.env_flag('commit', False, env_names) and not test
66 944 aaronmk
        # never commit in test mode
67 947 aaronmk
    redo = opts.env_flag('redo', test, env_names) and not commit
68
        # never redo in commit mode (manually run `make empty_db` instead)
69 1570 aaronmk
70
    # Ranges
71 1946 aaronmk
    start = util.cast(int, opts.get_env_var('start', 1, env_names)) # 1-based
72
    # Make start interally 0-based.
73
    # It's 1-based to the user to match up with the staging table row #s.
74
    start -= 1
75 1945 aaronmk
    if test: n_default = 1
76
    else: n_default = None
77
    n = util.cast(int, util.none_if(opts.get_env_var('n', n_default, env_names),
78
        u''))
79
    end = n
80 1570 aaronmk
    if end != None: end += start
81
82
    # Debugging
83 2458 aaronmk
    if test: default_verbosity = 0.5
84
    else: default_verbosity = 1
85 2456 aaronmk
    verbosity = util.cast(float, opts.get_env_var('verbosity',
86
        default_verbosity, env_names))
87 944 aaronmk
    opts.get_env_var('profile_to', None, env_names) # add to env_names
88 131 aaronmk
89 1570 aaronmk
    # DB
90
    def get_db_config(prefix):
91 1926 aaronmk
        return opts.get_env_vars(sql.db_config_names, prefix, env_names)
92 1570 aaronmk
    in_db_config = get_db_config('in')
93
    out_db_config = get_db_config('out')
94
    in_is_db = 'engine' in in_db_config
95
    out_is_db = 'engine' in out_db_config
96 1982 aaronmk
    in_schema = opts.get_env_var('in_schema', None, env_names)
97
    in_table = opts.get_env_var('in_table', None, env_names)
98 1570 aaronmk
99 1989 aaronmk
    # Optimization
100 2050 aaronmk
    cache_sql = opts.env_flag('cache_sql', True, env_names)
101 1989 aaronmk
    by_col = in_db_config == out_db_config and opts.env_flag('by_col', False,
102
        env_names) # by-column optimization only applies if mapping to same DB
103
    if test: cpus_default = 0
104 2044 aaronmk
    else: cpus_default = 0 # or None to use parallel processing by default
105 1989 aaronmk
    cpus = util.cast(int, util.none_if(opts.get_env_var('cpus', cpus_default,
106
        env_names), u''))
107
108 1570 aaronmk
    ##
109
110 838 aaronmk
    # Logging
111 2457 aaronmk
    verbose_errors = test and verbosity > 0
112 2449 aaronmk
    debug = verbosity > 1
113 2439 aaronmk
    def log(msg, level=1):
114
        '''Higher level -> more verbose'''
115 2452 aaronmk
        if level <= verbosity:
116
            if level > 1: msg = '['+str(level)+'] '+msg # include level in msg
117 2480 aaronmk
            sys.stderr.write(msg.rstrip('\n')+'\n')
118 2439 aaronmk
    if debug: log_debug = lambda msg, level=2: log(msg, level)
119 1901 aaronmk
    else: log_debug = sql.log_debug_none
120 662 aaronmk
121 53 aaronmk
    # Parse args
122 510 aaronmk
    map_paths = sys.argv[1:]
123 512 aaronmk
    if map_paths == []:
124
        if in_is_db or not out_is_db: usage_err()
125
        else: map_paths = [None]
126 53 aaronmk
127 646 aaronmk
    def connect_db(db_config):
128 1901 aaronmk
        log('Connecting to '+sql.db_config_str(db_config))
129 2192 aaronmk
        return sql.connect(db_config, caching=cache_sql,
130
            autocommit=debug and commit, log_debug=log_debug)
131 646 aaronmk
132 1945 aaronmk
    if end != None: end_str = str(end-1) # end is one past the last #
133 1573 aaronmk
    else: end_str = 'end'
134 1901 aaronmk
    log('Processing input rows '+str(start)+'-'+end_str)
135 1573 aaronmk
136 1014 aaronmk
    ex_tracker = exc.ExPercentTracker(iter_text='row')
137
    profiler = profiling.ItersProfiler(start_now=True, iter_text='row')
138
139 1856 aaronmk
    # Parallel processing
140 2035 aaronmk
    pool = parallelproc.MultiProducerPool(cpus)
141 1901 aaronmk
    log('Using '+str(pool.process_ct)+' parallel CPUs')
142 1846 aaronmk
143 1014 aaronmk
    doc = xml_dom.create_doc()
144
    root = doc.documentElement
145 751 aaronmk
    out_is_xml_ref = [False]
146 1014 aaronmk
    in_label_ref = [None]
147
    def update_in_label():
148
        if in_label_ref[0] != None:
149
            xpath.get(root, '/_ignore/inLabel="'+in_label_ref[0]+'"', True)
150
    def prep_root():
151
        root.clear()
152
        update_in_label()
153
    prep_root()
154 751 aaronmk
155 1991 aaronmk
    # Define before the out_is_db section because it's used by by_col
156
    row_ins_ct_ref = [0]
157
158 838 aaronmk
    def process_input(root, row_ready, map_path):
159 512 aaronmk
        '''Inputs datasource to XML tree, mapping if needed'''
160
        # Load map header
161
        in_is_xpaths = True
162 751 aaronmk
        out_is_xpaths = True
163 512 aaronmk
        out_label = None
164
        if map_path != None:
165
            metadata = []
166
            mappings = []
167
            stream = open(map_path, 'rb')
168
            reader = csv.reader(stream)
169
            in_label, out_label = reader.next()[:2]
170 1402 aaronmk
171 512 aaronmk
            def split_col_name(name):
172 1402 aaronmk
                label, sep, root = name.partition(':')
173
                label, sep2, prefixes_str = label.partition('[')
174
                prefixes_str = strings.remove_suffix(']', prefixes_str)
175
                prefixes = strings.split(',', prefixes_str)
176
                return label, sep != '', root, prefixes
177 1198 aaronmk
                    # extract datasrc from "datasrc[data_format]"
178 1402 aaronmk
179 1705 aaronmk
            in_label, in_root, prefixes = maps.col_info(in_label)
180
            in_is_xpaths = in_root != None
181 1014 aaronmk
            in_label_ref[0] = in_label
182
            update_in_label()
183 1705 aaronmk
            out_label, out_root = maps.col_info(out_label)[:2]
184
            out_is_xpaths = out_root != None
185 1841 aaronmk
            if out_is_xpaths: has_types = out_root.find('/*s/') >= 0
186 1705 aaronmk
                # outer elements are types
187 1402 aaronmk
188 512 aaronmk
            for row in reader:
189
                in_, out = row[:2]
190 2029 aaronmk
                if out != '': mappings.append([in_, out_root+out])
191 1402 aaronmk
192 512 aaronmk
            stream.close()
193
194
            root.ownerDocument.documentElement.tagName = out_label
195
        in_is_xml = in_is_xpaths and not in_is_db
196 751 aaronmk
        out_is_xml_ref[0] = out_is_xpaths and not out_is_db
197 56 aaronmk
198 1897 aaronmk
        def process_rows(process_row, rows, rows_start=0):
199
            '''Processes input rows
200 838 aaronmk
            @param process_row(in_row, i)
201 1945 aaronmk
            @rows_start The (0-based) row # of the first row in rows. Set this
202
                only if the pre-start rows have already been skipped.
203 297 aaronmk
            '''
204 1897 aaronmk
            rows = iter(rows)
205
206
            if end != None: row_nums = xrange(rows_start, end)
207
            else: row_nums = itertools.count(rows_start)
208 2038 aaronmk
            i = -1
209 1897 aaronmk
            for i in row_nums:
210 1718 aaronmk
                try: row = rows.next()
211 2038 aaronmk
                except StopIteration:
212
                    i -= 1 # last row # didn't count
213
                    break # no more rows
214 1718 aaronmk
                if i < start: continue # not at start row yet
215
216 838 aaronmk
                process_row(row, i)
217
                row_ready(i, row)
218 2033 aaronmk
            row_ct = i-start+1
219 982 aaronmk
            return row_ct
220 838 aaronmk
221 1898 aaronmk
        def map_rows(get_value, rows, **kw_args):
222 838 aaronmk
            '''Maps input rows
223
            @param get_value(in_, row):str
224
            '''
225 2030 aaronmk
            # Prevent collisions if multiple inputs mapping to same output
226
            outputs_idxs = dict()
227
            for i, mapping in enumerate(mappings):
228
                in_, out = mapping
229
                default = util.NamedTuple(count=1, first=i)
230
                idxs = outputs_idxs.setdefault(out, default)
231
                if idxs is not default: # key existed, so there was a collision
232
                    if idxs.count == 1: # first key does not yet have /_alt/#
233
                        mappings[idxs.first][1] += '/_alt/0'
234
                    mappings[i][1] += '/_alt/'+str(idxs.count)
235
                    idxs.count += 1
236
237 2026 aaronmk
            id_node = None
238
            if out_is_db:
239
                for i, mapping in enumerate(mappings):
240
                    in_, out = mapping
241
                    # All put_obj()s should return the same id_node
242
                    nodes, id_node = xpath.put_obj(root, out, '-1', has_types,
243
                        '$'+str(in_)) # value is placeholder that documents name
244 2032 aaronmk
                    mappings[i] = [in_, nodes]
245
                assert id_node != None
246 2026 aaronmk
247 2119 aaronmk
                if debug: # only calc if debug
248 2026 aaronmk
                    log_debug('Put template:\n'+str(root))
249
250 838 aaronmk
            def process_row(row, i):
251 316 aaronmk
                row_id = str(i)
252 2032 aaronmk
                if id_node != None: xml_dom.set_value(id_node, row_id)
253 316 aaronmk
                for in_, out in mappings:
254 2027 aaronmk
                    log_debug('Getting '+str(in_))
255 316 aaronmk
                    value = metadata_value(in_)
256 2027 aaronmk
                    if value == None: value = cleanup(get_value(in_, row))
257
                    log_debug('Putting '+repr(value)+' to '+str(out))
258 2032 aaronmk
                    if out_is_db: # out is list of XML nodes
259
                        for node in out: xml_dom.set_value(node, value)
260
                    elif value != None: # out is XPath
261 1360 aaronmk
                        xpath.put_obj(root, out, row_id, has_types, value)
262 1898 aaronmk
            return process_rows(process_row, rows, **kw_args)
263 297 aaronmk
264 1898 aaronmk
        def map_table(col_names, rows, **kw_args):
265 1416 aaronmk
            col_names_ct = len(col_names)
266 1403 aaronmk
            col_idxs = util.list_flip(col_names)
267
268 2029 aaronmk
            # Resolve prefixes
269 2025 aaronmk
            mappings_orig = mappings[:] # save a copy
270
            mappings[:] = [] # empty existing elements
271
            for in_, out in mappings_orig:
272 1403 aaronmk
                if metadata_value(in_) == None:
273 2040 aaronmk
                    try: cols = get_with_prefix(col_idxs, prefixes, in_)
274 2025 aaronmk
                    except KeyError: pass
275 2040 aaronmk
                    else: mappings[len(mappings):] = [[db_xml.ColRef(*col), out]
276
                        for col in cols] # can't use += because that uses =
277 1403 aaronmk
278 2040 aaronmk
            def get_value(in_, row): return row.list[in_.idx]
279 1416 aaronmk
            def wrap_row(row):
280
                return util.ListDict(util.list_as_length(row, col_names_ct),
281
                    col_names, col_idxs) # handle CSV rows of different lengths
282 1403 aaronmk
283 1898 aaronmk
            return map_rows(get_value, util.WrapIter(wrap_row, rows), **kw_args)
284 1403 aaronmk
285 1758 aaronmk
        stdin = streams.LineCountStream(sys.stdin)
286
        def on_error(e):
287
            exc.add_msg(e, term.emph('input line #:')+' '+str(stdin.line_num))
288
            ex_tracker.track(e)
289
290 1700 aaronmk
        if in_is_db:
291 1982 aaronmk
            in_db = connect_db(in_db_config)
292 126 aaronmk
293 1982 aaronmk
            # Get table and schema name
294
            schema = in_schema # modified, so can't have same name as outer var
295
            table = in_table # modified, so can't have same name as outer var
296
            if table == None:
297
                assert in_is_xpaths
298
                schema, sep, table = in_root.partition('.')
299
                if sep == '': # only the table name was specified
300
                    table = schema
301
                    schema = None
302 2313 aaronmk
            table = sql_gen.Table(table, schema)
303 1982 aaronmk
304 1991 aaronmk
            # Fetch rows
305
            if by_col: limit = 0 # only fetch column names
306
            else: limit = n
307 2313 aaronmk
            cur = sql.select(in_db, table, limit=limit, start=start,
308
                cacheable=False)
309 1991 aaronmk
            col_names = list(sql.col_names(cur))
310
311 1989 aaronmk
            if by_col:
312 2042 aaronmk
                map_table(col_names, []) # just create the template
313 2435 aaronmk
                xml_func.strip(root, ['_simplifyPath'])
314 2103 aaronmk
                if debug: log_debug('Putting stripped:\n'+str(root))
315 2119 aaronmk
                    # only calc if debug
316 2422 aaronmk
                in_row_ct_ref = [0]
317 2417 aaronmk
                db_xml.put_table(in_db, root.firstChild, table, commit,
318 2422 aaronmk
                    in_row_ct_ref, row_ins_ct_ref, n, start)
319
                row_ct = in_row_ct_ref[0]
320 1984 aaronmk
            else:
321
                # Use normal by-row method
322 1991 aaronmk
                row_ct = map_table(col_names, sql.rows(cur), rows_start=start)
323
                    # rows_start: pre-start rows have been skipped
324 1136 aaronmk
325 1849 aaronmk
            in_db.db.close()
326 161 aaronmk
        elif in_is_xml:
327 1715 aaronmk
            def get_rows(doc2rows):
328 1758 aaronmk
                return iters.flatten(itertools.imap(doc2rows,
329
                    xml_parse.docs_iter(stdin, on_error)))
330 1715 aaronmk
331 1714 aaronmk
            if map_path == None:
332 1715 aaronmk
                def doc2rows(in_xml_root):
333 1701 aaronmk
                    iter_ = xml_dom.NodeElemIter(in_xml_root)
334
                    util.skip(iter_, xml_dom.is_text) # skip metadata
335 1715 aaronmk
                    return iter_
336
337
                row_ct = process_rows(lambda row, i: root.appendChild(row),
338
                    get_rows(doc2rows))
339 1714 aaronmk
            else:
340 1715 aaronmk
                def doc2rows(in_xml_root):
341 1701 aaronmk
                    rows = xpath.get(in_xml_root, in_root, limit=end)
342 1714 aaronmk
                    if rows == []: raise SystemExit('Map error: Root "'
343
                        +in_root+'" not found in input')
344
                    return rows
345
346
                def get_value(in_, row):
347
                    in_ = './{'+(','.join(strings.with_prefixes(
348
                        ['']+prefixes, in_)))+'}' # also with no prefix
349
                    nodes = xpath.get(row, in_, allow_rooted=False)
350
                    if nodes != []: return xml_dom.value(nodes[0])
351
                    else: return None
352
353 1715 aaronmk
                row_ct = map_rows(get_value, get_rows(doc2rows))
354 56 aaronmk
        else: # input is CSV
355 133 aaronmk
            map_ = dict(mappings)
356 1389 aaronmk
            reader, col_names = csvs.reader_and_header(sys.stdin)
357 1403 aaronmk
            row_ct = map_table(col_names, reader)
358 982 aaronmk
359
        return row_ct
360 53 aaronmk
361 838 aaronmk
    def process_inputs(root, row_ready):
362 982 aaronmk
        row_ct = 0
363
        for map_path in map_paths:
364
            row_ct += process_input(root, row_ready, map_path)
365
        return row_ct
366 512 aaronmk
367 1886 aaronmk
    pool.share_vars(locals())
368 130 aaronmk
    if out_is_db:
369 646 aaronmk
        out_db = connect_db(out_db_config)
370 53 aaronmk
        try:
371 947 aaronmk
            if redo: sql.empty_db(out_db)
372 1886 aaronmk
            pool.share_vars(locals())
373 449 aaronmk
374 838 aaronmk
            def row_ready(row_num, input_row):
375 2119 aaronmk
                row_str_ = [None]
376
                def row_str():
377
                    if row_str_[0] == None:
378
                        # Row # is interally 0-based, but 1-based to the user
379
                        row_str_[0] = (term.emph('row #:')+' '+str(row_num+1)
380
                            +'\n'+term.emph('input row:')+'\n'+str(input_row))
381
                        if verbose_errors: row_str_[0] += ('\n'
382
                            +term.emph('output row:')+'\n'+str(root))
383
                    return row_str_[0]
384
385
                if debug: log_debug(row_str()) # only calc if debug
386
387 452 aaronmk
                def on_error(e):
388 2119 aaronmk
                    exc.add_msg(e, row_str())
389 2046 aaronmk
                    ex_tracker.track(e, row_num, detail=verbose_errors)
390 1886 aaronmk
                pool.share_vars(locals())
391 452 aaronmk
392 2032 aaronmk
                row_root = root.cloneNode(True) # deep copy so don't modify root
393 2106 aaronmk
                xml_func.process(row_root, on_error, out_db)
394 2032 aaronmk
                if not xml_dom.is_empty(row_root):
395
                    assert xml_dom.has_one_child(row_root)
396 442 aaronmk
                    try:
397 982 aaronmk
                        sql.with_savepoint(out_db,
398 2032 aaronmk
                            lambda: db_xml.put(out_db, row_root.firstChild,
399 1850 aaronmk
                                row_ins_ct_ref, on_error))
400 1849 aaronmk
                        if commit: out_db.db.commit()
401 449 aaronmk
                    except sql.DatabaseErrors, e: on_error(e)
402
403 982 aaronmk
            row_ct = process_inputs(root, row_ready)
404
            sys.stdout.write('Inserted '+str(row_ins_ct_ref[0])+
405 460 aaronmk
                ' new rows into database\n')
406 1868 aaronmk
407
            # Consume asynchronous tasks
408
            pool.main_loop()
409 53 aaronmk
        finally:
410 2166 aaronmk
            if out_db.connected():
411
                out_db.db.rollback()
412
                out_db.db.close()
413 751 aaronmk
    else:
414 759 aaronmk
        def on_error(e): ex_tracker.track(e)
415 838 aaronmk
        def row_ready(row_num, input_row): pass
416 982 aaronmk
        row_ct = process_inputs(root, row_ready)
417 759 aaronmk
        xml_func.process(root, on_error)
418 751 aaronmk
        if out_is_xml_ref[0]:
419
            doc.writexml(sys.stdout, **xml_dom.prettyxml_config)
420
        else: # output is CSV
421
            raise NotImplementedError('CSV output not supported yet')
422 985 aaronmk
423 1868 aaronmk
    # Consume any asynchronous tasks not already consumed above
424 1862 aaronmk
    pool.main_loop()
425
426 982 aaronmk
    profiler.stop(row_ct)
427
    ex_tracker.add_iters(row_ct)
428 2439 aaronmk
    log('Processed '+str(row_ct)+' input rows')
429
    log(profiler.msg())
430
    log(ex_tracker.msg())
431 985 aaronmk
    ex_tracker.exit()
432 53 aaronmk
433 847 aaronmk
def main():
434
    try: main_()
435 1719 aaronmk
    except Parser.SyntaxError, e: raise SystemExit(str(e))
436 847 aaronmk
437 846 aaronmk
if __name__ == '__main__':
438 847 aaronmk
    profile_to = opts.get_env_var('profile_to', None)
439
    if profile_to != None:
440
        import cProfile
441 1584 aaronmk
        sys.stderr.write('Profiling to '+profile_to+'\n')
442 847 aaronmk
        cProfile.run(main.func_code, profile_to)
443
    else: main()