~jfb-tempo-consulting/unifield-wm/sync-env-py3

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
from os import path
from time import sleep
import re

import csv

from xmlrpclib import Fault
openerplib = __import__('openerplib')

import config

try:
    import ipdb as pdb
except:
    import pdb

class db(object):
    def __init__(self, server, db_name, user=None, password=None, server_password=None):
        self.server = server
        try:
            self.server_password = config.db_password
        except:
            self.server_password = None
        finally:
            if server_password: self.server_password = server_password
        self.db_name = db_name
        self.service = server.get_service('db')
        if user: self.connect(user, password)

    def exec_workflow(self, model, method, ids):
        return self.server.get_service('object').exec_workflow(
            self.server.database, self.server.user_id, self.server.password,
            model, method, ids)

    def connect(self, login, password):
        self.server = openerplib.get_connection(hostname=self.server.connector.hostname, port=self.server.connector.port, database=self.db_name, login=login, password=password)
        return self

    def create_db(self, password, demo=False, lang='en_US', wait=False):
        if not self.server_password: raise Exception, "The server password is needed for this operation"
        self.id = self.service.create(self.server_password, self.db_name, demo, 'en_US', password)
        if wait: self.wait()
        return self

    create = create_db

    def wait(self):
        if not self.server_password: raise Exception, "The server password is needed for this operation"
        while True:
            try:
                running = self.service.get_progress(self.server_password, self.id)
                if not running: break
                sleep(1)
            except:
                break
        return self

    def drop(self):
        if not self.server_password: raise Exception, "The server password is needed for this operation"
        ## TODO does'n handle database current access prevent dropping
        #self.service.drop(self.server_password, self.db_name)
        try:
            self.service.drop(self.server_password, self.db_name)
        except Fault, e:
            if not re.search(r"database ([\"']).+\1 does not exist", e.faultCode): raise e
        finally:
            return self

    def get(self, model):
        return self.server.get_model(model)

    def wizard(self, model, data):
        return wizard(self, model, data)

    def ref(self, xml_id, index=0):
        return ref(xml_id).get(self)
    #def ref(self, model, xml_id, index=0):
    #    return ref(model, xml_id).get(self)

    def module(self, name, index=0):
        return module(self, name, index)

    def action_done(self, action_id):
        proxy = self.get('ir.actions.todo')
        ids = proxy.search([('action_id', '=', action_id)])
        if ids: proxy.write(ids, {'state' : 'done'})

    def user(self, login):
        return user(self, login)

    def group(self, name):
        return group(self, name)

    def search_data(self, model, data, domain=None, action=None, active='active', pdb=False):
        if isinstance(data, dict):
            searchDomain = [(field,'=',value) for (field,value) in data.items() if value]
        elif isinstance(data, (list,tuple)):
            searchDomain = list(data)
        else:
            raise NotImplementedError("Cannot transform %s (of type %s) to a domain (list)" % (data, type(data).__name__))
        if domain:
            searchDomain += list(domain)
        if active == 'unactive': searchDomain.append(('active','=',0))
        elif active == 'both': searchDomain.extend(['|',('active','=',0),('active','=',1)])
        elif not active == 'active': raise Exception, "'active' prameter must be one of these: active, unactive or both"
        proxy = self.get(model)
        ids = proxy.search(searchDomain)
        if not ids and pdb:
            pdb.set_trace()
        if action == 'unlink': return proxy.unlink(ids) if ids else True
        elif action == 'test': return bool(ids)
        else: return ids
        
   
    test = lambda self, model, data, **o: self.search_data(model, data, action='test', **o)
    clean = lambda self, model, data, **o: self.search_data(model, data, action='unlink', **o)
 
    def write(self, model, ids_or_domain, data):
        if all([isinstance(x, (tuple, list)) for x in ids_or_domain]):
            ids = self.get(model).search(ids_or_domain)
        elif isinstance(ids_or_domain, dict):
            ids = self.search_data(model, ids_or_domain)
        else:
            ids = ids_or_domain
        if ids and not self.get(model).write(ids, data):
            return 0
        return len(ids)

    def activate(self, model, domain, state=1):
        proxy = self.get(model)
        search_domain = [('active','=',1-state)] + domain
        ids = proxy.search(search_domain)
        proxy.write(ids, {'active':state})
        return len(ids)

    def desactivate(self, model, domain):
        return self.activate(model, domain, 0)

    def load_datas(self, filepath):
        datas = []
        if path.isfile(filepath):
            if filepath[-4:] == '.csv':
                ch = csv.reader(open(filepath, 'rb'), quotechar='"', delimiter=',')
                fields = ch.next()
                datetime_re = re.compile(r"^\d\d\d\d-\d\d-\d\d( \d\d:\d\d:\d\d)?$")
                for values in ch:
                    evaluated_values = []
                    for v in values:
                        try:
                            if datetime_re.match(v): raise Exception
                            evaluated_values += [eval(v)]
                        except:
                            evaluated_values += [eval('"""'+v+'"""')]
                    datas.append(dict(zip(fields, evaluated_values)))
            elif filepath[-4:] == '.xml':
                raise Exception, 'While importing XML file: Not yet implemented'
            else:
                raise Exception, 'Cannot import file "%s": unrecognized file type!' % (filepath,)
        else:
            raise IOError, 'Can\'t access to file "%s", check read permission!' % (filepath,)
        return datas

    def Import(self, model, datas, mode='init'):
        if type(datas) == dict:
            fields = datas.keys()
            datas = [datas.values()]
        elif not type(datas) == list:
            raise Exception, 'Cannot import data: datas must be a list of dict!'
        else:
            fields = []
            for i in datas:
                fields = set(list(fields) + i.keys())
            fields = list(fields)
            old_datas = datas
            datas = []
            for i in old_datas:
                datas.append([i.get(field, None) for field in fields])
        for data in datas:
            for i, v in enumerate(data):
                data[i] = str(data[i])
                #if v == None: data[i] = False
                #elif type(v) == bool: data[i] = 1 if v else 0
        if 'reconcile note' in fields: ipdb.set_trace()
        result, rows, warning_msg, dummy = self.get(model).import_data(fields, datas, mode)
        if result == -1:
            raise Exception, "Unable to import data: "+str(warning_msg)
        return self

class ref(object):
    def __init__(self, xml_id, index=0):
        self.xml_id = xml_id
        self.index = 0

    def get(self, db, refetch=False):
        try:
            if refetch: raise AttributeError
            return self.value
        except AttributeError:
            module, xml_id = self.xml_id.split('.')
            model, i = db.get('ir.model.data').get_object_reference(module, xml_id)
            self.value = db.get(model).read([i], ['id'])[0]
            return self.value['id']

class wizard(object):
    def __init__(self, db, model, data):
        for k, v in data.items():
            if isinstance(v, ref):
                data[k] = v.get(db)
        self.proxy = db.get(model)
        self.id = self.proxy.create(data)

    def __getattr__(self, attr):
        real_attr = getattr(self.proxy, attr)
        def foo(*args):
            return real_attr([self.id], *args)
        return foo

    def data(self, *attrs):
        data = self.proxy.read([self.id], attrs)
        if len(attrs) == 1: return data[0][attrs[0]]
        else: return data[0]

class module(object):
    def __init__(self, db, name, index=0):
        self.db = db
        self.module_proxy = db.get('ir.module.module')
        if name == 'all':
            name = 'base'
            #self.ids = self.module_proxy.search([('state','=','installed')])
            #if not self.ids:
            #    raise Exception, "No module installed!"
        #else:
        self.ids = self.module_proxy.search([('name','=',name)])
        if not self.ids:
            raise Exception, "Unable to find module %s!" % (name,)

    def remove(self):
        self.expect = 'uninstalled'
        self.module_proxy.button_uninstall(self.ids)
        return self

    uninstall = remove

    def install(self):
        self.expect = 'installed'
        self.module_proxy.button_install(self.ids)
        return self

    def upgrade(self):
        self.expect = 'installed'
        self.module_proxy.button_upgrade(self.ids)
        return self

    def do(self):
        self.db.get('base.module.upgrade').upgrade_module([])
        if self.module_proxy.search([('id','in',self.ids),('state','=',self.expect)], 0, False, False, True) == len(self.ids):
            raise Exception, "Modules modifications not applied"
        return self

class user(object):
    def __init__(self, db, login):
        self.db = db
        self.users = db.get('res.users')
        self.login = login
        try: self.id = self.users.search([('login','=',login)])[0]
        except: self.id = None

    def delGroups(self, *groups):
        if not self.id: raise Exception, 'Cannot remove groups to unknown user: '+self.login
        group_ids = [i.id if type(i) == group else group(self.db, i).id for i in groups]
        if None in group_ids: raise Exception, 'Cannot find some groups in this list: '+', '.join(groups)
        group_changes = [(3,i,) for i in group_ids]
        self.users.write([self.id],{'groups_id':group_changes})
        return self

    def addGroups(self, *groups):
        if not self.id: raise Exception, 'Cannot add groups to unknown user: '+self.login
        group_ids = [i.id if type(i) == group else group(self.db, i).id for i in groups]
        if None in group_ids: raise Exception, 'Cannot find some groups in this list: '+', '.join(groups)
        group_changes = [(4,i,) for i in group_ids]
        self.users.write([self.id],{'groups_id':group_changes})
        return self

    def add(self, password, name=None):
        if not self.id:
            self.id = self.users.create({'login':self.login,'name':(name or self.login),'password':password,})
        return self

    def exists(self):
        return bool(self.id)

class group(object):
    def __init__(self, db, name):
        self.db = db
        self.groups = db.get('res.groups')
        self.name = name
        try: self.id = self.groups.search([('name','=',name)])[0]
        except: self.id = None

    def delUsers(self, *users):
        if not self.id: self.add()
        user_ids = [i.id if type(i) == user else user(self.db, i).id for i in users]
        if None in user_ids: raise Exception, 'Cannot find some users in this list: '+', '.join(users)
        user_changes = [(3,i,) for i in user_ids]
        self.groups.write([self.id],{'users':user_changes})
        return self

    def addUsers(self, *users):
        if not self.id: self.add()
        user_ids = [i.id if type(i) == user else user(self.db, i).id for i in users]
        if None in user_ids: raise Exception, 'Cannot find some users in this list: '+', '.join(users)
        user_changes = [(4,i,) for i in user_ids]
        self.groups.write([self.id],{'users':user_changes})
        return self

    def add(self):
        if not self.id:
            self.id = self.groups.create({'name':self.name,})
        return self

    def exists(self):
        return bool(self.id)