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