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 344 aaronmk
import exc
17 1714 aaronmk
import iters
18 1705 aaronmk
import maps
19 64 aaronmk
import opts
20 281 aaronmk
import Parser
21 982 aaronmk
import profiling
22 131 aaronmk
import sql
23 1758 aaronmk
import streams
24 715 aaronmk
import strings
25 828 aaronmk
import term
26 310 aaronmk
import util
27 1014 aaronmk
import xpath
28 133 aaronmk
import xml_dom
29 86 aaronmk
import xml_func
30 1713 aaronmk
import xml_parse
31 53 aaronmk
32 1857 aaronmk
class Pool:
33
    def apply_async(func, args=None, kw_args=None, callback=None):
34
        if args == None: args = ()
35
        if kwds == None: kwds = {}
36
        if callback == None: callback = lambda v: None
37
38
        callback(func(*args, **kw_args))
39
40 1404 aaronmk
def get_with_prefix(map_, prefixes, key):
41 1484 aaronmk
    '''Gets all entries for the given key with any of the given prefixes'''
42
    values = []
43 1681 aaronmk
    for key_ in strings.with_prefixes(['']+prefixes, key): # also with no prefix
44
        try: value = map_[key_]
45 1484 aaronmk
        except KeyError, e: continue # keep going
46
        values.append(value)
47
48
    if values != []: return values
49
    else: raise e # re-raise last KeyError
50 1404 aaronmk
51 1018 aaronmk
def metadata_value(name): return None # this feature has been removed
52 84 aaronmk
53 1360 aaronmk
def cleanup(val):
54
    if val == None: return val
55 1374 aaronmk
    return util.none_if(strings.cleanup(strings.ustr(val)), u'', u'\\N')
56 1360 aaronmk
57 847 aaronmk
def main_():
58 131 aaronmk
    env_names = []
59
    def usage_err():
60 944 aaronmk
        raise SystemExit('Usage: '+opts.env_usage(env_names, True)+' '
61 847 aaronmk
            +sys.argv[0]+' [map_path...] [<input] [>output]')
62 838 aaronmk
63 1570 aaronmk
    ## Get config from env vars
64 838 aaronmk
65 1570 aaronmk
    # Modes
66 946 aaronmk
    test = opts.env_flag('test', False, env_names)
67 947 aaronmk
    commit = opts.env_flag('commit', False, env_names) and not test
68 944 aaronmk
        # never commit in test mode
69 947 aaronmk
    redo = opts.env_flag('redo', test, env_names) and not commit
70
        # never redo in commit mode (manually run `make empty_db` instead)
71 1570 aaronmk
72
    # Ranges
73
    start = util.cast(int, opts.get_env_var('start', '0', env_names))
74 1572 aaronmk
    if test: end_default = 1
75
    else: end_default = None
76
    end = util.cast(int, util.none_if(
77
        opts.get_env_var('n', end_default, env_names), u''))
78 1570 aaronmk
    if end != None: end += start
79
80 1846 aaronmk
    # Optimization
81 1851 aaronmk
    if test: cpus_default = 0
82
    else: cpus_default = None
83 1856 aaronmk
    cpus = util.cast(int, util.none_if(opts.get_env_var('cpus', cpus_default,
84
        env_names), u''))
85 1846 aaronmk
86 1570 aaronmk
    # Debugging
87 946 aaronmk
    debug = opts.env_flag('debug', False, env_names)
88 859 aaronmk
    sql.run_raw_query.debug = debug
89 1574 aaronmk
    verbose = debug or opts.env_flag('verbose', not test, env_names)
90 944 aaronmk
    opts.get_env_var('profile_to', None, env_names) # add to env_names
91 131 aaronmk
92 1570 aaronmk
    # DB
93
    db_config_names = ['engine', 'host', 'user', 'password', 'database']
94
    def get_db_config(prefix):
95
        return opts.get_env_vars(db_config_names, prefix, env_names)
96
    in_db_config = get_db_config('in')
97
    out_db_config = get_db_config('out')
98
    in_is_db = 'engine' in in_db_config
99
    out_is_db = 'engine' in out_db_config
100
101
    ##
102
103 838 aaronmk
    # Logging
104 662 aaronmk
    def log(msg, on=verbose):
105
        if on: sys.stderr.write(msg)
106 859 aaronmk
    def log_start(action, on=verbose): log(action+'...\n', on)
107 662 aaronmk
108 53 aaronmk
    # Parse args
109 510 aaronmk
    map_paths = sys.argv[1:]
110 512 aaronmk
    if map_paths == []:
111
        if in_is_db or not out_is_db: usage_err()
112
        else: map_paths = [None]
113 53 aaronmk
114 646 aaronmk
    def connect_db(db_config):
115 662 aaronmk
        log_start('Connecting to '+sql.db_config_str(db_config))
116 1014 aaronmk
        return sql.connect(db_config)
117 646 aaronmk
118 1573 aaronmk
    if end != None: end_str = str(end-1)
119
    else: end_str = 'end'
120
    log_start('Processing input rows '+str(start)+'-'+end_str)
121
122 1014 aaronmk
    ex_tracker = exc.ExPercentTracker(iter_text='row')
123
    profiler = profiling.ItersProfiler(start_now=True, iter_text='row')
124
125 1856 aaronmk
    # Parallel processing
126 1857 aaronmk
    try:
127
        if cpus == 0: raise ImportError('Parallel processing turned off')
128
        import multiprocessing
129
        import multiprocessing.pool
130
    except ImportError, e:
131
        log_start('Not using parallel processing: '+str(e))
132
        job_server = Pool()
133
    else:
134
        if cpus == None: cpus = multiprocessing.cpu_count()
135
        log_start('Using '+str(cpus)+' CPUs')
136
        job_server = multiprocessing.pool.Pool(processes=cpus)
137 1846 aaronmk
138 1014 aaronmk
    doc = xml_dom.create_doc()
139
    root = doc.documentElement
140 751 aaronmk
    out_is_xml_ref = [False]
141 1014 aaronmk
    in_label_ref = [None]
142
    def update_in_label():
143
        if in_label_ref[0] != None:
144
            xpath.get(root, '/_ignore/inLabel="'+in_label_ref[0]+'"', True)
145
    def prep_root():
146
        root.clear()
147
        update_in_label()
148
    prep_root()
149 751 aaronmk
150 838 aaronmk
    def process_input(root, row_ready, map_path):
151 512 aaronmk
        '''Inputs datasource to XML tree, mapping if needed'''
152
        # Load map header
153
        in_is_xpaths = True
154 751 aaronmk
        out_is_xpaths = True
155 512 aaronmk
        out_label = None
156
        if map_path != None:
157
            metadata = []
158
            mappings = []
159
            stream = open(map_path, 'rb')
160
            reader = csv.reader(stream)
161
            in_label, out_label = reader.next()[:2]
162 1402 aaronmk
163 512 aaronmk
            def split_col_name(name):
164 1402 aaronmk
                label, sep, root = name.partition(':')
165
                label, sep2, prefixes_str = label.partition('[')
166
                prefixes_str = strings.remove_suffix(']', prefixes_str)
167
                prefixes = strings.split(',', prefixes_str)
168
                return label, sep != '', root, prefixes
169 1198 aaronmk
                    # extract datasrc from "datasrc[data_format]"
170 1402 aaronmk
171 1705 aaronmk
            in_label, in_root, prefixes = maps.col_info(in_label)
172
            in_is_xpaths = in_root != None
173 1014 aaronmk
            in_label_ref[0] = in_label
174
            update_in_label()
175 1705 aaronmk
            out_label, out_root = maps.col_info(out_label)[:2]
176
            out_is_xpaths = out_root != None
177 1841 aaronmk
            if out_is_xpaths: has_types = out_root.find('/*s/') >= 0
178 1705 aaronmk
                # outer elements are types
179 1402 aaronmk
180 512 aaronmk
            for row in reader:
181
                in_, out = row[:2]
182
                if out != '':
183
                    if out_is_xpaths: out = xpath.parse(out_root+out)
184
                    mappings.append((in_, out))
185 1402 aaronmk
186 512 aaronmk
            stream.close()
187
188
            root.ownerDocument.documentElement.tagName = out_label
189
        in_is_xml = in_is_xpaths and not in_is_db
190 751 aaronmk
        out_is_xml_ref[0] = out_is_xpaths and not out_is_db
191 56 aaronmk
192 1148 aaronmk
        def process_rows(process_row, rows):
193 838 aaronmk
            '''Processes input rows
194
            @param process_row(in_row, i)
195 297 aaronmk
            '''
196 1718 aaronmk
            i = 0
197
            while end == None or i < end:
198
                try: row = rows.next()
199
                except StopIteration: break # no more rows
200
                if i < start: continue # not at start row yet
201
202 838 aaronmk
                process_row(row, i)
203
                row_ready(i, row)
204 1718 aaronmk
                i += 1
205
            row_ct = i-start
206 982 aaronmk
            return row_ct
207 838 aaronmk
208
        def map_rows(get_value, rows):
209
            '''Maps input rows
210
            @param get_value(in_, row):str
211
            '''
212
            def process_row(row, i):
213 316 aaronmk
                row_id = str(i)
214
                for in_, out in mappings:
215
                    value = metadata_value(in_)
216 662 aaronmk
                    if value == None:
217 857 aaronmk
                        log_start('Getting '+str(in_), debug)
218 1360 aaronmk
                        value = cleanup(get_value(in_, row))
219 1348 aaronmk
                    if value != None:
220
                        log_start('Putting '+str(out), debug)
221 1360 aaronmk
                        xpath.put_obj(root, out, row_id, has_types, value)
222 982 aaronmk
            return process_rows(process_row, rows)
223 297 aaronmk
224 1403 aaronmk
        def map_table(col_names, rows):
225 1416 aaronmk
            col_names_ct = len(col_names)
226 1403 aaronmk
            col_idxs = util.list_flip(col_names)
227
228
            i = 0
229
            while i < len(mappings): # mappings len changes in loop
230
                in_, out = mappings[i]
231
                if metadata_value(in_) == None:
232 1404 aaronmk
                    try: mappings[i] = (
233
                        get_with_prefix(col_idxs, prefixes, in_), out)
234
                    except KeyError:
235
                        del mappings[i]
236
                        continue # keep i the same
237 1403 aaronmk
                i += 1
238
239
            def get_value(in_, row):
240 1484 aaronmk
                return util.coalesce(*util.list_subset(row.list, in_))
241 1416 aaronmk
            def wrap_row(row):
242
                return util.ListDict(util.list_as_length(row, col_names_ct),
243
                    col_names, col_idxs) # handle CSV rows of different lengths
244 1403 aaronmk
245
            return map_rows(get_value, util.WrapIter(wrap_row, rows))
246
247 1758 aaronmk
        stdin = streams.LineCountStream(sys.stdin)
248
        def on_error(e):
249
            exc.add_msg(e, term.emph('input line #:')+' '+str(stdin.line_num))
250
            ex_tracker.track(e)
251
252 1700 aaronmk
        if in_is_db:
253 130 aaronmk
            assert in_is_xpaths
254 126 aaronmk
255 1136 aaronmk
            in_db = connect_db(in_db_config)
256
            cur = sql.select(in_db, table=in_root, fields=None, conds=None,
257
                limit=end, start=0)
258 1403 aaronmk
            row_ct = map_table(list(sql.col_names(cur)), sql.rows(cur))
259 1136 aaronmk
260 1849 aaronmk
            in_db.db.close()
261 161 aaronmk
        elif in_is_xml:
262 1715 aaronmk
            def get_rows(doc2rows):
263 1758 aaronmk
                return iters.flatten(itertools.imap(doc2rows,
264
                    xml_parse.docs_iter(stdin, on_error)))
265 1715 aaronmk
266 1714 aaronmk
            if map_path == None:
267 1715 aaronmk
                def doc2rows(in_xml_root):
268 1701 aaronmk
                    iter_ = xml_dom.NodeElemIter(in_xml_root)
269
                    util.skip(iter_, xml_dom.is_text) # skip metadata
270 1715 aaronmk
                    return iter_
271
272
                row_ct = process_rows(lambda row, i: root.appendChild(row),
273
                    get_rows(doc2rows))
274 1714 aaronmk
            else:
275 1715 aaronmk
                def doc2rows(in_xml_root):
276 1701 aaronmk
                    rows = xpath.get(in_xml_root, in_root, limit=end)
277 1714 aaronmk
                    if rows == []: raise SystemExit('Map error: Root "'
278
                        +in_root+'" not found in input')
279
                    return rows
280
281
                def get_value(in_, row):
282
                    in_ = './{'+(','.join(strings.with_prefixes(
283
                        ['']+prefixes, in_)))+'}' # also with no prefix
284
                    nodes = xpath.get(row, in_, allow_rooted=False)
285
                    if nodes != []: return xml_dom.value(nodes[0])
286
                    else: return None
287
288 1715 aaronmk
                row_ct = map_rows(get_value, get_rows(doc2rows))
289 56 aaronmk
        else: # input is CSV
290 133 aaronmk
            map_ = dict(mappings)
291 1389 aaronmk
            reader, col_names = csvs.reader_and_header(sys.stdin)
292 1403 aaronmk
            row_ct = map_table(col_names, reader)
293 982 aaronmk
294
        return row_ct
295 53 aaronmk
296 838 aaronmk
    def process_inputs(root, row_ready):
297 982 aaronmk
        row_ct = 0
298
        for map_path in map_paths:
299
            row_ct += process_input(root, row_ready, map_path)
300
        return row_ct
301 512 aaronmk
302 130 aaronmk
    if out_is_db:
303 53 aaronmk
        import db_xml
304
305 646 aaronmk
        out_db = connect_db(out_db_config)
306 53 aaronmk
        try:
307 947 aaronmk
            if redo: sql.empty_db(out_db)
308 982 aaronmk
            row_ins_ct_ref = [0]
309 449 aaronmk
310 838 aaronmk
            def row_ready(row_num, input_row):
311 452 aaronmk
                def on_error(e):
312 835 aaronmk
                    exc.add_msg(e, term.emph('row #:')+' '+str(row_num))
313 828 aaronmk
                    exc.add_msg(e, term.emph('input row:')+'\n'+str(input_row))
314
                    exc.add_msg(e, term.emph('output row:')+'\n'+str(root))
315 1617 aaronmk
                    ex_tracker.track(e, row_num)
316 452 aaronmk
317 449 aaronmk
                xml_func.process(root, on_error)
318 442 aaronmk
                if not xml_dom.is_empty(root):
319
                    assert xml_dom.has_one_child(root)
320
                    try:
321 982 aaronmk
                        sql.with_savepoint(out_db,
322
                            lambda: db_xml.put(out_db, root.firstChild,
323 1850 aaronmk
                                row_ins_ct_ref, on_error))
324 1849 aaronmk
                        if commit: out_db.db.commit()
325 449 aaronmk
                    except sql.DatabaseErrors, e: on_error(e)
326 1010 aaronmk
                prep_root()
327 449 aaronmk
328 982 aaronmk
            row_ct = process_inputs(root, row_ready)
329
            sys.stdout.write('Inserted '+str(row_ins_ct_ref[0])+
330 460 aaronmk
                ' new rows into database\n')
331 53 aaronmk
        finally:
332 1849 aaronmk
            out_db.db.rollback()
333
            out_db.db.close()
334 751 aaronmk
    else:
335 759 aaronmk
        def on_error(e): ex_tracker.track(e)
336 838 aaronmk
        def row_ready(row_num, input_row): pass
337 982 aaronmk
        row_ct = process_inputs(root, row_ready)
338 759 aaronmk
        xml_func.process(root, on_error)
339 751 aaronmk
        if out_is_xml_ref[0]:
340
            doc.writexml(sys.stdout, **xml_dom.prettyxml_config)
341
        else: # output is CSV
342
            raise NotImplementedError('CSV output not supported yet')
343 985 aaronmk
344 982 aaronmk
    profiler.stop(row_ct)
345
    ex_tracker.add_iters(row_ct)
346 990 aaronmk
    if verbose:
347
        sys.stderr.write('Processed '+str(row_ct)+' input rows\n')
348
        sys.stderr.write(profiler.msg()+'\n')
349
        sys.stderr.write(ex_tracker.msg()+'\n')
350 985 aaronmk
    ex_tracker.exit()
351 53 aaronmk
352 847 aaronmk
def main():
353
    try: main_()
354 1719 aaronmk
    except Parser.SyntaxError, e: raise SystemExit(str(e))
355 847 aaronmk
356 846 aaronmk
if __name__ == '__main__':
357 847 aaronmk
    profile_to = opts.get_env_var('profile_to', None)
358
    if profile_to != None:
359
        import cProfile
360 1584 aaronmk
        sys.stderr.write('Profiling to '+profile_to+'\n')
361 847 aaronmk
        cProfile.run(main.func_code, profile_to)
362
    else: main()