root / zinnplugins / tryton / party.py @ a899390a2a8aca3b3b4617dd7cd96a19e64b489c

View | Annotate | Download (16 KB)

1
# -*- coding: utf-8 -*-
2
# Copyright (c) 2009 Lost Oasis, IELO <info@ielo.net>
3
#
4
# This file is part of ZINN.
5
# 
6
# SINN is free software: you can redistribute it and/or modify it under
7
# the terms of the GNU General Public License as published by the Free
8
# Software Foundation, either version 3 of the License, or (at your
9
# option) any later version.
10
# 
11
# SINN is distributed in the hope that it will be useful, but WITHOUT
12
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13
# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
14
# for more details.
15
# 
16
# You should have received a copy of the GNU General Public License
17
# along with ZINN.  If not, see <http://www.gnu.org/licenses/>.
18
19
import xmlrpclib
20
21
from erp import Itransport
22
from erp import Iparty
23
from erp import IpartyAddress
24
from erp import IpartyContactMechanism
25
26
from urlparse import urlparse
27
28
class TrytonTransport(Itransport.Itransport):
29
    def __init__(self, dsn):
30
        super(TrytonTransport, self).__init__(dsn)
31
        port = ''
32
        o = urlparse(dsn)
33
        if o.port: port = ':'+str(o.port)
34
        self.uri = o.scheme+'://'+o.hostname+port+o.path
35
        self.user     = o.username
36
        self.password = o.password
37
38
        self.proxy = xmlrpclib.ServerProxy(self.uri)
39
        self.user_id, self.session = self.proxy.common.db.login(self.user,
40
                self.password)
41
        self.context = self.proxy.model.res.user.get_preferences(self.user_id,
42
                self.session, True, {})
43
44
    def getProxy(self):
45
        return self.proxy
46
47
    def getContext(self):
48
        return self.context
49
50
    def getSession(self):
51
        return self.session
52
53
    def getUserId(self):
54
        return self.user_id
55
56
57
class CountrySubdivision():
58
    def __init__(self, transport):
59
        self.transport = transport
60
        self.proxy     = self.transport.getProxy().model.country.subdivision
61
62
    def list(self, countryName, subdivisionName=None):
63
        subdivision = self.byName(countryName, subdivisionName)
64
        if not subdivision:
65
            return subdivision
66
        else:
67
            return [s['name'] for s in subdivision]
68
69
    def byId(self, subdivisionId):
70
        if not subdivisionId:
71
            return False
72
        where = [('id', '=', subdivisionId)]
73
        args = [self.transport.getUserId(), self.transport.getSession(),
74
                where,                          # where
75
                0,                              # offset
76
                1,                              # limit
77
                [],                             # order
78
                self.transport.getContext(),    # context
79
                ['name']                        # fields
80
                ]
81
        subdivision = self.proxy.search_read(*args)
82
        return subdivision['name'] if len(subdivision) else False
83
84
85
    def byName(self, countryName=None, subdivisionName=None):
86
        c = PartyCountry(self.transport)
87
        country = c.byName(countryName) if countryName else None
88
        where = []
89
        if country:
90
            where.append(('country', '=', country[0]['id']))
91
        if subdivisionName:
92
            where.append(('name', '=', subdivisionName))
93
        args = [self.transport.getUserId(), self.transport.getSession(),
94
                where,                          # where
95
                0,                              # offset
96
                10000000,                       # limit
97
                [],                             # order
98
                self.transport.getContext(),    # context
99
                ['id', 'name']                  # fields
100
                ]
101
        subdivision = self.proxy.search_read(*args)
102
        return subdivision if len(subdivision) else None
103
104
105
class PartyCountry():
106
    def __init__(self, transport):
107
        self.transport   = transport
108
        self.subdivision = CountrySubdivision(self.transport)
109
        self.proxy       = self.transport.getProxy().model.country.country
110
111
    def list(self, countryName=None):
112
        country = self.byName(countryName)
113
        if not country:
114
            return country
115
        else:
116
            return [c['name'] for c in country]
117
118
    def byId(self, countryId):
119
        if not countryId:
120
            return False
121
        where = [('id', '=', countryId)]
122
        args = [self.transport.getUserId(), self.transport.getSession(),
123
                where,                          # where
124
                0,                              # offset
125
                1,                              # limit
126
                [],                             # order
127
                self.transport.getContext(),    # context
128
                ['id', 'name']                  # fields
129
                ]
130
        country = self.proxy.search_read(*args)
131
        return country if len(country) else False
132
133
    def byName(self, countryName=None):
134
        where = [('name', '=', countryName)] if countryName else []
135
        args = [self.transport.getUserId(), self.transport.getSession(),
136
                where,                          # where
137
                0,                              # offset
138
                1000000,                        # limit
139
                [],                             # order
140
                self.transport.getContext(),    # context
141
                ['id', 'name']                  # fields
142
                ]
143
        country = self.proxy.search_read(*args)
144
        return country if len(country) else None
145
146
class PartyAddress(IpartyAddress.IpartyAddress):
147
    def __init__(self, transport):
148
        self.transport   = transport
149
        self.country     = PartyCountry(self.transport)
150
        self.subdivision = CountrySubdivision(self.transport)
151
        self.proxy       = self.transport.getProxy().model.party.address
152
153
    def list(self, partyLabel, label=None):
154
        fields = [] if label else ['name']
155
        address = self.byLabel(partyLabel, label, fields)
156
        if not address:
157
            return address
158
        if not label:
159
            return [p['name'] for p in address]
160
        else:
161
            return self._decorate(address)
162
163
    def create(self, partyLabel, label, attributes):
164
        address = self.byLabel(partyLabel, label)
165
        if address:
166
            return False
167
        else:
168
            p = Party(self.transport)
169
            party = p.byLabel(partyLabel)
170
            if not party:
171
                return False
172
            else:
173
                default = {
174
                        'name'  : label,
175
                        'party' : party[0]['id'],
176
                        }
177
                attr = self.AttributesFilter(attributes, default)
178
                attr = self.transformAttributesInRelationsId(attr)
179
                attr['name'] = str(attr['name']).strip(' \t\r\n')
180
                args = [self.transport.getUserId(), self.transport.getSession(),
181
                        attr, self.transport.getContext()]
182
                return self.proxy.create(*args)
183
184
    def update(self, partyLabel, label, attributes):
185
        address = self.byLabel(partyLabel, label)
186
        if not address:
187
            return False
188
        else:
189
            if not address[0]['party']:
190
                return False
191
            else:
192
                attr = self.AttributesFilter(attributes)
193
                attr = self.transformAttributesInRelationsId(attr)
194
                args = [self.transport.getUserId(), self.transport.getSession(),
195
                        address[0]['id'], attr, self.transport.getContext()]
196
                return self.proxy.write(*args)
197
198
    def delete(self, partyLabel, label):
199
        address = self.byLabel(partyLabel, label)
200
        if not address:
201
            return False # already deleted
202
        else:
203
            args = [self.transport.getUserId(), self.transport.getSession(),
204
                    address[0]['id'], self.transport.getContext()]
205
            return self.proxy.delete(*args)
206
207
    def attributes(self, onlyMutables=False):
208
        return self.getAttributes(onlyMutables)
209
210
    def byLabel(self, partyLabel, label=None, fields=[]):
211
        p = Party(self.transport)
212
        party = p.byLabel(partyLabel)
213
        if not party:
214
            return None
215
        where = ['AND', ('party', '=', party[0]['id'])]
216
        if not label:
217
            whereAnd = []
218
        else:
219
            if isinstance(label, list):
220
                whereAnd = ['OR']
221
                for l in label:
222
                    whereAnd.append(('name', '=', l))
223
            else:
224
                whereAnd = [('name', '=', label)]
225
        where.append(whereAnd)
226
227
        args = [self.transport.getUserId(), self.transport.getSession(),
228
                where,                          # where
229
                0,                              # offset
230
                10000,                          # limit # FIXME
231
                [],                             # order
232
                self.transport.getContext(),    # context
233
                []                              # fields
234
                ]
235
        address = self.proxy.search_read(*args)
236
        return None if len(address) == 0 else address
237
238
    def transformAttributesInRelationsId(self, attributes):
239
        attr = attributes
240
        country_label = None
241
        if attr.has_key('country') and attr['country']:
242
            country = self.country.byName(attr['country'])
243
            if country:
244
                country_label = attr['country']
245
                attr['country'] = country[0]['id']
246
            else:
247
                del attr['country']
248
249
        if attr.has_key('subdivision') and attr['subdivision']:
250
            subdivision = self.subdivision.byName(country_label,
251
                    attr['subdivision'])
252
            if subdivision:
253
                attr['subdivision'] = subdivision[0]['id']
254
            else:
255
                del attr['subdivision']
256
257
        return attr
258
259
260
class PartyContactMechanism(IpartyContactMechanism.IpartyContactMechanism):
261
    def __init__(self, transport):
262
        self.transport = transport
263
        self.proxy     = self.transport.getProxy().model.party.contact_mechanism
264
265
    def list(self, partyLabel, contactType=None):
266
        fields = ['value', 'type']
267
        contact_mechanism = self.byType(partyLabel, contactType, None, fields)
268
        if not contact_mechanism:
269
            return contact_mechanism
270
        else:
271
            return [{'type': cm['type'], 'value': cm['value']} for cm in
272
                    contact_mechanism]
273
274
    def byType(self, partyLabel, contactType=None, contactValue=None, fields=[]):
275
        p = Party(self.transport)
276
        party = p.byLabel(partyLabel)
277
        if not party:
278
            return None
279
        else:
280
            where = [('party', '=', party[0]['id'])]
281
            if contactType in self.types:
282
                where.append(('type', '=', contactType))
283
            if contactValue:
284
                where.append(('value', '=', contactValue))
285
            args = [self.transport.getUserId(), self.transport.getSession(),
286
                    where,                          # where
287
                    0,                              # offset
288
                    1000,                           # limit #FIXME
289
                    [('type', 'ASC')],              # order
290
                    self.transport.getContext(),    # context
291
                    fields                          # fields
292
                    ]
293
            return self.proxy.search_read(*args)
294
295
    def create(self, partyLabel, attributes):
296
        if not isinstance(attributes, dict):
297
            return False
298
        if not (attributes.has_key('type') and attributes['type'] in self.types):
299
            return False
300
        if not attributes.has_key('value'):
301
            return False
302
        attributes['value'] = str(attributes['value']).strip(' \n\t\r')
303
        contact_mechanism = self.byType(partyLabel, attributes['type'], attributes['value'])
304
        if contact_mechanism:
305
            return False
306
        p = Party(self.transport)
307
        party = p.byLabel(partyLabel)
308
        if not party:
309
            return False
310
        else:
311
            default = {'party' : party[0]['id']}
312
            attr = self.AttributesFilter(attributes, default)
313
            args = [self.transport.getUserId(), self.transport.getSession(),
314
                            attr, self.transport.getContext()]
315
            return self.proxy.create(*args)
316
317
    def delete(self, partyLabel, attributes):
318
        if not isinstance(attributes, dict):
319
            return False
320
        if not (attributes.has_key('type') and attributes['type'] in self.types):
321
            return False
322
        if not attributes.has_key('value'):
323
            return False
324
        attributes['value'] = str(attributes['value']).strip(' \n\t\r')
325
        contact_mechanism = self.byType(partyLabel, attributes['type'], attributes['value'])
326
        if not contact_mechanism:
327
            return False
328
        else:
329
            args = [self.transport.getUserId(), self.transport.getSession(),
330
                            contact_mechanism[0]['id'],
331
                            self.transport.getContext()]
332
            return self.proxy.delete(*args)
333
334
    def attributes(self, onlyMutables=False):
335
        return self.getAttributes(onlyMutables)
336
337
338
class Party(Iparty.Iparty):
339
    def __init__(self, transport):
340
        super(Party, self).__init__(transport)
341
        self.transport = transport
342
        self.proxy   = self.transport.getProxy().model.party.party
343
        self.country = PartyCountry(self.transport)
344
        self.address = PartyAddress(self.transport)
345
        self.contact_mechanism = PartyContactMechanism(self.transport)
346
347
    def list(self, label=None):
348
        fields = [] if label else ['name']
349
        party = self.byLabel(label, fields)
350
        if not party:
351
            return party
352
        if not label:
353
            return [p['name'] for p in party]
354
        else:
355
            return self._decorate(party)
356
357
    def create(self, label, attributes):
358
        party = self.byLabel(label)
359
        if party:
360
            return False
361
        else:
362
            default = {
363
                    'active'       : True,
364
                    'name'         : label,
365
                    'full_name'    : label,
366
                    'vat_country'  : 'FR'
367
                    }
368
            attr = self.AttributesFilter(attributes, default)
369
            attr['name'] = str(attr['name']).strip(' \t\r\n')
370
            attr['full_name'] = str(attr['full_name']).strip(' \t\r\n')
371
            args = [self.transport.getUserId(), self.transport.getSession(),
372
                    attr, self.transport.getContext()]
373
            ret = self.proxy.create(*args)
374
            return ret
375
376
    def update(self, label, attributes):
377
        party = self.byLabel(label)
378
        if party is None:
379
            return False
380
        else:
381
            attr = self.AttributesFilter(attributes)
382
            args = [self.transport.getUserId(), self.transport.getSession(),
383
                    party[0]['id'], attr, self.transport.getContext()]
384
            return self.proxy.write(*args)
385
386
    def delete(self, label):
387
        party = self.byLabel(label)
388
        if party:
389
            args = [self.transport.getUserId(), self.transport.getSession(),
390
                    party[0]['id'], self.transport.getContext()]
391
            return self.proxy.delete(*args)
392
        else:
393
            return False # already deleted
394
395
    def attributes(self, onlyMutables=False):
396
        return self.getAttributes(onlyMutables)
397
398
    def byLabel(self, label=None, fields=[]):
399
        if not label:
400
            where = []
401
        else:
402
            if isinstance(label, list):
403
                where = ['OR']
404
                for l in label:
405
                    where.append(('name', '=', l))
406
            else:
407
                where = [('name', '=', label)]
408
        ctx = self.transport.getContext()
409
        ctx['active_test'] = False
410
        args = [self.transport.getUserId(), self.transport.getSession(),
411
                where,                          # where
412
                0,                              # offset
413
                10000,                          # limit FIXME
414
                [],                             # order
415
                ctx,                            # context
416
                fields                          # fields
417
                ]
418
        party = self.proxy.search_read(*args)
419
        return None if len(party) == 0 else party