# copyright 2003-2011 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr

#

# This file is part of CubicWeb.

#

# CubicWeb is free software: you can redistribute it and/or modify it under the

# terms of the GNU Lesser General Public License as published by the Free

# Software Foundation, either version 2.1 of the License, or (at your option)

# any later version.

#

# CubicWeb 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 Lesser General Public License for more

# details.

#

# You should have received a copy of the GNU Lesser General Public License along

# with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.

"""Source to query another RQL repository using pyro"""

__docformat__ = "restructuredtext en"

_ = unicode

import threading

from os.path import join

from time import mktime

from datetime import datetime

from base64 import b64decode

from Pyro.errors import PyroError, ConnectionClosedError

from logilab.common.configuration import REQUIRED

from logilab.common.optik_ext import check_yn

from yams.schema import role_name

from rql.nodes import Constant

from rql.utils import rqlvar_maker

from cubicweb import dbapi, server

from cubicweb import ValidationError, BadConnectionId, UnknownEid, ConnectionError

from cubicweb.schema import VIRTUAL_RTYPES

from cubicweb.cwconfig import register_persistent_options

from cubicweb.server.sources import (AbstractSource, ConnectionWrapper,

                                     TimedCache, dbg_st_search, dbg_results)

from cubicweb.server.msplanner import neged_relation

def uidtype(union, col, etype, args):

    select, col = union.locate_subquery(col, etype, args)

    return getattr(select.selection[col], 'uidtype', None)

class ReplaceByInOperator(Exception):

    def __init__(self, eids):

        self.eids = eids

class PyroRQLSource(AbstractSource):

    """External repository source, using Pyro connection"""

    # boolean telling if modification hooks should be called when something is

    # modified in this source

    should_call_hooks = False

    # boolean telling if the repository should connect to this source during

    # migration

    connect_for_migration = False

    options = (

        # XXX pyro-ns host/port

        ('pyro-ns-id',

         {'type' : 'string',

          'default': REQUIRED,

          'help': 'identifier of the repository in the pyro name server',

          'group': 'pyro-source', 'level': 0,

          }),

        ('cubicweb-user',

         {'type' : 'string',

          'default': REQUIRED,

          'help': 'user to use for connection on the distant repository',

          'group': 'pyro-source', 'level': 0,

          }),

        ('cubicweb-password',

         {'type' : 'password',

          'default': '',

          'help': 'user to use for connection on the distant repository',

          'group': 'pyro-source', 'level': 0,

          }),

        ('base-url',

         {'type' : 'string',

          'default': '',

          'help': 'url of the web site for the distant repository, if you want '

          'to generate external link to entities from this repository',

          'group': 'pyro-source', 'level': 1,

          }),

        ('skip-external-entities',

         {'type' : 'yn',

          'default': False,

          'help': 'should entities not local to the source be considered or not',

          'group': 'pyro-source', 'level': 0,

          }),

        ('pyro-ns-host',

         {'type' : 'string',

          'default': None,

          'help': 'Pyro name server\'s host. If not set, default to the value \

from all_in_one.conf. It may contains port information using <host>:<port> notation.',

          'group': 'pyro-source', 'level': 1,

          }),

        ('pyro-ns-group',

         {'type' : 'string',

          'default': None,

          'help': 'Pyro name server\'s group where the repository will be \

registered. If not set, default to the value from all_in_one.conf.',

          'group': 'pyro-source', 'level': 2,

          }),

        ('synchronization-interval',

          'help': 'interval between synchronization with the external \

repository (default to 5 minutes).',

          'group': 'pyro-source', 'level': 2,

          }),

    )

    PUBLIC_KEYS = AbstractSource.PUBLIC_KEYS + ('base-url',)

    _conn = None

    def __init__(self, repo, source_config, eid=None):

        AbstractSource.__init__(self, repo, source_config, eid)

        self.update_config(None, self.check_conf_dict(eid, source_config,

                                                      fail_if_unknown=False))

        myoptions = (('%s.latest-update-time' % self.uri,

                      {'type' : 'int', 'sitewide': True,

                       'default': 0,

                       'help': _('timestamp of the latest source synchronization.'),

                       'group': 'sources',

                       }),)

        register_persistent_options(myoptions)

        self._query_cache = TimedCache(1800)

    def update_config(self, source_entity, processed_config):

        """update configuration from source entity"""

        # XXX get it through pyro if unset

        baseurl = processed_config.get('base-url')

        if baseurl and not baseurl.endswith('/'):

            processed_config['base-url'] += '/'

        self.config = processed_config

        self._skip_externals = processed_config['skip-external-entities']

    def reset_caches(self):

        """method called during test to reset potential source caches"""

        self._query_cache = TimedCache(1800)

    def last_update_time(self):

        pkey = u'sources.%s.latest-update-time' % self.uri

        session = self.repo.internal_session()

        try:

            rset = session.execute('Any V WHERE X is CWProperty, X value V, X pkey %(k)s',

                                   {'k': pkey})

            if not rset:

                # insert it

                session.execute('INSERT CWProperty X: X pkey %(k)s, X value %(v)s',

                                {'k': pkey, 'v': u'0'})

                session.commit()

                timestamp = 0

            else:

                assert len(rset) == 1

                timestamp = int(rset[0][0])

            return datetime.fromtimestamp(timestamp)

        finally:

            session.close()

    def init(self, activated, session=None):

        """method called by the repository once ready to handle request"""

        self.load_mapping(session)

        if activated:

            self.repo.looping_task(interval, self.synchronize)

            self.repo.looping_task(self._query_cache.ttl.seconds/10,

                                   self._query_cache.clear_expired)

    def load_mapping(self, session=None):

        self.support_entities = {}

        self.support_relations = {}

        self.dont_cross_relations = set(('owned_by', 'created_by'))

        self.cross_relations = set()

        assert self.eid is not None

        self._schemacfg_idx = {}

        if session is None:

            _session = self.repo.internal_session()

        else:

            _session = session

        try:

            for schemacfg in _session.execute(

                'Any CFG,CFGO,SN,S WHERE '

                'CFG options CFGO, CFG cw_schema S, S name SN, '

                'CFG cw_for_source X, X eid %(x)s', {'x': self.eid}).entities():

                self.add_schema_config(schemacfg)

        finally:

            if session is None:

                _session.close()

    etype_options = set(('write',))

    rtype_options = set(('maycross', 'dontcross', 'write',))

    def _check_options(self, schemacfg, allowedoptions):

        if schemacfg.options:

            options = set(w.strip() for w in schemacfg.options.split(':'))

        else:

            options = set()

        if options - allowedoptions:

            options = ', '.join(sorted(options - allowedoptions))

            msg = _('unknown option(s): %s' % options)

            raise ValidationError(schemacfg.eid, {role_name('options', 'subject'): msg})

        return options

    def add_schema_config(self, schemacfg, checkonly=False):

        """added CWSourceSchemaConfig, modify mapping accordingly"""

        try:

            ertype = schemacfg.schema.name

        except AttributeError:

            msg = schemacfg._cw._("attribute/relation can't be mapped, only "

                                  "entity and relation types")

            raise ValidationError(schemacfg.eid, {role_name('cw_for_schema', 'subject'): msg})

        if schemacfg.schema.__regid__ == 'CWEType':

            options = self._check_options(schemacfg, self.etype_options)

            if not checkonly:

                self.support_entities[ertype] = 'write' in options

        else: # CWRType

            if ertype in ('is', 'is_instance_of', 'cw_source') or ertype in VIRTUAL_RTYPES:

                msg = schemacfg._cw._('%s relation should not be in mapped') % rtype

                raise ValidationError(schemacfg.eid, {role_name('cw_for_schema', 'subject'): msg})

            options = self._check_options(schemacfg, self.rtype_options)

            if 'dontcross' in options:

                if 'maycross' in options:

                    msg = schemacfg._("can't mix dontcross and maycross options")

                    raise ValidationError(schemacfg.eid, {role_name('options', 'subject'): msg})

                if 'write' in options:

                    msg = schemacfg._("can't mix dontcross and write options")

                    raise ValidationError(schemacfg.eid, {role_name('options', 'subject'): msg})

                if not checkonly:

                    self.dont_cross_relations.add(ertype)

            elif not checkonly:

                self.support_relations[ertype] = 'write' in options

                if 'maycross' in options:

                    self.cross_relations.add(ertype)

        if not checkonly:

            # add to an index to ease deletion handling

            self._schemacfg_idx[schemacfg.eid] = ertype

    def del_schema_config(self, schemacfg, checkonly=False):

        """deleted CWSourceSchemaConfig, modify mapping accordingly"""

        if checkonly:

            return

        try:

            ertype = self._schemacfg_idx[schemacfg.eid]

            if ertype[0].isupper():

                del self.support_entities[ertype]

            else:

                if ertype in self.support_relations:

                    del self.support_relations[ertype]

                    if ertype in self.cross_relations:

                        self.cross_relations.remove(ertype)

                else:

                    self.dont_cross_relations.remove(ertype)

        except:

            self.error('while updating mapping consequently to removal of %s',

                       schemacfg)

    def local_eid(self, cnx, extid, session):

        etype, dexturi, dextid = cnx.describe(extid)

        if dexturi == 'system' or not (

            dexturi in self.repo.sources_by_uri or self._skip_externals):

            return self.repo.extid2eid(self, str(extid), etype, session), True

        if dexturi in self.repo.sources_by_uri:

            source = self.repo.sources_by_uri[dexturi]

            cnx = session.pool.connection(source.uri)

            eid = source.local_eid(cnx, dextid, session)[0]

            return eid, False

        return None, None

    def synchronize(self, mtime=None):

        """synchronize content known by this repository with content in the

        external repository

        """

        self.info('synchronizing pyro source %s', self.uri)

        cnx = self.get_connection()

        try:

            extrepo = cnx._repo

        except AttributeError:

            # fake connection wrapper returned when we can't connect to the

            # external source (hence we've no chance to synchronize...)

            return

        etypes = self.support_entities.keys()

        if mtime is None:

            mtime = self.last_update_time()

        updatetime, modified, deleted = extrepo.entities_modified_since(

            etypes, mtime)

        self._query_cache.clear()

        repo = self.repo

        session = repo.internal_session()

        source = repo.system_source

        try:

            for etype, extid in modified:

                try:

                    eid = self.local_eid(cnx, extid, session)[0]

                    if eid is not None:

                        rset = session.eid_rset(eid, etype)

                        entity = rset.get_entity(0, 0)

                        entity.complete(entity.e_schema.indexable_attributes())

                        source.index_entity(session, entity)

                except:

                    self.exception('while updating %s with external id %s of source %s',

                                   etype, extid, self.uri)

                    continue

            for etype, extid in deleted:

                try:

                    eid = self.extid2eid(str(extid), etype, session,

                                         insert=False)

                    # entity has been deleted from external repository but is not known here

                    if eid is not None:

                        entity = session.entity_from_eid(eid, etype)

                        repo.delete_info(session, entity, self.uri, extid,

                                         scleanup=self.eid)

                except:

                    self.exception('while updating %s with external id %s of source %s',

                                   etype, extid, self.uri)

                    continue

            session.execute('SET X value %(v)s WHERE X pkey %(k)s',

                            {'k': u'sources.%s.latest-update-time' % self.uri,

                             'v': unicode(int(mktime(updatetime.timetuple())))})

            session.commit()

        finally:

            session.close()

    def _get_connection(self):

        """open and return a connection to the source"""

        nshost = self.config.get('pyro-ns-host') or self.repo.config['pyro-ns-host']

        nsgroup = self.config.get('pyro-ns-group') or self.repo.config['pyro-ns-group']

        self.info('connecting to instance :%s.%s for user %s',

                  nsgroup, self.config['pyro-ns-id'], self.config['cubicweb-user'])

        #cnxprops = ConnectionProperties(cnxtype=self.config['cnx-type'])

        return dbapi.connect(database=self.config['pyro-ns-id'],

                             login=self.config['cubicweb-user'],

                             password=self.config['cubicweb-password'],

                             host=nshost, group=nsgroup,

                             setvreg=False) #cnxprops=cnxprops)

    def get_connection(self):

        try:

            return self._get_connection()

        except (ConnectionError, PyroError):

            self.critical("can't get connection to source %s", self.uri,

                          exc_info=1)

            return ConnectionWrapper()

    def check_connection(self, cnx):

        """check connection validity, return None if the connection is still valid

        else a new connection

        """

        # we have to transfer manually thread ownership. This can be done safely

        # since the pool to which belong the connection is affected to one

        # session/thread and can't be called simultaneously

        try:

            cnx._repo._transferThread(threading.currentThread())

        except AttributeError:

            # inmemory connection

            pass

        if not isinstance(cnx, ConnectionWrapper):

            try:

                cnx.check()

                return # ok

            except (BadConnectionId, ConnectionClosedError):

                pass

        # try to reconnect

        return self.get_connection()

    def syntax_tree_search(self, session, union, args=None, cachekey=None,