Project

General

Profile

1
#!/usr/bin/env python
2
# Maps one datasource to another, using a map spreadsheet if needed
3
# Exit status is the # of errors in the import, up to the maximum exit status
4
# 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 csv
8
import os.path
9
import sys
10
import xml.dom.minidom as minidom
11

    
12
sys.path.append(os.path.dirname(__file__)+"/../lib")
13

    
14
import exc
15
import opts
16
import Parser
17
import profiling
18
import sql
19
import strings
20
import term
21
import util
22
import xpath
23
import xml_dom
24
import xml_func
25

    
26
def metadata_value(name): return None # this feature has been removed
27

    
28
def main_():
29
    env_names = []
30
    def usage_err():
31
        raise SystemExit('Usage: '+opts.env_usage(env_names, True)+' '
32
            +sys.argv[0]+' [map_path...] [<input] [>output]')
33
    
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
    test = opts.env_flag('test', False, env_names)
48
    commit = opts.env_flag('commit', False, env_names) and not test
49
        # never commit in test mode
50
    redo = opts.env_flag('redo', test, env_names) and not commit
51
        # never redo in commit mode (manually run `make empty_db` instead)
52
    debug = opts.env_flag('debug', False, env_names)
53
    sql.run_raw_query.debug = debug
54
    verbose = debug or opts.env_flag('verbose', False, env_names)
55
    opts.get_env_var('profile_to', None, env_names) # add to env_names
56
    
57
    # Logging
58
    def log(msg, on=verbose):
59
        if on: sys.stderr.write(msg)
60
    def log_start(action, on=verbose): log(action+'...\n', on)
61
    
62
    # Parse args
63
    map_paths = sys.argv[1:]
64
    if map_paths == []:
65
        if in_is_db or not out_is_db: usage_err()
66
        else: map_paths = [None]
67
    
68
    def connect_db(db_config):
69
        log_start('Connecting to '+sql.db_config_str(db_config))
70
        return sql.connect(db_config)
71
    
72
    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
    out_is_xml_ref = [False]
78
    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
    
87
    def process_input(root, row_ready, map_path):
88
        '''Inputs datasource to XML tree, mapping if needed'''
89
        # Load map header
90
        in_is_xpaths = True
91
        out_is_xpaths = True
92
        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
            in_label_ref[0] = in_label
104
            update_in_label()
105
            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
        out_is_xml_ref[0] = out_is_xpaths and not out_is_db
117
        
118
        if in_is_xml:
119
            doc0 = minidom.parse(sys.stdin)
120
            doc0_root = doc0.documentElement
121
            if out_label == None: out_label = doc0_root.tagName
122
        
123
        def process_rows(process_row, rows):
124
            '''Processes input rows
125
            @param process_row(in_row, i)
126
            '''
127
            i = -1 # in case for loop does not execute
128
            for i, row in enumerate(rows):
129
                if i < start: continue
130
                if end != None and i >= end: break
131
                process_row(row, i)
132
                row_ready(i, row)
133
            row_ct = i-start+1
134
            return row_ct
135
        
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
                row_id = str(i)
142
                for in_, out in mappings:
143
                    value = metadata_value(in_)
144
                    if value == None:
145
                        log_start('Getting '+str(in_), debug)
146
                        value = get_value(in_, row)
147
                    if value != None: xpath.put_obj(root, out, row_id,
148
                        has_types, strings.cleanup(value))
149
            return process_rows(process_row, rows)
150
        
151
        if map_path == None:
152
            iter_ = xml_dom.NodeElemIter(doc0_root)
153
            util.skip(iter_, xml_dom.is_text) # skip metadata
154
            row_ct = process_rows(lambda row, i: root.appendChild(row), iter_)
155
        elif in_is_db:
156
            assert in_is_xpaths
157
            
158
            import db_xml
159
            
160
            in_root_xml = xpath.path2xml(in_root)
161
            for i, mapping in enumerate(mappings):
162
                in_, out = mapping
163
                if metadata_value(in_) == None:
164
                    mappings[i] = (xpath.path2xml(in_root+'/'+in_), out)
165
            
166
            in_db = connect_db(in_db_config)
167
            in_pkeys = {}
168
            def get_value(in_, row):
169
                pkey, = row
170
                in_ = in_.cloneNode(True) # don't modify orig value!
171
                xml_dom.set_id(xpath.get(in_, in_root)[0], pkey)
172
                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
            row_ct = map_rows(get_value, sql.rows(db_xml.get(in_db, in_root_xml,
176
                in_pkeys, end, 0)))
177
            in_db.close()
178
        elif in_is_xml:
179
            def get_value(in_, row):
180
                nodes = xpath.get(row, in_, allow_rooted=False)
181
                if nodes != []: return xml_dom.value(nodes[0])
182
                else: return None
183
            rows = xpath.get(doc0_root, in_root, limit=end)
184
            if rows == []: raise SystemExit('Map error: Root "'+in_root
185
                +'" not found in input')
186
            row_ct = map_rows(get_value, rows)
187
        else: # input is CSV
188
            map_ = dict(mappings)
189
            reader = csv.reader(sys.stdin)
190
            cols = reader.next()
191
            col_idxs = dict([(value, idx) for idx, value in enumerate(cols)])
192
            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
            
198
            def get_value(in_, row):
199
                value = row[in_]
200
                if value != '': return value
201
                else: return None
202
            row_ct = map_rows(get_value, reader)
203
        
204
        return row_ct
205
    
206
    def process_inputs(root, row_ready):
207
        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
    
212
    if out_is_db:
213
        import db_xml
214
        
215
        out_db = connect_db(out_db_config)
216
        out_pkeys = {}
217
        try:
218
            if redo: sql.empty_db(out_db)
219
            row_ins_ct_ref = [0]
220
            
221
            def row_ready(row_num, input_row):
222
                def on_error(e):
223
                    exc.add_msg(e, term.emph('row #:')+' '+str(row_num))
224
                    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
                    ex_tracker.track(e)
227
                
228
                xml_func.process(root, on_error)
229
                if not xml_dom.is_empty(root):
230
                    assert xml_dom.has_one_child(root)
231
                    try:
232
                        sql.with_savepoint(out_db,
233
                            lambda: db_xml.put(out_db, root.firstChild,
234
                                out_pkeys, row_ins_ct_ref, on_error))
235
                        if commit: out_db.commit()
236
                    except sql.DatabaseErrors, e: on_error(e)
237
                prep_root()
238
            
239
            row_ct = process_inputs(root, row_ready)
240
            sys.stdout.write('Inserted '+str(row_ins_ct_ref[0])+
241
                ' new rows into database\n')
242
        finally:
243
            out_db.rollback()
244
            out_db.close()
245
    else:
246
        def on_error(e): ex_tracker.track(e)
247
        def row_ready(row_num, input_row): pass
248
        row_ct = process_inputs(root, row_ready)
249
        xml_func.process(root, on_error)
250
        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
    
255
    profiler.stop(row_ct)
256
    ex_tracker.add_iters(row_ct)
257
    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
    ex_tracker.exit()
262

    
263
def main():
264
    try: main_()
265
    except Parser.SyntaxException, e: raise SystemExit(str(e))
266

    
267
if __name__ == '__main__':
268
    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()
(15-15/25)