Project

General

Profile

1
#!/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
import xml.dom.minidom as minidom
9

    
10
sys.path.append(os.path.dirname(__file__)+"/../lib")
11

    
12
import exc
13
import opts
14
import Parser
15
import sql
16
import strings
17
import term
18
import util
19
import xml_dom
20
import xml_func
21

    
22
def metadata_value(name):
23
    if util.is_str(name) and name.startswith(':'): return name[1:]
24
    else: return None
25

    
26
def main_():
27
    ex_tracker = exc.ExPercentTracker()
28
    
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
        db = sql.connect(db_config)
71
        return db
72
    
73
    out_is_xml_ref = [False]
74
    
75
    def process_input(root, row_ready, map_path):
76
        '''Inputs datasource to XML tree, mapping if needed'''
77
        # Load map header
78
        in_is_xpaths = True
79
        out_is_xpaths = True
80
        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
        out_is_xml_ref[0] = out_is_xpaths and not out_is_db
108
        
109
        if in_is_xml:
110
            doc0 = minidom.parse(sys.stdin)
111
            if out_label == None: out_label = doc0.documentElement.tagName
112
        
113
        def process_rows(process_row, rows):
114
            '''Processes input rows
115
            @param process_row(in_row, i)
116
            '''
117
            i = -1 # in case for loop does not execute
118
            for i, row in enumerate(rows):
119
                if i < start: continue
120
                if end != None and i >= end: break
121
                process_row(row, i)
122
                row_ready(i, row)
123
            
124
            row_ct = i-start+1
125
            ex_tracker.add_iters(row_ct)
126
            sys.stderr.write('Processed '+str(row_ct)+' input rows\n')
127
        
128
        def map_rows(get_value, rows):
129
            '''Maps input rows
130
            @param get_value(in_, row):str
131
            '''
132
            def process_row(row, i):
133
                row_id = str(i)
134
                for in_, out in mappings:
135
                    value = metadata_value(in_)
136
                    if value == None:
137
                        log_start('Getting '+str(in_), debug)
138
                        value = get_value(in_, row)
139
                    if value != None: xpath.put_obj(root, out, row_id,
140
                        has_types, strings.cleanup(value))
141
            process_rows(process_row, rows)
142
        
143
        if map_path == None:
144
            iter_ = xml_dom.NodeElemIter(doc0.documentElement)
145
            util.skip(iter_, xml_dom.is_text) # skip metadata
146
            process_rows(lambda row, i: root.appendChild(row), iter_)
147
        elif in_is_db:
148
            assert in_is_xpaths
149
            
150
            import db_xml
151
            
152
            in_root_xml = xpath.path2xml(in_root)
153
            for i, mapping in enumerate(mappings):
154
                in_, out = mapping
155
                if metadata_value(in_) == None:
156
                    mappings[i] = (xpath.path2xml(in_root+'/'+in_), out)
157
            
158
            in_db = connect_db(in_db_config)
159
            in_pkeys = {}
160
            def get_value(in_, row):
161
                pkey, = row
162
                in_ = in_.cloneNode(True) # don't modify orig value!
163
                xml_dom.set_id(xpath.get(in_, in_root)[0], pkey)
164
                value = sql.value_or_none(db_xml.get(in_db, in_, in_pkeys))
165
                if value != None: return str(value)
166
                else: return None
167
            map_rows(get_value, sql.rows(db_xml.get(in_db, in_root_xml,
168
                in_pkeys, end, 0)))
169
            in_db.close()
170
        elif in_is_xml:
171
            def get_value(in_, row):
172
                nodes = xpath.get(row, in_)
173
                if nodes != []: return xml_dom.value(nodes[0])
174
                else: return None
175
            rows = xpath.get(doc0.documentElement, in_root, limit=end)
176
            if rows == []: raise SystemExit('Map error: Root "'+in_root
177
                +'" not found in input')
178
            map_rows(get_value, rows)
179
        else: # input is CSV
180
            map_ = dict(mappings)
181
            reader = csv.reader(sys.stdin)
182
            cols = reader.next()
183
            col_idxs = dict([(value, idx) for idx, value in enumerate(cols)])
184
            for i, mapping in enumerate(mappings):
185
                in_, out = mapping
186
                if metadata_value(in_) == None:
187
                    try: mappings[i] = (col_idxs[in_], out)
188
                    except KeyError: pass
189
            
190
            def get_value(in_, row):
191
                value = row[in_]
192
                if value != '': return value
193
                else: return None
194
            map_rows(get_value, reader)
195
    
196
    def process_inputs(root, row_ready):
197
        for map_path in map_paths: process_input(root, row_ready, map_path)
198
    
199
    # Output XML tree
200
    doc = xml_dom.create_doc()
201
    root = doc.documentElement
202
    if out_is_db:
203
        import db_xml
204
        
205
        out_db = connect_db(out_db_config)
206
        out_pkeys = {}
207
        try:
208
            if redo: sql.empty_db(out_db)
209
            row_ct_ref = [0]
210
            
211
            def row_ready(row_num, input_row):
212
                def on_error(e):
213
                    exc.add_msg(e, term.emph('row #:')+' '+str(row_num))
214
                    exc.add_msg(e, term.emph('input row:')+'\n'+str(input_row))
215
                    exc.add_msg(e, term.emph('output row:')+'\n'+str(root))
216
                    ex_tracker.track(e)
217
                
218
                xml_func.process(root, on_error)
219
                if not xml_dom.is_empty(root):
220
                    assert xml_dom.has_one_child(root)
221
                    try:
222
                        sql.with_savepoint(out_db, lambda: db_xml.put(out_db,
223
                            root.firstChild, out_pkeys, row_ct_ref, on_error))
224
                        if commit: out_db.commit()
225
                    except sql.DatabaseErrors, e: on_error(e)
226
                root.clear()
227
            
228
            process_inputs(root, row_ready)
229
            sys.stdout.write('Inserted '+str(row_ct_ref[0])+
230
                ' new rows into database\n')
231
        finally:
232
            out_db.rollback()
233
            out_db.close()
234
    else:
235
        def on_error(e): ex_tracker.track(e)
236
        def row_ready(row_num, input_row): pass
237
        process_inputs(root, row_ready)
238
        xml_func.process(root, on_error)
239
        if out_is_xml_ref[0]:
240
            doc.writexml(sys.stdout, **xml_dom.prettyxml_config)
241
        else: # output is CSV
242
            raise NotImplementedError('CSV output not supported yet')
243

    
244
def main():
245
    try: main_()
246
    except Parser.SyntaxException, e: raise SystemExit(str(e))
247

    
248
if __name__ == '__main__':
249
    profile_to = opts.get_env_var('profile_to', None)
250
    if profile_to != None:
251
        import cProfile
252
        cProfile.run(main.func_code, profile_to)
253
    else: main()
(15-15/25)