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