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 434 aaronmk
    ex_tracker = exc.ExTracker()
28
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 944 aaronmk
    test = opts.env_flag('test', env_names)
48
    commit = not test and opts.env_flag('commit', env_names)
49
        # never commit in test mode
50
    debug = opts.env_flag('debug', env_names)
51 859 aaronmk
    sql.run_raw_query.debug = debug
52 944 aaronmk
    verbose = debug or opts.env_flag('verbose', env_names)
53
    opts.get_env_var('profile_to', None, env_names) # add to env_names
54 131 aaronmk
55 838 aaronmk
    # Logging
56 662 aaronmk
    def log(msg, on=verbose):
57
        if on: sys.stderr.write(msg)
58 859 aaronmk
    def log_start(action, on=verbose): log(action+'...\n', on)
59 662 aaronmk
60 53 aaronmk
    # Parse args
61 510 aaronmk
    map_paths = sys.argv[1:]
62 512 aaronmk
    if map_paths == []:
63
        if in_is_db or not out_is_db: usage_err()
64
        else: map_paths = [None]
65 53 aaronmk
66 646 aaronmk
    def connect_db(db_config):
67 662 aaronmk
        log_start('Connecting to '+sql.db_config_str(db_config))
68 646 aaronmk
        db = sql.connect(db_config)
69
        return db
70
71 751 aaronmk
    out_is_xml_ref = [False]
72
73 838 aaronmk
    def process_input(root, row_ready, map_path):
74 512 aaronmk
        '''Inputs datasource to XML tree, mapping if needed'''
75
        # Load map header
76
        in_is_xpaths = True
77 751 aaronmk
        out_is_xpaths = True
78 512 aaronmk
        out_label = None
79
        if map_path != None:
80
            import copy
81
            import csv
82
83
            import xpath
84
85
            metadata = []
86
            mappings = []
87
            stream = open(map_path, 'rb')
88
            reader = csv.reader(stream)
89
            in_label, out_label = reader.next()[:2]
90
            def split_col_name(name):
91
                name, sep, root = name.partition(':')
92
                return name, sep != '', root
93
            in_label, in_is_xpaths, in_root = split_col_name(in_label)
94
            out_label, out_is_xpaths, out_root = split_col_name(out_label)
95
            has_types = out_root.startswith('/*s/') # outer elements are types
96
            for row in reader:
97
                in_, out = row[:2]
98
                if out != '':
99
                    if out_is_xpaths: out = xpath.parse(out_root+out)
100
                    mappings.append((in_, out))
101
            stream.close()
102
103
            root.ownerDocument.documentElement.tagName = out_label
104
        in_is_xml = in_is_xpaths and not in_is_db
105 751 aaronmk
        out_is_xml_ref[0] = out_is_xpaths and not out_is_db
106 56 aaronmk
107 512 aaronmk
        if in_is_xml:
108
            doc0 = minidom.parse(sys.stdin)
109
            if out_label == None: out_label = doc0.documentElement.tagName
110 53 aaronmk
111 838 aaronmk
        def process_rows(process_row, rows):
112
            '''Processes input rows
113
            @param process_row(in_row, i)
114 297 aaronmk
            '''
115 838 aaronmk
            i = -1 # in case for loop does not execute
116 314 aaronmk
            for i, row in enumerate(rows):
117 838 aaronmk
                if i < start: continue
118
                if end != None and i >= end: break
119
                process_row(row, i)
120
                row_ready(i, row)
121
            sys.stderr.write('Processed '+str(i-start+1)+' input rows\n')
122
123
        def map_rows(get_value, rows):
124
            '''Maps input rows
125
            @param get_value(in_, row):str
126
            '''
127
            def process_row(row, i):
128 316 aaronmk
                row_id = str(i)
129
                for in_, out in mappings:
130
                    value = metadata_value(in_)
131 662 aaronmk
                    if value == None:
132 857 aaronmk
                        log_start('Getting '+str(in_), debug)
133 662 aaronmk
                        value = get_value(in_, row)
134 715 aaronmk
                    if value != None: xpath.put_obj(root, out, row_id,
135
                        has_types, strings.cleanup(value))
136 838 aaronmk
            process_rows(process_row, rows)
137 297 aaronmk
138 310 aaronmk
        if map_path == None:
139
            iter_ = xml_dom.NodeElemIter(doc0.documentElement)
140
            util.skip(iter_, xml_dom.is_text) # skip metadata
141 838 aaronmk
            process_rows(lambda row, i: root.appendChild(row), iter_)
142 309 aaronmk
        elif in_is_db:
143 130 aaronmk
            assert in_is_xpaths
144 126 aaronmk
145 117 aaronmk
            import db_xml
146
147 161 aaronmk
            in_root_xml = xpath.path2xml(in_root)
148 164 aaronmk
            for i, mapping in enumerate(mappings):
149
                in_, out = mapping
150
                if metadata_value(in_) == None:
151 168 aaronmk
                    mappings[i] = (xpath.path2xml(in_root+'/'+in_), out)
152 126 aaronmk
153 646 aaronmk
            in_db = connect_db(in_db_config)
154 133 aaronmk
            in_pkeys = {}
155 297 aaronmk
            def get_value(in_, row):
156 167 aaronmk
                pkey, = row
157 297 aaronmk
                in_ = in_.cloneNode(True) # don't modify orig value!
158 886 aaronmk
                xml_dom.set_id(xpath.get(in_, in_root)[0], pkey)
159 297 aaronmk
                value = sql.value_or_none(db_xml.get(in_db, in_, in_pkeys))
160
                if value != None: return str(value)
161
                else: return None
162 838 aaronmk
            map_rows(get_value, sql.rows(db_xml.get(in_db, in_root_xml,
163 865 aaronmk
                in_pkeys, end, 0)))
164 117 aaronmk
            in_db.close()
165 161 aaronmk
        elif in_is_xml:
166 297 aaronmk
            def get_value(in_, row):
167 886 aaronmk
                nodes = xpath.get(row, in_)
168
                if nodes != []: return xml_dom.value(nodes[0])
169 297 aaronmk
                else: return None
170 890 aaronmk
            rows = xpath.get(doc0.documentElement, in_root, limit=end)
171 886 aaronmk
            if rows == []: raise SystemExit('Map error: Root "'+in_root
172 883 aaronmk
                +'" not found in input')
173 890 aaronmk
            map_rows(get_value, rows)
174 56 aaronmk
        else: # input is CSV
175 133 aaronmk
            map_ = dict(mappings)
176 59 aaronmk
            reader = csv.reader(sys.stdin)
177 84 aaronmk
            cols = reader.next()
178 162 aaronmk
            col_idxs = dict([(value, idx) for idx, value in enumerate(cols)])
179 164 aaronmk
            for i, mapping in enumerate(mappings):
180
                in_, out = mapping
181
                if metadata_value(in_) == None:
182
                    try: mappings[i] = (col_idxs[in_], out)
183
                    except KeyError: pass
184 162 aaronmk
185 297 aaronmk
            def get_value(in_, row):
186
                value = row[in_]
187
                if value != '': return value
188
                else: return None
189 838 aaronmk
            map_rows(get_value, reader)
190 53 aaronmk
191 838 aaronmk
    def process_inputs(root, row_ready):
192
        for map_path in map_paths: process_input(root, row_ready, map_path)
193 512 aaronmk
194 53 aaronmk
    # Output XML tree
195 512 aaronmk
    doc = xml_dom.create_doc()
196 316 aaronmk
    root = doc.documentElement
197 130 aaronmk
    if out_is_db:
198 53 aaronmk
        import db_xml
199
200 646 aaronmk
        out_db = connect_db(out_db_config)
201 310 aaronmk
        out_pkeys = {}
202 53 aaronmk
        try:
203 400 aaronmk
            if test: sql.empty_db(out_db)
204 53 aaronmk
            row_ct_ref = [0]
205 449 aaronmk
206 838 aaronmk
            def row_ready(row_num, input_row):
207 452 aaronmk
                def on_error(e):
208 835 aaronmk
                    exc.add_msg(e, term.emph('row #:')+' '+str(row_num))
209 828 aaronmk
                    exc.add_msg(e, term.emph('input row:')+'\n'+str(input_row))
210
                    exc.add_msg(e, term.emph('output row:')+'\n'+str(root))
211 452 aaronmk
                    ex_tracker.track(e)
212
213 449 aaronmk
                xml_func.process(root, on_error)
214 442 aaronmk
                if not xml_dom.is_empty(root):
215
                    assert xml_dom.has_one_child(root)
216
                    try:
217
                        sql.with_savepoint(out_db, lambda: db_xml.put(out_db,
218 462 aaronmk
                            root.firstChild, out_pkeys, row_ct_ref, on_error))
219 442 aaronmk
                        if commit: out_db.commit()
220 449 aaronmk
                    except sql.DatabaseErrors, e: on_error(e)
221 316 aaronmk
                root.clear()
222 449 aaronmk
223 838 aaronmk
            process_inputs(root, row_ready)
224 717 aaronmk
            sys.stdout.write('Inserted '+str(row_ct_ref[0])+
225 460 aaronmk
                ' new rows into database\n')
226 53 aaronmk
        finally:
227 133 aaronmk
            out_db.rollback()
228
            out_db.close()
229 751 aaronmk
    else:
230 759 aaronmk
        def on_error(e): ex_tracker.track(e)
231 838 aaronmk
        def row_ready(row_num, input_row): pass
232
        process_inputs(root, row_ready)
233 759 aaronmk
        xml_func.process(root, on_error)
234 751 aaronmk
        if out_is_xml_ref[0]:
235
            doc.writexml(sys.stdout, **xml_dom.prettyxml_config)
236
        else: # output is CSV
237
            raise NotImplementedError('CSV output not supported yet')
238 53 aaronmk
239 847 aaronmk
def main():
240
    try: main_()
241
    except Parser.SyntaxException, e: raise SystemExit(str(e))
242
243 846 aaronmk
if __name__ == '__main__':
244 847 aaronmk
    profile_to = opts.get_env_var('profile_to', None)
245
    if profile_to != None:
246
        import cProfile
247
        cProfile.run(main.func_code, profile_to)
248
    else: main()