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 131 aaronmk
import sql
16 715 aaronmk
import strings
17 828 aaronmk
import term
18 310 aaronmk
import util
19 133 aaronmk
import xml_dom
20 86 aaronmk
import xml_func
21 53 aaronmk
22 84 aaronmk
def metadata_value(name):
23 892 aaronmk
    if util.is_str(name) and name.startswith(':'): return name[1:]
24 84 aaronmk
    else: return None
25
26 847 aaronmk
def main_():
27 978 aaronmk
    ex_tracker = exc.ExPercentTracker()
28 434 aaronmk
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
            if out_label == None: out_label = doc0.documentElement.tagName
112 53 aaronmk
113 838 aaronmk
        def process_rows(process_row, rows):
114
            '''Processes input rows
115
            @param process_row(in_row, i)
116 297 aaronmk
            '''
117 838 aaronmk
            i = -1 # in case for loop does not execute
118 314 aaronmk
            for i, row in enumerate(rows):
119 838 aaronmk
                if i < start: continue
120
                if end != None and i >= end: break
121
                process_row(row, i)
122
                row_ready(i, row)
123 978 aaronmk
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 838 aaronmk
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 316 aaronmk
                row_id = str(i)
134
                for in_, out in mappings:
135
                    value = metadata_value(in_)
136 662 aaronmk
                    if value == None:
137 857 aaronmk
                        log_start('Getting '+str(in_), debug)
138 662 aaronmk
                        value = get_value(in_, row)
139 715 aaronmk
                    if value != None: xpath.put_obj(root, out, row_id,
140
                        has_types, strings.cleanup(value))
141 838 aaronmk
            process_rows(process_row, rows)
142 297 aaronmk
143 310 aaronmk
        if map_path == None:
144
            iter_ = xml_dom.NodeElemIter(doc0.documentElement)
145
            util.skip(iter_, xml_dom.is_text) # skip metadata
146 838 aaronmk
            process_rows(lambda row, i: root.appendChild(row), iter_)
147 309 aaronmk
        elif in_is_db:
148 130 aaronmk
            assert in_is_xpaths
149 126 aaronmk
150 117 aaronmk
            import db_xml
151
152 161 aaronmk
            in_root_xml = xpath.path2xml(in_root)
153 164 aaronmk
            for i, mapping in enumerate(mappings):
154
                in_, out = mapping
155
                if metadata_value(in_) == None:
156 168 aaronmk
                    mappings[i] = (xpath.path2xml(in_root+'/'+in_), out)
157 126 aaronmk
158 646 aaronmk
            in_db = connect_db(in_db_config)
159 133 aaronmk
            in_pkeys = {}
160 297 aaronmk
            def get_value(in_, row):
161 167 aaronmk
                pkey, = row
162 297 aaronmk
                in_ = in_.cloneNode(True) # don't modify orig value!
163 886 aaronmk
                xml_dom.set_id(xpath.get(in_, in_root)[0], pkey)
164 297 aaronmk
                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 838 aaronmk
            map_rows(get_value, sql.rows(db_xml.get(in_db, in_root_xml,
168 865 aaronmk
                in_pkeys, end, 0)))
169 117 aaronmk
            in_db.close()
170 161 aaronmk
        elif in_is_xml:
171 297 aaronmk
            def get_value(in_, row):
172 886 aaronmk
                nodes = xpath.get(row, in_)
173
                if nodes != []: return xml_dom.value(nodes[0])
174 297 aaronmk
                else: return None
175 890 aaronmk
            rows = xpath.get(doc0.documentElement, in_root, limit=end)
176 886 aaronmk
            if rows == []: raise SystemExit('Map error: Root "'+in_root
177 883 aaronmk
                +'" not found in input')
178 890 aaronmk
            map_rows(get_value, rows)
179 56 aaronmk
        else: # input is CSV
180 133 aaronmk
            map_ = dict(mappings)
181 59 aaronmk
            reader = csv.reader(sys.stdin)
182 84 aaronmk
            cols = reader.next()
183 162 aaronmk
            col_idxs = dict([(value, idx) for idx, value in enumerate(cols)])
184 164 aaronmk
            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 162 aaronmk
190 297 aaronmk
            def get_value(in_, row):
191
                value = row[in_]
192
                if value != '': return value
193
                else: return None
194 838 aaronmk
            map_rows(get_value, reader)
195 53 aaronmk
196 838 aaronmk
    def process_inputs(root, row_ready):
197
        for map_path in map_paths: process_input(root, row_ready, map_path)
198 512 aaronmk
199 53 aaronmk
    # Output XML tree
200 512 aaronmk
    doc = xml_dom.create_doc()
201 316 aaronmk
    root = doc.documentElement
202 130 aaronmk
    if out_is_db:
203 53 aaronmk
        import db_xml
204
205 646 aaronmk
        out_db = connect_db(out_db_config)
206 310 aaronmk
        out_pkeys = {}
207 53 aaronmk
        try:
208 947 aaronmk
            if redo: sql.empty_db(out_db)
209 53 aaronmk
            row_ct_ref = [0]
210 449 aaronmk
211 838 aaronmk
            def row_ready(row_num, input_row):
212 452 aaronmk
                def on_error(e):
213 835 aaronmk
                    exc.add_msg(e, term.emph('row #:')+' '+str(row_num))
214 828 aaronmk
                    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 452 aaronmk
                    ex_tracker.track(e)
217
218 449 aaronmk
                xml_func.process(root, on_error)
219 442 aaronmk
                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 462 aaronmk
                            root.firstChild, out_pkeys, row_ct_ref, on_error))
224 442 aaronmk
                        if commit: out_db.commit()
225 449 aaronmk
                    except sql.DatabaseErrors, e: on_error(e)
226 316 aaronmk
                root.clear()
227 449 aaronmk
228 838 aaronmk
            process_inputs(root, row_ready)
229 717 aaronmk
            sys.stdout.write('Inserted '+str(row_ct_ref[0])+
230 460 aaronmk
                ' new rows into database\n')
231 53 aaronmk
        finally:
232 133 aaronmk
            out_db.rollback()
233
            out_db.close()
234 751 aaronmk
    else:
235 759 aaronmk
        def on_error(e): ex_tracker.track(e)
236 838 aaronmk
        def row_ready(row_num, input_row): pass
237
        process_inputs(root, row_ready)
238 759 aaronmk
        xml_func.process(root, on_error)
239 751 aaronmk
        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 53 aaronmk
244 847 aaronmk
def main():
245
    try: main_()
246
    except Parser.SyntaxException, e: raise SystemExit(str(e))
247
248 846 aaronmk
if __name__ == '__main__':
249 847 aaronmk
    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()