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