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