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