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 310 aaronmk
import util
17 133 aaronmk
import xml_dom
18 86 aaronmk
import xml_func
19 53 aaronmk
20 84 aaronmk
def metadata_value(name):
21 164 aaronmk
    if type(name) == str and name.startswith(':'): return name[1:]
22 84 aaronmk
    else: return None
23
24 53 aaronmk
def main():
25 434 aaronmk
    ex_tracker = exc.ExTracker()
26
27 131 aaronmk
    env_names = []
28
    def usage_err():
29 400 aaronmk
        raise SystemExit('Usage: '+opts.env_usage(env_names, True)+' [commit=1]'
30 510 aaronmk
            ' [test=1] '+sys.argv[0]+' [map_path...] [<input] [>output]')
31 146 aaronmk
    limit = opts.get_env_var('n', None, env_names)
32
    if limit != None: limit = int(limit)
33 400 aaronmk
    test = opts.env_flag('test')
34
    commit = not test and opts.env_flag('commit') # never commit in test mode
35 131 aaronmk
36 53 aaronmk
    # Get db config from env vars
37 131 aaronmk
    db_config_names = ['engine', 'host', 'user', 'password', 'database']
38 53 aaronmk
    def get_db_config(prefix):
39 64 aaronmk
        return opts.get_env_vars(db_config_names, prefix, env_names)
40 67 aaronmk
    in_db_config = get_db_config('in')
41
    out_db_config = get_db_config('out')
42 131 aaronmk
    in_is_db = 'engine' in in_db_config
43
    out_is_db = 'engine' in out_db_config
44 53 aaronmk
45
    # Parse args
46 510 aaronmk
    map_paths = sys.argv[1:]
47 512 aaronmk
    if map_paths == []:
48
        if in_is_db or not out_is_db: usage_err()
49
        else: map_paths = [None]
50 53 aaronmk
51 646 aaronmk
    def connect_db(db_config):
52
        db = sql.connect(db_config)
53
        sys.stderr.write('Connected to '+sql.db_config_str(db_config)+'\n')
54
        return db
55
56 512 aaronmk
    def process_input(root, process_row, map_path):
57
        '''Inputs datasource to XML tree, mapping if needed'''
58
        # Load map header
59
        in_is_xpaths = True
60
        out_label = None
61
        if map_path != None:
62
            import copy
63
            import csv
64
65
            import xpath
66
67
            metadata = []
68
            mappings = []
69
            stream = open(map_path, 'rb')
70
            reader = csv.reader(stream)
71
            in_label, out_label = reader.next()[:2]
72
            def split_col_name(name):
73
                name, sep, root = name.partition(':')
74
                return name, sep != '', root
75
            in_label, in_is_xpaths, in_root = split_col_name(in_label)
76
            out_label, out_is_xpaths, out_root = split_col_name(out_label)
77
            assert out_is_xpaths # CSV output not supported yet
78
            has_types = out_root.startswith('/*s/') # outer elements are types
79
            for row in reader:
80
                in_, out = row[:2]
81
                if out != '':
82
                    if out_is_xpaths: out = xpath.parse(out_root+out)
83
                    mappings.append((in_, out))
84
            stream.close()
85
86
            root.ownerDocument.documentElement.tagName = out_label
87
        in_is_xml = in_is_xpaths and not in_is_db
88 56 aaronmk
89 512 aaronmk
        if in_is_xml:
90
            doc0 = minidom.parse(sys.stdin)
91
            if out_label == None: out_label = doc0.documentElement.tagName
92 53 aaronmk
93 309 aaronmk
        def process_rows(get_value, rows):
94 297 aaronmk
            '''Processes input values
95
            @param get_value f(in_, row):str
96
            '''
97 314 aaronmk
            for i, row in enumerate(rows):
98 316 aaronmk
                if not (limit == None or i < limit): break
99
                row_id = str(i)
100
                for in_, out in mappings:
101
                    value = metadata_value(in_)
102
                    if value == None: value = get_value(in_, row)
103
                    if value != None:
104
                        xpath.put_obj(root, out, row_id, has_types, value)
105 452 aaronmk
                process_row(row)
106 460 aaronmk
            sys.stderr.write('Processed '+str(i+1)+' input rows\n')
107 297 aaronmk
108 310 aaronmk
        if map_path == None:
109
            iter_ = xml_dom.NodeElemIter(doc0.documentElement)
110
            util.skip(iter_, xml_dom.is_text) # skip metadata
111 317 aaronmk
            for child in iter_:
112
                root.appendChild(child)
113 452 aaronmk
                process_row(child)
114 309 aaronmk
        elif in_is_db:
115 130 aaronmk
            assert in_is_xpaths
116 126 aaronmk
117 117 aaronmk
            import db_xml
118
119 161 aaronmk
            in_root_xml = xpath.path2xml(in_root)
120 164 aaronmk
            for i, mapping in enumerate(mappings):
121
                in_, out = mapping
122
                if metadata_value(in_) == None:
123 168 aaronmk
                    mappings[i] = (xpath.path2xml(in_root+'/'+in_), out)
124 126 aaronmk
125 646 aaronmk
            in_db = connect_db(in_db_config)
126 133 aaronmk
            in_pkeys = {}
127 297 aaronmk
            def get_value(in_, row):
128 167 aaronmk
                pkey, = row
129 297 aaronmk
                in_ = in_.cloneNode(True) # don't modify orig value!
130
                xml_dom.set_id(xpath.get(in_, in_root), pkey)
131
                value = sql.value_or_none(db_xml.get(in_db, in_, in_pkeys))
132
                if value != None: return str(value)
133
                else: return None
134 309 aaronmk
            process_rows(get_value, sql.rows(db_xml.get(in_db, in_root_xml,
135
                in_pkeys, limit)))
136 117 aaronmk
            in_db.close()
137 161 aaronmk
        elif in_is_xml:
138 297 aaronmk
            def get_value(in_, row):
139
                node = xpath.get(row, in_)
140
                if node != None: return xml_dom.value(node)
141
                else: return None
142
            row0 = xpath.get(doc0.documentElement, in_root)
143 309 aaronmk
            process_rows(get_value, xml_dom.NodeElemIter(row0.parentNode))
144 56 aaronmk
        else: # input is CSV
145 133 aaronmk
            map_ = dict(mappings)
146 59 aaronmk
            reader = csv.reader(sys.stdin)
147 84 aaronmk
            cols = reader.next()
148 162 aaronmk
            col_idxs = dict([(value, idx) for idx, value in enumerate(cols)])
149 164 aaronmk
            for i, mapping in enumerate(mappings):
150
                in_, out = mapping
151
                if metadata_value(in_) == None:
152
                    try: mappings[i] = (col_idxs[in_], out)
153
                    except KeyError: pass
154 162 aaronmk
155 297 aaronmk
            def get_value(in_, row):
156
                value = row[in_]
157
                if value != '': return value
158
                else: return None
159 309 aaronmk
            process_rows(get_value, reader)
160 53 aaronmk
161 512 aaronmk
    def process_inputs(root, process_row):
162
        for map_path in map_paths: process_input(root, process_row, map_path)
163
164 53 aaronmk
    # Output XML tree
165 512 aaronmk
    doc = xml_dom.create_doc()
166 316 aaronmk
    root = doc.documentElement
167 130 aaronmk
    if out_is_db:
168 53 aaronmk
        from psycopg2.extensions import ISOLATION_LEVEL_SERIALIZABLE
169
        import db_xml
170
171 646 aaronmk
        out_db = connect_db(out_db_config)
172 310 aaronmk
        out_pkeys = {}
173 53 aaronmk
        try:
174 400 aaronmk
            if test: sql.empty_db(out_db)
175 53 aaronmk
            row_ct_ref = [0]
176 449 aaronmk
177 452 aaronmk
            def process_row(input_row):
178
                def on_error(e):
179
                    exc.add_msg(e, 'output row:\n'+str(root))
180
                    exc.add_msg(e, 'input row:\n'+str(input_row))
181
                    ex_tracker.track(e)
182
183 449 aaronmk
                xml_func.process(root, on_error)
184 442 aaronmk
                if not xml_dom.is_empty(root):
185
                    assert xml_dom.has_one_child(root)
186
                    try:
187
                        sql.with_savepoint(out_db, lambda: db_xml.put(out_db,
188 462 aaronmk
                            root.firstChild, out_pkeys, row_ct_ref, on_error))
189 442 aaronmk
                        if commit: out_db.commit()
190 449 aaronmk
                    except sql.DatabaseErrors, e: on_error(e)
191 316 aaronmk
                root.clear()
192 449 aaronmk
193 512 aaronmk
            process_inputs(root, process_row)
194 460 aaronmk
            sys.stderr.write('Inserted '+str(row_ct_ref[0])+
195
                ' new rows into database\n')
196 53 aaronmk
        finally:
197 133 aaronmk
            out_db.rollback()
198
            out_db.close()
199 299 aaronmk
    else: # output is XML
200 452 aaronmk
        def process_row(input_row): pass
201 512 aaronmk
        process_inputs(root, process_row)
202 316 aaronmk
        xml_func.process(root)
203
        doc.writexml(sys.stdout, **xml_dom.prettyxml_config)
204 53 aaronmk
205 133 aaronmk
try: main()
206 294 aaronmk
except Parser.SyntaxException, e: raise SystemExit(str(e))