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