Project

General

Profile

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