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 2977 aaronmk
        # never redo in commit mode (run `make schemas/reinstall` 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 2883 aaronmk
            sys.stderr.write(strings.to_raw_str(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 2920 aaronmk
        return sql.connect(db_config, caching=cache_sql, autocommit=commit,
136 2916 aaronmk
            debug_temp=verbosity > 3 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 2881 aaronmk
                # Row # is interally 0-based, but 1-based to the user
228
                log('Processing input row #'+str(i+1), level=1.1)
229 838 aaronmk
                process_row(row, i)
230
                row_ready(i, row)
231 2033 aaronmk
            row_ct = i-start+1
232 982 aaronmk
            return row_ct
233 838 aaronmk
234 1898 aaronmk
        def map_rows(get_value, rows, **kw_args):
235 838 aaronmk
            '''Maps input rows
236
            @param get_value(in_, row):str
237
            '''
238 2030 aaronmk
            # Prevent collisions if multiple inputs mapping to same output
239
            outputs_idxs = dict()
240
            for i, mapping in enumerate(mappings):
241
                in_, out = mapping
242
                default = util.NamedTuple(count=1, first=i)
243
                idxs = outputs_idxs.setdefault(out, default)
244
                if idxs is not default: # key existed, so there was a collision
245
                    if idxs.count == 1: # first key does not yet have /_alt/#
246
                        mappings[idxs.first][1] += '/_alt/0'
247
                    mappings[i][1] += '/_alt/'+str(idxs.count)
248
                    idxs.count += 1
249
250 2026 aaronmk
            id_node = None
251
            if out_is_db:
252
                for i, mapping in enumerate(mappings):
253
                    in_, out = mapping
254
                    # All put_obj()s should return the same id_node
255
                    nodes, id_node = xpath.put_obj(root, out, '-1', has_types,
256
                        '$'+str(in_)) # value is placeholder that documents name
257 2032 aaronmk
                    mappings[i] = [in_, nodes]
258
                assert id_node != None
259 2026 aaronmk
260 2119 aaronmk
                if debug: # only calc if debug
261 2026 aaronmk
                    log_debug('Put template:\n'+str(root))
262
263 838 aaronmk
            def process_row(row, i):
264 316 aaronmk
                row_id = str(i)
265 2032 aaronmk
                if id_node != None: xml_dom.set_value(id_node, row_id)
266 316 aaronmk
                for in_, out in mappings:
267 2027 aaronmk
                    log_debug('Getting '+str(in_))
268 316 aaronmk
                    value = metadata_value(in_)
269 2027 aaronmk
                    if value == None: value = cleanup(get_value(in_, row))
270
                    log_debug('Putting '+repr(value)+' to '+str(out))
271 2032 aaronmk
                    if out_is_db: # out is list of XML nodes
272
                        for node in out: xml_dom.set_value(node, value)
273
                    elif value != None: # out is XPath
274 1360 aaronmk
                        xpath.put_obj(root, out, row_id, has_types, value)
275 1898 aaronmk
            return process_rows(process_row, rows, **kw_args)
276 297 aaronmk
277 1898 aaronmk
        def map_table(col_names, rows, **kw_args):
278 1416 aaronmk
            col_names_ct = len(col_names)
279 1403 aaronmk
            col_idxs = util.list_flip(col_names)
280
281 2029 aaronmk
            # Resolve prefixes
282 2025 aaronmk
            mappings_orig = mappings[:] # save a copy
283
            mappings[:] = [] # empty existing elements
284
            for in_, out in mappings_orig:
285 1403 aaronmk
                if metadata_value(in_) == None:
286 2040 aaronmk
                    try: cols = get_with_prefix(col_idxs, prefixes, in_)
287 2025 aaronmk
                    except KeyError: pass
288 2040 aaronmk
                    else: mappings[len(mappings):] = [[db_xml.ColRef(*col), out]
289
                        for col in cols] # can't use += because that uses =
290 1403 aaronmk
291 2040 aaronmk
            def get_value(in_, row): return row.list[in_.idx]
292 1416 aaronmk
            def wrap_row(row):
293
                return util.ListDict(util.list_as_length(row, col_names_ct),
294
                    col_names, col_idxs) # handle CSV rows of different lengths
295 1403 aaronmk
296 1898 aaronmk
            return map_rows(get_value, util.WrapIter(wrap_row, rows), **kw_args)
297 1403 aaronmk
298 1700 aaronmk
        if in_is_db:
299 2809 aaronmk
            def on_error(e): ex_tracker.track(e)
300
301 1982 aaronmk
            in_db = connect_db(in_db_config)
302 126 aaronmk
303 1982 aaronmk
            # Get table and schema name
304
            schema = in_schema # modified, so can't have same name as outer var
305
            table = in_table # modified, so can't have same name as outer var
306
            if table == None:
307
                assert in_is_xpaths
308
                schema, sep, table = in_root.partition('.')
309
                if sep == '': # only the table name was specified
310
                    table = schema
311
                    schema = None
312 2313 aaronmk
            table = sql_gen.Table(table, schema)
313 1982 aaronmk
314 1991 aaronmk
            # Fetch rows
315
            if by_col: limit = 0 # only fetch column names
316
            else: limit = n
317 2313 aaronmk
            cur = sql.select(in_db, table, limit=limit, start=start,
318
                cacheable=False)
319 1991 aaronmk
            col_names = list(sql.col_names(cur))
320
321 1989 aaronmk
            if by_col:
322 2042 aaronmk
                map_table(col_names, []) # just create the template
323 2550 aaronmk
324 2734 aaronmk
                if n == None: # doing full import
325
                    log('Clearing errors table')
326
                    sql.truncate(in_db, sql.errors_table(in_db, table))
327
328 2550 aaronmk
                # Strip XML functions not in the DB
329 2602 aaronmk
                special_funcs = db_xml.put_table_special_funcs | rel_funcs
330
                xml_func.process(root, rel_funcs=special_funcs)
331 2103 aaronmk
                if debug: log_debug('Putting stripped:\n'+str(root))
332 2119 aaronmk
                    # only calc if debug
333 2550 aaronmk
334
                # Import rows
335 2422 aaronmk
                in_row_ct_ref = [0]
336 2928 aaronmk
                db_xml.put_table(in_db, root.firstChild, table, in_row_ct_ref,
337
                    row_ins_ct_ref, n, start, on_error)
338 2422 aaronmk
                row_ct = in_row_ct_ref[0]
339 1984 aaronmk
            else:
340
                # Use normal by-row method
341 1991 aaronmk
                row_ct = map_table(col_names, sql.rows(cur), rows_start=start)
342
                    # rows_start: pre-start rows have been skipped
343 1136 aaronmk
344 1849 aaronmk
            in_db.db.close()
345 161 aaronmk
        elif in_is_xml:
346 2809 aaronmk
            stdin = streams.LineCountStream(sys.stdin)
347
            def on_error(e):
348
                exc.add_msg(e, term.emph('input line #:')+' '
349
                    +str(stdin.line_num))
350
                ex_tracker.track(e)
351
352 1715 aaronmk
            def get_rows(doc2rows):
353 1758 aaronmk
                return iters.flatten(itertools.imap(doc2rows,
354
                    xml_parse.docs_iter(stdin, on_error)))
355 1715 aaronmk
356 1714 aaronmk
            if map_path == None:
357 1715 aaronmk
                def doc2rows(in_xml_root):
358 1701 aaronmk
                    iter_ = xml_dom.NodeElemIter(in_xml_root)
359
                    util.skip(iter_, xml_dom.is_text) # skip metadata
360 1715 aaronmk
                    return iter_
361
362
                row_ct = process_rows(lambda row, i: root.appendChild(row),
363
                    get_rows(doc2rows))
364 1714 aaronmk
            else:
365 1715 aaronmk
                def doc2rows(in_xml_root):
366 1701 aaronmk
                    rows = xpath.get(in_xml_root, in_root, limit=end)
367 1714 aaronmk
                    if rows == []: raise SystemExit('Map error: Root "'
368
                        +in_root+'" not found in input')
369
                    return rows
370
371
                def get_value(in_, row):
372
                    in_ = './{'+(','.join(strings.with_prefixes(
373
                        ['']+prefixes, in_)))+'}' # also with no prefix
374
                    nodes = xpath.get(row, in_, allow_rooted=False)
375
                    if nodes != []: return xml_dom.value(nodes[0])
376
                    else: return None
377
378 1715 aaronmk
                row_ct = map_rows(get_value, get_rows(doc2rows))
379 56 aaronmk
        else: # input is CSV
380 133 aaronmk
            map_ = dict(mappings)
381 1389 aaronmk
            reader, col_names = csvs.reader_and_header(sys.stdin)
382 1403 aaronmk
            row_ct = map_table(col_names, reader)
383 982 aaronmk
384
        return row_ct
385 53 aaronmk
386 838 aaronmk
    def process_inputs(root, row_ready):
387 982 aaronmk
        row_ct = 0
388
        for map_path in map_paths:
389
            row_ct += process_input(root, row_ready, map_path)
390
        return row_ct
391 512 aaronmk
392 1886 aaronmk
    pool.share_vars(locals())
393 130 aaronmk
    if out_is_db:
394 53 aaronmk
        try:
395 947 aaronmk
            if redo: sql.empty_db(out_db)
396 1886 aaronmk
            pool.share_vars(locals())
397 449 aaronmk
398 838 aaronmk
            def row_ready(row_num, input_row):
399 2119 aaronmk
                row_str_ = [None]
400
                def row_str():
401
                    if row_str_[0] == None:
402
                        # Row # is interally 0-based, but 1-based to the user
403
                        row_str_[0] = (term.emph('row #:')+' '+str(row_num+1)
404
                            +'\n'+term.emph('input row:')+'\n'+str(input_row))
405
                        if verbose_errors: row_str_[0] += ('\n'
406
                            +term.emph('output row:')+'\n'+str(root))
407
                    return row_str_[0]
408
409
                if debug: log_debug(row_str()) # only calc if debug
410
411 452 aaronmk
                def on_error(e):
412 2119 aaronmk
                    exc.add_msg(e, row_str())
413 2046 aaronmk
                    ex_tracker.track(e, row_num, detail=verbose_errors)
414 1886 aaronmk
                pool.share_vars(locals())
415 452 aaronmk
416 2032 aaronmk
                row_root = root.cloneNode(True) # deep copy so don't modify root
417 2639 aaronmk
                xml_func.process(row_root, on_error, rel_funcs, out_db)
418 2032 aaronmk
                if not xml_dom.is_empty(row_root):
419
                    assert xml_dom.has_one_child(row_root)
420 442 aaronmk
                    try:
421 982 aaronmk
                        sql.with_savepoint(out_db,
422 2032 aaronmk
                            lambda: db_xml.put(out_db, row_root.firstChild,
423 1850 aaronmk
                                row_ins_ct_ref, on_error))
424 449 aaronmk
                    except sql.DatabaseErrors, e: on_error(e)
425
426 982 aaronmk
            row_ct = process_inputs(root, row_ready)
427
            sys.stdout.write('Inserted '+str(row_ins_ct_ref[0])+
428 460 aaronmk
                ' new rows into database\n')
429 1868 aaronmk
430
            # Consume asynchronous tasks
431
            pool.main_loop()
432 53 aaronmk
        finally:
433 2927 aaronmk
            if out_db.connected(): out_db.db.close()
434 751 aaronmk
    else:
435 759 aaronmk
        def on_error(e): ex_tracker.track(e)
436 838 aaronmk
        def row_ready(row_num, input_row): pass
437 982 aaronmk
        row_ct = process_inputs(root, row_ready)
438 759 aaronmk
        xml_func.process(root, on_error)
439 751 aaronmk
        if out_is_xml_ref[0]:
440
            doc.writexml(sys.stdout, **xml_dom.prettyxml_config)
441
        else: # output is CSV
442
            raise NotImplementedError('CSV output not supported yet')
443 985 aaronmk
444 1868 aaronmk
    # Consume any asynchronous tasks not already consumed above
445 1862 aaronmk
    pool.main_loop()
446
447 982 aaronmk
    profiler.stop(row_ct)
448
    ex_tracker.add_iters(row_ct)
449 2439 aaronmk
    log('Processed '+str(row_ct)+' input rows')
450
    log(profiler.msg())
451
    log(ex_tracker.msg())
452 985 aaronmk
    ex_tracker.exit()
453 53 aaronmk
454 847 aaronmk
def main():
455
    try: main_()
456 1719 aaronmk
    except Parser.SyntaxError, e: raise SystemExit(str(e))
457 847 aaronmk
458 846 aaronmk
if __name__ == '__main__':
459 847 aaronmk
    profile_to = opts.get_env_var('profile_to', None)
460
    if profile_to != None:
461
        import cProfile
462 1584 aaronmk
        sys.stderr.write('Profiling to '+profile_to+'\n')
463 847 aaronmk
        cProfile.run(main.func_code, profile_to)
464
    else: main()