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