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