# -*- coding: utf-8 -*- # Copyright 2018 Fabien Bourgeois # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as # published by the Free Software Foundation, either version 3 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . """ Odoo Radicale Storage Plugin """ # # PLAN # 1. Implement readonly from Odoo only # 2. Implement contacts first (and so vcf) # 3. Implement unique events (.ics) # 4. Implement recurrent events # 5. Begin write (two way) for contacts # 5. Begin write (two way) for calendar from contextlib import contextmanager from time import strftime, strptime import vobject from odoorpc import ODOO from odoorpc.error import RPCError from radicale import xmlutils from radicale.storage import BaseCollection, Item, get_etag, get_uid_from_object from radicale_odoo_auth import Auth class Collection(BaseCollection): """ BaseCollection implementation for Odoo Radicale Storage """ odoo = False def __init__(self, path): """ Init function """ self.__class__.odoo = Auth.odoo if not self.__class__.odoo: self.logger.error('No auth Odoo found...') raise RuntimeError('No auth Odoo found') self.__class__.odoo_connect() # self.odoo_init() attributes = path.strip('/').split('/') self.tag = None self.props = {} if 'odoo-contact' in attributes: self.tag = 'VADDRESSBOOK' self.odoo_model = 'res.partner' self.content_suffix = '.vcf' self.props.update({'tag': 'VADDRESSBOOK', 'D:displayname': 'Odoo contacts', 'CR:addressbook-description': 'Contacts form your Odoo account'}) elif 'odoo-calendar' in attributes: self.tag = 'VCALENDAR' self.odoo_model = 'calendar.event' self.content_suffix = '.ics' self.props.update({'tag': 'VCALENDAR', 'D:displayname': 'Odoo calendar', 'C:calendar-description': 'Events form your Odoo calendar'}) self.path = path.strip('/') self.owner = attributes[0] self.is_principal = len(attributes) == 0 @classmethod def odoo_connect(cls): """ Global Odoo connection : server and admin account """ host = cls.configuration.get('storage', 'odoo_host', fallback='127.0.0.1') port = cls.configuration.get('storage', 'odoo_port', fallback=8069) admin = cls.configuration.get('storage', 'odoo_admin_username') password = cls.configuration.get('storage', 'odoo_admin_password') database = cls.configuration.get('storage', 'odoo_database') try: cls.odoo = ODOO(host, port=port) except RPCError as rpcerr: cls.logger.error(rpcerr) raise RuntimeError(rpcerr) try: cls.odoo.login(database, admin, password) cls.logger.info('Login successfull for {} on database {}'.format(admin, database)) except RPCError as rpcerr: cls.logger.error('Login problem for {} on database {}'.format(cls, database)) cls.logger.error(rpcerr) raise RuntimeError(rpcerr) return True def odoo_init(self): """ Init Odoo collections if not found """ user_ids = self.odoo.env['res.users'].search([]) users = self.odoo.execute('res.users', 'read', user_ids, ['login', 'email']) for user in users: principal_path = user.get('login') self.logger.debug('Check collections from Odoo for %s' % principal_path) contact_path = '%s/odoo-contact' % principal_path calendar_path = '%s/odoo-calendar' % principal_path collections = self.discover(principal_path, depth='1') paths = [coll.path for coll in collections] if contact_path not in paths: props = {'tag': 'VADDRESSBOOK', 'D:displayname': 'Odoo contacts', 'C:calendar-description': 'Contacts form your Odoo account'} self.create_collection(contact_path, props=props) self.logger.info('Collection creation for Odoo Sync : %s' % contact_path) if calendar_path not in paths: props = {'tag': 'VCALENDAR', 'D:displayname': 'Odoo calendar', 'C:calendar-description': 'Events form your Odoo calendar'} self.create_collection(calendar_path, props=props) self.logger.info('Collection creation for Odoo Sync : %s' % calendar_path) @classmethod @contextmanager def acquire_lock(cls, mode, user=None): cls.user = user yield @classmethod def discover(cls, path, depth="0"): """Discover a list of collections under the given ``path``. ``path`` is sanitized. If ``depth`` is "0", only the actual object under ``path`` is returned. If ``depth`` is anything but "0", it is considered as "1" and direct children are included in the result. The root collection "/" must always exist. """ attributes = path.strip('/').split('/') or [] if path and not cls.user: cls.user = attributes[0] cls.logger.warning('Discover : %s (path), %s (depth), %s (cls.user), %s (attributes)' % (path, depth, cls.user, attributes)) yield cls(path) if len(attributes) == 1: # Got all if root is needed contact_path = '%sodoo-contact' % path calendar_path = '%sodoo-calendar' % path yield cls(contact_path) yield cls(calendar_path) elif len(attributes) == 2: # Then we need children collection = cls(path) for item in collection.list(): yield collection.get(item) def get_meta(self, key=None): """Get metadata value for collection """ if key: return self.props.get(key) else: return self.props @classmethod def create_collection(cls, href, collection=None, props=None): """ Create collection implementation : only warns ATM """ cls.logger.error('Attemmpt to create a new collection for %s' % href) @classmethod def get_contacts_from_odoo(cls, login): """ Gets all contacts available from one Odoo login """ cls.logger.info('Get contacts for Odoo user %s' % login) partner_ids = cls.odoo.env['res.partner'].search([]) return ['res.partner:%s' % pid for pid in partner_ids] def sync(self, old_token=None): """ Debug purpose """ token, ilist = super(Collection, self).sync(old_token) self.logger.debug('Sync token : %s' % token) self.logger.debug('Sync list : %s' % ilist) return token, ilist def list(self): """List collection items.""" # TODO : get all ICS from Odoo... self.logger.debug('List collection %s' % self.path) self.logger.debug('Collection tag %s' % self.tag) if self.tag: if self.tag == 'VADDRESSBOOK': for oid in self.__class__.get_contacts_from_odoo(self.owner): yield oid + self.content_suffix def _get_with_metadata(self, href): """Fetch a single item from Odoo database""" model, database_id = href.split(':') database_id = int(database_id[:-len(self.content_suffix)]) # fields = ['name', 'write_date', 'comment', 'street', 'street2', 'zip', # 'city', 'state_id' 'country_id'] # data = self.odoo.execute(model, 'read', [database_id], fields)[0] data = self.odoo.env[model].browse([database_id]) if model == 'res.partner': # last_modified = strftime("%a, %d %b %Y %H:%M:%S GMT", # strptime(data.get('write_date'), '%Y-%m-%d %H:%M:%S')) last_modified = str(data.write_date) vobject_item = vobject.vCard() vobject_item.add('n') vobject_item.add('fn') vobject_item.fn.value = data.name vobject_item.add('adr') state_name = data.state_id.name if data.state_id else '' country_name = data.country_id.name if data.country_id else '' vobject_item.adr.value = vobject.vcard.Address( street=' '.join([data.street or '', data.street2 or '']), code=data.zip or '', city=data.city or '', region=state_name, country=country_name) if data.image: vobject_item.add('photo;encoding=b;type=jpeg').value = data.image.replace('\n', '') if data.phone: vobject_item.add('tel').value = data.phone if data.email: vobject_item.add('email').value = data.email if data.website: vobject_item.add('url').value = data.website if data.function: vobject_item.add('role').value = data.function if data.comment: vobject_item.add('note').value = data.comment if data.category_id: categs = [categ.name for categ in data.category_id] vobject_item.add('categories').value = categs if data.parent_id: vobject_item.add('org').value = [data.parent_id.name] vobject_item.add('uid').value = href vobject_item.add('rev').value = last_modified tag, start, end = xmlutils.find_tag_and_time_range(vobject_item) text = vobject_item.serialize() etag = get_etag(text) uid = get_uid_from_object(vobject_item) return Item( self, href=href, last_modified=last_modified, etag=etag, text=text, item=vobject_item, uid=href, name=vobject_item.name, component_name=tag), (tag, start, end) elif model == 'calendar.event': raise NotImplementedError else: raise NotImplementedError def get(self, href, verify_href=True): item, metadata = self._get_with_metadata(href) return item def delete(self, href=None): """ Can not delete collection but item, yes """ if href is None: # Delete the collection self.logger.error('Attempt to delete collection %s' % self.path) raise ValueError('Can not delete collection') else: # Delete an item raise NotImplementedError @property def last_modified(self): """ Return last modified """ last = self.odoo.env[self.odoo_model].search([], limit=1, order='write_date desc') last_fields = self.odoo.execute(self.odoo_model, 'read', last, ['write_date'])[0] self.logger.debug(last_fields) # return strftime("%a, %d %b %Y %H:%M:%S GMT", # strptime(last_fields.get('write_date'), '%Y-%m-%d %H:%M:%S')) return str(last_fields['write_date'])