# copyright 2003-2010 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 rql.nodes import Constant

from rql.utils import rqlvar_maker

from cubicweb import dbapi, server

from cubicweb import BadConnectionId, UnknownEid, ConnectionError

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)

def load_mapping_file(mappingfile):

    mapping = {}

    execfile(mappingfile, mapping)

    for junk in ('__builtins__', '__doc__'):

        mapping.pop(junk, None)

    mapping.setdefault('support_relations', {})

    mapping.setdefault('dont_cross_relations', set())

    mapping.setdefault('cross_relations', set())

    # do some basic checks of the mapping content

    assert 'support_entities' in mapping, \

           'mapping file should at least define support_entities'

    assert isinstance(mapping['support_entities'], dict)

    assert isinstance(mapping['support_relations'], dict)

    assert isinstance(mapping['dont_cross_relations'], set)

    assert isinstance(mapping['cross_relations'], set)

    unknown = set(mapping) - set( ('support_entities', 'support_relations',

                                   'dont_cross_relations', 'cross_relations') )

    assert not unknown, 'unknown mapping attribute(s): %s' % unknown

    # relations that are necessarily not crossed

    mapping['dont_cross_relations'] |= set(('owned_by', 'created_by'))

    for rtype in ('is', 'is_instance_of', 'cw_source'):

        assert rtype not in mapping['dont_cross_relations'], \

               '%s relation should not be in dont_cross_relations' % rtype

        assert rtype not in mapping['support_relations'], \

               '%s relation should not be in support_relations' % rtype

    return mapping

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,

          }),

        ('mapping-file',

         {'type' : 'string',

          'default': REQUIRED,

          'help': 'path to a python file with the schema mapping definition',

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

          }),

        ('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',

         {'type' : 'int',

          'default': 5*60,

          '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, *args, **kwargs):

        AbstractSource.__init__(self, repo, source_config, *args, **kwargs)

        mappingfile = source_config['mapping-file']

        if not mappingfile[0] == '/':

            mappingfile = join(repo.config.apphome, mappingfile)

        try:

            mapping = load_mapping_file(mappingfile)

        except IOError:

            self.disabled = True

            self.error('cant read mapping file %s, source disabled',

                       mappingfile)

            self.support_entities = {}

            self.support_relations = {}

            self.dont_cross_relations = set()

            self.cross_relations = set()

        else:

            self.support_entities = mapping['support_entities']

            self.support_relations = mapping['support_relations']

            self.dont_cross_relations = mapping['dont_cross_relations']

            self.cross_relations = mapping['cross_relations']

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

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

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

        self.config = source_config

        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)

        self._skip_externals = check_yn(None, 'skip-external-entities',

                                        source_config.get('skip-external-entities', False))

    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):

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

        interval = int(self.config.get('synchronization-interval', 5*60))

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

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

                               self._query_cache.clear_expired)

    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=True)

                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,

                           varmap=None):

        assert dbg_st_search(self.uri, union, varmap, args, cachekey)

        rqlkey = union.as_string(kwargs=args)

        try:

            results = self._query_cache[rqlkey]

        except KeyError:

            results = self._syntax_tree_search(session, union, args)

            self._query_cache[rqlkey] = results

        assert dbg_results(results)

        return results

    def _syntax_tree_search(self, session, union, args):

        """return result from this source for a rql query (actually from a rql

        syntax tree and a solution dictionary mapping each used variable to a

        possible type). If cachekey is given, the query necessary to fetch the

        results (but not the results themselves) may be cached using this key.

        """

        if not args is None:

            args = args.copy()

        # get cached cursor anyway

        cu = session.pool[self.uri]

        if cu is None:

            # this is a ConnectionWrapper instance

            msg = session._("can't connect to source %s, some data may be missing")

            session.set_shared_data('sources_error', msg % self.uri)

            return []

        translator = RQL2RQL(self)

        try:

            rql = translator.generate(session, union, args)

        except UnknownEid, ex:

            if server.DEBUG:

                print '  unknown eid', ex, 'no results'

            return []

        if server.DEBUG & server.DBG_RQL:

            print '  translated rql', rql

        try:

            rset = cu.execute(rql, args)

        except Exception, ex:

            self.exception(str(ex))

            msg = session._("error while querying source %s, some data may be missing")

            session.set_shared_data('sources_error', msg % self.uri)

            return []

        descr = rset.description

        if rset:

            needtranslation = []

            rows = rset.rows

            for i, etype in enumerate(descr[0]):

                if (etype is None or not self.schema.eschema(etype).final