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
                return name, sep != '', root
102
            in_label, in_is_xpaths, in_root = split_col_name(in_label)
103 1014 aaronmk
            in_label_ref[0] = in_label
104
            update_in_label()
105 512 aaronmk
            out_label, out_is_xpaths, out_root = split_col_name(out_label)
106
            has_types = out_root.startswith('/*s/') # outer elements are types
107
            for row in reader:
108
                in_, out = row[:2]
109
                if out != '':
110
                    if out_is_xpaths: out = xpath.parse(out_root+out)
111
                    mappings.append((in_, out))
112
            stream.close()
113
114
            root.ownerDocument.documentElement.tagName = out_label
115
        in_is_xml = in_is_xpaths and not in_is_db
116 751 aaronmk
        out_is_xml_ref[0] = out_is_xpaths and not out_is_db
117 56 aaronmk
118 512 aaronmk
        if in_is_xml:
119
            doc0 = minidom.parse(sys.stdin)
120 1006 aaronmk
            doc0_root = doc0.documentElement
121
            if out_label == None: out_label = doc0_root.tagName
122 53 aaronmk
123 838 aaronmk
        def process_rows(process_row, rows):
124
            '''Processes input rows
125
            @param process_row(in_row, i)
126 297 aaronmk
            '''
127 838 aaronmk
            i = -1 # in case for loop does not execute
128 314 aaronmk
            for i, row in enumerate(rows):
129 838 aaronmk
                if i < start: continue
130
                if end != None and i >= end: break
131
                process_row(row, i)
132
                row_ready(i, row)
133 978 aaronmk
            row_ct = i-start+1
134 982 aaronmk
            return row_ct
135 838 aaronmk
136
        def map_rows(get_value, rows):
137
            '''Maps input rows
138
            @param get_value(in_, row):str
139
            '''
140
            def process_row(row, i):
141 316 aaronmk
                row_id = str(i)
142
                for in_, out in mappings:
143
                    value = metadata_value(in_)
144 662 aaronmk
                    if value == None:
145 857 aaronmk
                        log_start('Getting '+str(in_), debug)
146 662 aaronmk
                        value = get_value(in_, row)
147 715 aaronmk
                    if value != None: xpath.put_obj(root, out, row_id,
148
                        has_types, strings.cleanup(value))
149 982 aaronmk
            return process_rows(process_row, rows)
150 297 aaronmk
151 310 aaronmk
        if map_path == None:
152 1006 aaronmk
            iter_ = xml_dom.NodeElemIter(doc0_root)
153 310 aaronmk
            util.skip(iter_, xml_dom.is_text) # skip metadata
154 982 aaronmk
            row_ct = process_rows(lambda row, i: root.appendChild(row), iter_)
155 309 aaronmk
        elif in_is_db:
156 130 aaronmk
            assert in_is_xpaths
157 126 aaronmk
158 117 aaronmk
            import db_xml
159
160 161 aaronmk
            in_root_xml = xpath.path2xml(in_root)
161 164 aaronmk
            for i, mapping in enumerate(mappings):
162
                in_, out = mapping
163
                if metadata_value(in_) == None:
164 168 aaronmk
                    mappings[i] = (xpath.path2xml(in_root+'/'+in_), out)
165 126 aaronmk
166 646 aaronmk
            in_db = connect_db(in_db_config)
167 133 aaronmk
            in_pkeys = {}
168 297 aaronmk
            def get_value(in_, row):
169 167 aaronmk
                pkey, = row
170 297 aaronmk
                in_ = in_.cloneNode(True) # don't modify orig value!
171 886 aaronmk
                xml_dom.set_id(xpath.get(in_, in_root)[0], pkey)
172 297 aaronmk
                value = sql.value_or_none(db_xml.get(in_db, in_, in_pkeys))
173
                if value != None: return str(value)
174
                else: return None
175 982 aaronmk
            row_ct = map_rows(get_value, sql.rows(db_xml.get(in_db, in_root_xml,
176 865 aaronmk
                in_pkeys, end, 0)))
177 117 aaronmk
            in_db.close()
178 161 aaronmk
        elif in_is_xml:
179 297 aaronmk
            def get_value(in_, row):
180 1005 aaronmk
                nodes = xpath.get(row, in_, allow_rooted=False)
181 886 aaronmk
                if nodes != []: return xml_dom.value(nodes[0])
182 297 aaronmk
                else: return None
183 1006 aaronmk
            rows = xpath.get(doc0_root, in_root, limit=end)
184 886 aaronmk
            if rows == []: raise SystemExit('Map error: Root "'+in_root
185 883 aaronmk
                +'" not found in input')
186 982 aaronmk
            row_ct = map_rows(get_value, rows)
187 56 aaronmk
        else: # input is CSV
188 133 aaronmk
            map_ = dict(mappings)
189 59 aaronmk
            reader = csv.reader(sys.stdin)
190 84 aaronmk
            cols = reader.next()
191 162 aaronmk
            col_idxs = dict([(value, idx) for idx, value in enumerate(cols)])
192 164 aaronmk
            for i, mapping in enumerate(mappings):
193
                in_, out = mapping
194
                if metadata_value(in_) == None:
195
                    try: mappings[i] = (col_idxs[in_], out)
196
                    except KeyError: pass
197 162 aaronmk
198 297 aaronmk
            def get_value(in_, row):
199
                value = row[in_]
200
                if value != '': return value
201
                else: return None
202 982 aaronmk
            row_ct = map_rows(get_value, reader)
203
204
        return row_ct
205 53 aaronmk
206 838 aaronmk
    def process_inputs(root, row_ready):
207 982 aaronmk
        row_ct = 0
208
        for map_path in map_paths:
209
            row_ct += process_input(root, row_ready, map_path)
210
        return row_ct
211 512 aaronmk
212 130 aaronmk
    if out_is_db:
213 53 aaronmk
        import db_xml
214
215 646 aaronmk
        out_db = connect_db(out_db_config)
216 310 aaronmk
        out_pkeys = {}
217 53 aaronmk
        try:
218 947 aaronmk
            if redo: sql.empty_db(out_db)
219 982 aaronmk
            row_ins_ct_ref = [0]
220 449 aaronmk
221 838 aaronmk
            def row_ready(row_num, input_row):
222 452 aaronmk
                def on_error(e):
223 835 aaronmk
                    exc.add_msg(e, term.emph('row #:')+' '+str(row_num))
224 828 aaronmk
                    exc.add_msg(e, term.emph('input row:')+'\n'+str(input_row))
225
                    exc.add_msg(e, term.emph('output row:')+'\n'+str(root))
226 452 aaronmk
                    ex_tracker.track(e)
227
228 449 aaronmk
                xml_func.process(root, on_error)
229 442 aaronmk
                if not xml_dom.is_empty(root):
230
                    assert xml_dom.has_one_child(root)
231
                    try:
232 982 aaronmk
                        sql.with_savepoint(out_db,
233
                            lambda: db_xml.put(out_db, root.firstChild,
234
                                out_pkeys, row_ins_ct_ref, on_error))
235 442 aaronmk
                        if commit: out_db.commit()
236 449 aaronmk
                    except sql.DatabaseErrors, e: on_error(e)
237 1010 aaronmk
                prep_root()
238 449 aaronmk
239 982 aaronmk
            row_ct = process_inputs(root, row_ready)
240
            sys.stdout.write('Inserted '+str(row_ins_ct_ref[0])+
241 460 aaronmk
                ' new rows into database\n')
242 53 aaronmk
        finally:
243 133 aaronmk
            out_db.rollback()
244
            out_db.close()
245 751 aaronmk
    else:
246 759 aaronmk
        def on_error(e): ex_tracker.track(e)
247 838 aaronmk
        def row_ready(row_num, input_row): pass
248 982 aaronmk
        row_ct = process_inputs(root, row_ready)
249 759 aaronmk
        xml_func.process(root, on_error)
250 751 aaronmk
        if out_is_xml_ref[0]:
251
            doc.writexml(sys.stdout, **xml_dom.prettyxml_config)
252
        else: # output is CSV
253
            raise NotImplementedError('CSV output not supported yet')
254 985 aaronmk
255 982 aaronmk
    profiler.stop(row_ct)
256
    ex_tracker.add_iters(row_ct)
257 990 aaronmk
    if verbose:
258
        sys.stderr.write('Processed '+str(row_ct)+' input rows\n')
259
        sys.stderr.write(profiler.msg()+'\n')
260
        sys.stderr.write(ex_tracker.msg()+'\n')
261 985 aaronmk
    ex_tracker.exit()
262 53 aaronmk
263 847 aaronmk
def main():
264
    try: main_()
265
    except Parser.SyntaxException, e: raise SystemExit(str(e))
266
267 846 aaronmk
if __name__ == '__main__':
268 847 aaronmk
    profile_to = opts.get_env_var('profile_to', None)
269
    if profile_to != None:
270
        import cProfile
271
        cProfile.run(main.func_code, profile_to)
272
    else: main()