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