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