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