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