# 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"

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

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

    support_entities = None

    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,

          }),

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

          }),

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

        AbstractSource.__init__(self, repo, appschema, source_config,

                                *args, **kwargs)

        mappingfile = source_config['mapping-file']

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

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

        mapping = {}

        execfile(mappingfile, mapping)

        self.support_entities = mapping['support_entities']

        self.support_relations = mapping.get('support_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)

    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

        rql = 'Any V WHERE X is CWProperty, X value V, X pkey %(k)s'

        session = self.repo.internal_session()

        try:

            rset = session.execute(rql, {'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 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:

                    exturi = cnx.describe(extid)[1]

                    if exturi == 'system' or not exturi in repo.sources_by_uri:

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

                        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)

                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 []

        try:

            rql = RQL2RQL(self).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

                    or uidtype(union, i, etype, args)):

                    needtranslation.append(i)

            if needtranslation:

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

                for rowindex in xrange(rset.rowcount - 1, -1, -1):

                    row = rows[rowindex]

                    for colindex in needtranslation:

                        if row[colindex] is not None: # optional variable

                            etype = descr[rowindex][colindex]

                            exttype, exturi, extid = cnx.describe(row[colindex])

                            if exturi == 'system' or not exturi in self.repo.sources_by_uri:

                                eid = self.extid2eid(str(row[colindex]), etype,

                                                     session)

                                row[colindex] = eid

                            else:

                                # skip this row

                                del rows[rowindex]

                                del descr[rowindex]

                                break

            results = rows

        else:

            results = []

        return results

    def _entity_relations_and_kwargs(self, session, entity):

        relations = []

        kwargs = {'x': self.eid2extid(entity.eid, session)}

        for key, val in entity.iteritems():

            relations.append('X %s %%(%s)s' % (key, key))

            kwargs[key] = val

        return relations, kwargs

    def add_entity(self, session, entity):

        """add a new entity to the source"""

        raise NotImplementedError()

    def update_entity(self, session, entity):

        """update an entity in the source"""

        relations, kwargs = self._entity_relations_and_kwargs(session, entity)

        cu = session.pool[self.uri]

        cu.execute('SET %s WHERE X eid %%(x)s' % ','.join(relations), kwargs)

        self._query_cache.clear()

        entity.clear_all_caches()

    def delete_entity(self, session, entity):

        """delete an entity from the source"""

        cu = session.pool[self.uri]

        cu.execute('DELETE %s X WHERE X eid %%(x)s' % entity.__regid__,

                   {'x': self.eid2extid(entity.eid, session)})

        self._query_cache.clear()

    def add_relation(self, session, subject, rtype, object):

        """add a relation to the source"""

        cu = session.pool[self.uri]

        cu.execute('SET X %s Y WHERE X eid %%(x)s, Y eid %%(y)s' % rtype,