addons/im/im.py

   1 # -*- coding: utf-8 -*-
   2 ##############################################################################
   3 #
   4 #    OpenERP, Open Source Management Solution
   5 #    Copyright (C) 2004-2010 Tiny SPRL (<http://tiny.be>).
   6 #
   7 #    This program is free software: you can redistribute it and/or modify
   8 #    it under the terms of the GNU Affero General Public License as
   9 #    published by the Free Software Foundation, either version 3 of the
  10 #    License, or (at your option) any later version.
  11 #
  12 #    This program is distributed in the hope that it will be useful,
  13 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 #    GNU Affero General Public License for more details.
  16 #
  17 #    You should have received a copy of the GNU Affero General Public License
  18 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19 #
  20 ##############################################################################
  21 import datetime
  22 import json
  23 import logging
  24 import select
  25 import time
  26
  27 import openerp
  28 import openerp.tools.config
  29 import openerp.modules.registry
  30 from openerp import http
  31 from openerp.http import request
  32 from openerp.osv import osv, fields, expression
  33 from openerp.tools.misc import DEFAULT_SERVER_DATETIME_FORMAT
  34
  35 _logger = logging.getLogger(__name__)
  36
  37 def listen_channel(cr, channel_name, handle_message, check_stop=(lambda: False), check_stop_timer=60.):
  38     """
  39         Begin a loop, listening on a PostgreSQL channel. This method does never terminate by default, you need to provide a check_stop
  40         callback to do so. This method also assume that all notifications will include a message formated using JSON (see the
  41         corresponding notify_channel() method).
  42
  43         :param db_name: database name
  44         :param channel_name: the name of the PostgreSQL channel to listen
  45         :param handle_message: function that will be called when a message is received. It takes one argument, the message
  46             attached to the notification.
  47         :type handle_message: function (one argument)
  48         :param check_stop: function that will be called periodically (see the check_stop_timer argument). If it returns True
  49             this function will stop to watch the channel.
  50         :type check_stop: function (no arguments)
  51         :param check_stop_timer: The maximum amount of time between calls to check_stop_timer (can be shorter if messages
  52             are received).
  53     """
  54     try:
  55         conn = cr._cnx
  56         cr.execute("listen " + channel_name + ";")
  57         cr.commit();
  58         stopping = False
  59         while not stopping:
  60             if check_stop():
  61                 stopping = True
  62                 break
  63             if select.select([conn], [], [], check_stop_timer) == ([],[],[]):
  64                 pass
  65             else:
  66                 conn.poll()
  67                 while conn.notifies:
  68                     message = json.loads(conn.notifies.pop().payload)
  69                     handle_message(message)
  70     finally:
  71         try:
  72             cr.execute("unlisten " + channel_name + ";")
  73             cr.commit()
  74         except:
  75             pass # can't do anything if that fails
  76
  77 def notify_channel(cr, channel_name, message):
  78     """
  79         Send a message through a PostgreSQL channel. The message will be formatted using JSON. This method will
  80         commit the given transaction because the notify command in Postgresql seems to work correctly when executed in
  81         a separate transaction (despite what is written in the documentation).
  82
  83         :param cr: The cursor.
  84         :param channel_name: The name of the PostgreSQL channel.
  85         :param message: The message, must be JSON-compatible data.
  86     """
  87     cr.commit()
  88     cr.execute("notify " + channel_name + ", %s", [json.dumps(message)])
  89     cr.commit()
  90
  91 POLL_TIMER = 30
  92 DISCONNECTION_TIMER = POLL_TIMER + 5
  93 WATCHER_ERROR_DELAY = 10
  94
  95 class LongPollingController(http.Controller):
  96
  97     @http.route('/longpolling/im/poll', type="json", auth="none")
  98     def poll(self, last=None, users_watch=None, db=None, uid=None, password=None, uuid=None):
  99         assert_uuid(uuid)
 100         if not openerp.evented:
 101             raise Exception("Not usable in a server not running gevent")
 102         from openerp.addons.im.watcher import ImWatcher
 103         if db is not None:
 104             openerp.service.security.check(db, uid, password)
 105         else:
 106             uid = request.session.uid
 107             db = request.session.db
 108
 109         registry = openerp.modules.registry.RegistryManager.get(db)
 110         with registry.cursor() as cr:
 111             registry.get('im.user').im_connect(cr, uid, uuid=uuid, context=request.context)
 112             my_id = registry.get('im.user').get_my_id(cr, uid, uuid, request.context)
 113         num = 0
 114         while True:
 115             with registry.cursor() as cr:
 116                 res = registry.get('im.message').get_messages(cr, uid, last, users_watch, uuid=uuid, context=request.context)
 117             if num >= 1 or len(res["res"]) > 0:
 118                 return res
 119             last = res["last"]
 120             num += 1
 121             ImWatcher.get_watcher(res["dbname"]).stop(my_id, users_watch or [], POLL_TIMER)
 122
 123     @http.route('/longpolling/im/activated', type="json", auth="none")
 124     def activated(self):
 125         return not not openerp.evented
 126
 127     @http.route('/longpolling/im/gen_uuid', type="json", auth="none")
 128     def gen_uuid(self):
 129         import uuid
 130         return "%s" % uuid.uuid1()
 131
 132 def assert_uuid(uuid):
 133     if not isinstance(uuid, (str, unicode, type(None))) and uuid != False:
 134         raise Exception("%s is not a uuid" % uuid)
 135
 136
 137 class im_message(osv.osv):
 138     _name = 'im.message'
 139
 140     _order = "date desc"
 141
 142     _columns = {
 143         'message': fields.text(string="Message", required=True),
 144         'from_id': fields.many2one("im.user", "From", required= True, ondelete='cascade'),
 145         'session_id': fields.many2one("im.session", "Session", required=True, select=True, ondelete='cascade'),
 146         'to_id': fields.many2many("im.user", "im_message_users", 'message_id', 'user_id', 'To'),
 147         'date': fields.datetime("Date", required=True, select=True),
 148         'technical': fields.boolean("Technical Message"),
 149     }
 150
 151     _defaults = {
 152         'date': lambda *args: datetime.datetime.now().strftime(DEFAULT_SERVER_DATETIME_FORMAT),
 153         'technical': False,
 154     }
 155
 156     def get_messages(self, cr, uid, last=None, users_watch=None, uuid=None, context=None):
 157         assert_uuid(uuid)
 158         users_watch = users_watch or []
 159
 160         # complex stuff to determine the last message to show
 161         users = self.pool.get("im.user")
 162         my_id = users.get_my_id(cr, uid, uuid, context=context)
 163         c_user = users.browse(cr, openerp.SUPERUSER_ID, my_id, context=context)
 164         if last:
 165             if c_user.im_last_received < last:
 166                 users.write(cr, openerp.SUPERUSER_ID, my_id, {'im_last_received': last}, context=context)
 167         else:
 168             last = c_user.im_last_received or -1
 169
 170         # how fun it is to always need to reorder results from read
 171         mess_ids = self.search(cr, openerp.SUPERUSER_ID, ["&", ['id', '>', last], "|", ['from_id', '=', my_id], ['to_id', 'in', [my_id]]], order="id", context=context)
 172         mess = self.read(cr, openerp.SUPERUSER_ID, mess_ids, ["id", "message", "from_id", "session_id", "date", "technical"], context=context)
 173         index = {}
 174         for i in xrange(len(mess)):
 175             index[mess[i]["id"]] = mess[i]
 176         mess = []
 177         for i in mess_ids:
 178             mess.append(index[i])
 179
 180         if len(mess) > 0:
 181             last = mess[-1]["id"]
 182         users_status = users.read(cr, openerp.SUPERUSER_ID, users_watch, ["im_status"], context=context)
 183         return {"res": mess, "last": last, "dbname": cr.dbname, "users_status": users_status}
 184
 185     def post(self, cr, uid, message, to_session_id, technical=False, uuid=None, context=None):
 186         assert_uuid(uuid)
 187         my_id = self.pool.get('im.user').get_my_id(cr, uid, uuid)
 188         session = self.pool.get('im.session').browse(cr, uid, to_session_id, context)
 189         to_ids = [x.id for x in session.user_ids if x.id != my_id]
 190         self.create(cr, openerp.SUPERUSER_ID, {"message": message, 'from_id': my_id,
 191             'to_id': [(6, 0, to_ids)], 'session_id': to_session_id, 'technical': technical}, context=context)
 192         notify_channel(cr, "im_channel", {'type': 'message', 'receivers': [my_id] + to_ids})
 193         return False
 194
 195 class im_session(osv.osv):
 196     _name = 'im.session'
 197
 198     def _calc_name(self, cr, uid, ids, something, something_else, context=None):
 199         res = {}
 200         for obj in self.browse(cr, uid, ids, context=context):
 201             res[obj.id] = ", ".join([x.name for x in obj.user_ids])
 202         return res
 203
 204     _columns = {
 205         'user_ids': fields.many2many('im.user'),
 206         "name": fields.function(_calc_name, string="Name", type='char'),
 207     }
 208
 209     # Todo: reuse existing sessions if possible
 210     def session_get(self, cr, uid, users_to, uuid=None, context=None):
 211         my_id = self.pool.get("im.user").get_my_id(cr, uid, uuid, context=context)
 212         users = [my_id] + users_to
 213         domain = []
 214         for user_to in users:
 215             domain.append(('user_ids', 'in', [user_to]))
 216         sids = self.search(cr, openerp.SUPERUSER_ID, domain, context=context, limit=1)
 217         session_id = None
 218         for session in self.browse(cr, uid, sids, context=context):
 219             if len(session.user_ids) == len(users):
 220                 session_id = session.id
 221                 break
 222         if not session_id:
 223             session_id = self.create(cr, openerp.SUPERUSER_ID, {
 224                 'user_ids': [(6, 0, users)]
 225             }, context=context)
 226         return self.read(cr, uid, session_id, context=context)
 227
 228     def add_to_session(self, cr, uid, session_id, user_id, uuid=None, context=None):
 229         my_id = self.pool.get("im.user").get_my_id(cr, uid, uuid, context=context)
 230         session = self.read(cr, uid, session_id, context=context)
 231         if my_id not in session.get("user_ids"):
 232             raise Exception("Not allowed to modify a session when you are not in it.")
 233         self.write(cr, uid, session_id, {"user_ids": [(4, user_id)]}, context=context)
 234
 235     def remove_me_from_session(self, cr, uid, session_id, uuid=None, context=None):
 236         my_id = self.pool.get("im.user").get_my_id(cr, uid, uuid, context=context)
 237         self.write(cr, openerp.SUPERUSER_ID, session_id, {"user_ids": [(3, my_id)]}, context=context)
 238
 239 class im_user(osv.osv):
 240     _name = "im.user"
 241
 242     def _im_status(self, cr, uid, ids, something, something_else, context=None):
 243         res = {}
 244         current = datetime.datetime.now()
 245         delta = datetime.timedelta(0, DISCONNECTION_TIMER)
 246         data = self.read(cr, openerp.SUPERUSER_ID, ids, ["im_last_status_update", "im_last_status"], context=context)
 247         for obj in data:
 248             last_update = datetime.datetime.strptime(obj["im_last_status_update"], DEFAULT_SERVER_DATETIME_FORMAT)
 249             res[obj["id"]] = obj["im_last_status"] and (last_update + delta) > current
 250         return res
 251
 252     def _status_search(self, cr, uid, obj, name, domain, context=None):
 253         current = datetime.datetime.now()
 254         delta = datetime.timedelta(0, DISCONNECTION_TIMER)
 255         field, operator, value = domain[0]
 256         if operator in expression.NEGATIVE_TERM_OPERATORS:
 257             value = not value
 258         if value:
 259             return ['&', ('im_last_status', '=', True), ('im_last_status_update', '>', (current - delta).strftime(DEFAULT_SERVER_DATETIME_FORMAT))]
 260         else:
 261             return ['|', ('im_last_status', '=', False), ('im_last_status_update', '<=', (current - delta).strftime(DEFAULT_SERVER_DATETIME_FORMAT))]
 262
 263     def search_users(self, cr, uid, text_search, fields, limit, context=None):
 264         my_id = self.get_my_id(cr, uid, None, context)
 265         group_employee = self.pool['ir.model.data'].get_object_reference(cr, uid, 'base', 'group_user')[1]
 266         found = self.search(cr, uid, [["name", "ilike", text_search], ["id", "<>", my_id], ["uuid", "=", False], ["im_status", "=", True], ["user_id.groups_id", "in", [group_employee]]],
 267             order="name asc", limit=limit, context=context)
 268         if len(found) < limit:
 269             found += self.search(cr, uid, [["name", "ilike", text_search], ["id", "<>", my_id], ["uuid", "=", False], ["im_status", "=", True], ["id", "not in", found]],
 270                 order="name asc", limit=limit, context=context)
 271         if len(found) < limit:
 272             found += self.search(cr, uid, [["name", "ilike", text_search], ["id", "<>", my_id], ["uuid", "=", False], ["im_status", "=", False], ["id", "not in", found]],
 273                 order="name asc", limit=limit-len(found), context=context)
 274         users = self.read(cr, uid, found, fields, context=context)
 275         users.sort(key=lambda obj: found.index(obj['id']))
 276         return users
 277
 278     def im_connect(self, cr, uid, uuid=None, context=None):
 279         assert_uuid(uuid)
 280         return self._im_change_status(cr, uid, True, uuid, context)
 281
 282     def im_disconnect(self, cr, uid, uuid=None, context=None):
 283         assert_uuid(uuid)
 284         return self._im_change_status(cr, uid, False, uuid, context)
 285
 286     def _im_change_status(self, cr, uid, new_one, uuid=None, context=None):
 287         assert_uuid(uuid)
 288         id = self.get_my_id(cr, uid, uuid, context=context)
 289         current_status = self.read(cr, openerp.SUPERUSER_ID, id, ["im_status"], context=None)["im_status"]
 290         self.write(cr, openerp.SUPERUSER_ID, id, {"im_last_status": new_one,
 291             "im_last_status_update": datetime.datetime.now().strftime(DEFAULT_SERVER_DATETIME_FORMAT)}, context=context)
 292         if current_status != new_one:
 293             notify_channel(cr, "im_channel", {'type': 'status', 'user': id})
 294         return True
 295
 296     def get_my_id(self, cr, uid, uuid=None, context=None):
 297         assert_uuid(uuid)
 298         if uuid:
 299             users = self.search(cr, openerp.SUPERUSER_ID, [["uuid", "=", uuid]], context=None)
 300         else:
 301             users = self.search(cr, openerp.SUPERUSER_ID, [["user_id", "=", uid]], context=None)
 302         my_id = users[0] if len(users) >= 1 else False
 303         if not my_id:
 304             my_id = self.create(cr, openerp.SUPERUSER_ID, {"user_id": uid if not uuid else False, "uuid": uuid if uuid else False}, context=context)
 305         return my_id
 306
 307     def assign_name(self, cr, uid, uuid, name, context=None):
 308         assert_uuid(uuid)
 309         id = self.get_my_id(cr, uid, uuid, context=context)
 310         self.write(cr, openerp.SUPERUSER_ID, id, {"assigned_name": name}, context=context)
 311         return True
 312
 313     def _get_name(self, cr, uid, ids, name, arg, context=None):
 314         res = {}
 315         for record in self.browse(cr, uid, ids, context=context):
 316             res[record.id] = record.assigned_name
 317             if record.user_id:
 318                 res[record.id] = record.user_id.name
 319                 continue
 320         return res
 321
 322     _columns = {
 323         'name': fields.function(_get_name, type='char', size=200, string="Name", store=True, readonly=True),
 324         'assigned_name': fields.char(string="Assigned Name", size=200, required=False),
 325         'image': fields.related('user_id', 'image_small', type='binary', string="Image", readonly=True),
 326         'user_id': fields.many2one("res.users", string="User", select=True, ondelete='cascade', oldname='user'),
 327         'uuid': fields.char(string="UUID", size=50, select=True),
 328         'im_last_received': fields.integer(string="Instant Messaging Last Received Message"),
 329         'im_last_status': fields.boolean(strint="Instant Messaging Last Status"),
 330         'im_last_status_update': fields.datetime(string="Instant Messaging Last Status Update"),
 331         'im_status': fields.function(_im_status, string="Instant Messaging Status", type='boolean', fnct_search=_status_search),
 332     }
 333
 334     _defaults = {
 335         'im_last_received': -1,
 336         'im_last_status': False,
 337         'im_last_status_update': lambda *args: datetime.datetime.now().strftime(DEFAULT_SERVER_DATETIME_FORMAT),
 338     }
 339
 340     _sql_constraints = [
 341         ('user_uniq', 'unique (user_id)', 'Only one chat user per OpenERP user.'),
 342         ('uuid_uniq', 'unique (uuid)', 'Chat identifier already used.'),
 343     ]