forked from OCA/connector-telephony
-
Notifications
You must be signed in to change notification settings - Fork 0
/
asterisk_click2dial.py
405 lines (372 loc) · 16.6 KB
/
asterisk_click2dial.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
# -*- coding: utf-8 -*-
# © 2010-2016 Akretion (Alexis de Lattre <alexis.delattre@akretion.com>)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
from openerp import models, fields, api, _
from openerp.exceptions import UserError, ValidationError
import logging
from pprint import pformat
try:
# pip install py-Asterisk
from Asterisk import Manager
except ImportError:
Manager = None
_logger = logging.getLogger(__name__)
class AsteriskServer(models.Model):
'''Asterisk server object, stores the parameters of the Asterisk IPBXs'''
_name = "asterisk.server"
_description = "Asterisk Servers"
name = fields.Char(string='Asterisk Server Name', required=True)
active = fields.Boolean(
string='Active', default=True)
ip_address = fields.Char(
string='Asterisk IP address or DNS', required=True)
port = fields.Integer(
string='Port', required=True, default=5038,
help="TCP port on which the Asterisk Manager Interface listens. "
"Defined in /etc/asterisk/manager.conf on Asterisk.")
out_prefix = fields.Char(
string='Out Prefix', size=4, help="Prefix to dial to make outgoing "
"calls. If you don't use a prefix to make outgoing calls, "
"leave empty.")
login = fields.Char(
string='AMI Login', required=True,
help="Login that Odoo will use to communicate with the "
"Asterisk Manager Interface. Refer to /etc/asterisk/manager.conf "
"on your Asterisk server.")
password = fields.Char(
string='AMI Password', required=True,
help="Password that Odoo will use to communicate with the "
"Asterisk Manager Interface. Refer to /etc/asterisk/manager.conf "
"on your Asterisk server.")
context = fields.Char(
string='Dialplan Context', required=True,
help="Asterisk dialplan context from which the calls will be "
"made. Refer to /etc/asterisk/extensions.conf on your Asterisk "
"server.")
wait_time = fields.Integer(
string='Wait Time', required=True, default=15,
help="Amount of time (in seconds) Asterisk will try to reach "
"the user's phone before hanging up.")
extension_priority = fields.Integer(
string='Extension Priority', required=True, default=1,
help="Priority of the extension in the Asterisk dialplan. Refer "
"to /etc/asterisk/extensions.conf on your Asterisk server.")
alert_info = fields.Char(
string='Alert-Info SIP Header',
help="Set Alert-Info header in SIP request to user's IP Phone "
"for the click2dial feature. If empty, the Alert-Info header "
"will not be added. You can use it to have a special ring tone "
"for click2dial (a silent one !) or to activate auto-answer "
"for example.")
company_id = fields.Many2one(
'res.company', string='Company',
default=lambda self: self.env['res.company']._company_default_get(
'asterisk.server'),
help="Company who uses the Asterisk server.")
@api.multi
@api.constrains(
'out_prefix', 'wait_time', 'extension_priority', 'port',
'context', 'alert_info', 'login', 'password')
def _check_validity(self):
for server in self:
out_prefix = ('Out prefix', server.out_prefix)
dialplan_context = ('Dialplan context', server.context)
alert_info = ('Alert-Info SIP header', server.alert_info)
login = ('AMI login', server.login)
password = ('AMI password', server.password)
if out_prefix[1] and not out_prefix[1].isdigit():
raise ValidationError(
_("Only use digits for the '%s' on the Asterisk server "
"'%s'" % (out_prefix[0], server.name)))
if server.wait_time < 1 or server.wait_time > 120:
raise ValidationError(
_("You should set a 'Wait time' value between 1 and 120 "
"seconds for the Asterisk server '%s'" % server.name))
if server.extension_priority < 1:
raise ValidationError(
_("The 'extension priority' must be a positive value for "
"the Asterisk server '%s'" % server.name))
if server.port > 65535 or server.port < 1:
raise ValidationError(
_("You should set a TCP port between 1 and 65535 for the "
"Asterisk server '%s'" % server.name))
for check_str in [dialplan_context, alert_info, login, password]:
if check_str[1]:
try:
check_str[1].encode('ascii')
except UnicodeEncodeError:
raise ValidationError(
_("The '%s' should only have ASCII caracters for "
"the Asterisk server '%s'"
% (check_str[0], server.name)))
@api.model
def _connect_to_asterisk(self):
'''
Open the connection to the Asterisk Manager
Returns an instance of the Asterisk Manager
'''
user = self.env.user
ast_server = user.get_asterisk_server_from_user()
# We check if the current user has a chan type
if not user.asterisk_chan_type:
raise UserError(
_('No channel type configured for the current user.'))
# We check if the current user has an internal number
if not user.resource:
raise UserError(
_('No resource name configured for the current user'))
_logger.debug(
"User's phone: %s/%s", user.asterisk_chan_type, user.resource)
_logger.debug(
"Asterisk server: %s:%d", ast_server.ip_address, ast_server.port)
# Connect to the Asterisk Manager Interface
try:
ast_manager = Manager.Manager(
(ast_server.ip_address, ast_server.port),
ast_server.login, ast_server.password)
except Exception, e:
_logger.error(
"Error in the request to the Asterisk Manager Interface %s",
ast_server.ip_address)
_logger.error("Here is the error message: %s", e)
raise UserError(
_("Problem in the request from Odoo to Asterisk. "
"Here is the error message: %s" % e))
return (user, ast_server, ast_manager)
@api.multi
def test_ami_connection(self):
self.ensure_one()
ast_manager = False
try:
ast_manager = Manager.Manager(
(self.ip_address, self.port),
self.login,
self.password)
except Exception, e:
raise UserError(
_("Connection Test Failed! The error message is: %s" % e))
finally:
if ast_manager:
ast_manager.Logoff()
raise UserError(_(
"Connection Test Successfull! Odoo can successfully login to "
"the Asterisk Manager Interface."))
@api.model
def _get_calling_number_from_channel(self, chan, user):
'''Method designed to be inherited to work with
very old or very new versions of Asterisk'''
sip_account = user.asterisk_chan_type + '/' + user.resource
internal_number = user.internal_number
# 4 = Ring
# 6 = Up
if (
chan.get('ChannelState') in ('4', '6') and (
chan.get('ConnectedLineNum') == internal_number or
chan.get('EffectiveConnectedLineNum') == internal_number or
sip_account in chan.get('BridgedChannel', ''))):
_logger.debug(
"Found a matching Event with channelstate = %s",
chan.get('ChannelState'))
return chan.get('CallerIDNum')
# Compatibility with Asterisk 1.4
if (
chan.get('State') == 'Up' and
sip_account in chan.get('Link', '')):
_logger.debug("Found a matching Event in 'Up' state")
return chan.get('CallerIDNum')
return False
@api.model
def _get_calling_number(self):
user, ast_server, ast_manager = self._connect_to_asterisk()
calling_party_number = False
try:
list_chan = ast_manager.Status()
# from pprint import pprint
# pprint(list_chan)
_logger.debug("Result of Status AMI request:")
_logger.debug(pformat(list_chan))
for chan in list_chan.values():
calling_party_number = self._get_calling_number_from_channel(
chan, user)
if calling_party_number:
break
except Exception, e:
_logger.error(
"Error in the Status request to Asterisk server %s",
ast_server.ip_address)
_logger.error(
"Here are the details of the error: '%s'", unicode(e))
raise UserError(
_("Can't get calling number from Asterisk.\nHere is the "
"error: '%s'" % unicode(e)))
finally:
ast_manager.Logoff()
_logger.debug("Calling party number: '%s'", calling_party_number)
return calling_party_number
@api.model
def get_record_from_my_channel(self):
calling_number = self.env['asterisk.server']._get_calling_number()
# calling_number = "0641981246"
if calling_number:
record = self.env['phone.common'].get_record_from_phone_number(
calling_number)
if record:
return record
else:
return calling_number
else:
return False
class ResUsers(models.Model):
_inherit = "res.users"
internal_number = fields.Char(
string='Internal Number', copy=False,
help="User's internal phone number.")
dial_suffix = fields.Char(
string='User-specific Dial Suffix',
help="User-specific dial suffix such as aa=2wb for SCCP auto answer.")
callerid = fields.Char(
string='Caller ID', copy=False,
help="Caller ID used for the calls initiated by this user.")
# You'd probably think: Asterisk should reuse the callerID of sip.conf!
# But it cannot, cf
# http://lists.digium.com/pipermail/asterisk-users/
# 2012-January/269787.html
cdraccount = fields.Char(
string='CDR Account',
help="Call Detail Record (CDR) account used for billing this user.")
asterisk_chan_type = fields.Selection([
('SIP', 'SIP'),
('IAX2', 'IAX2'),
('DAHDI', 'DAHDI'),
('Zap', 'Zap'),
('Skinny', 'Skinny'),
('MGCP', 'MGCP'),
('mISDN', 'mISDN'),
('H323', 'H323'),
('SCCP', 'SCCP'),
# Local works for click2dial, but it won't work in
# _get_calling_number() when trying to identify the
# channel of the user, so it's better not to propose it
# ('Local', 'Local'),
], string='Asterisk Channel Type', default='SIP',
help="Asterisk channel type, as used in the Asterisk dialplan. "
"If the user has a regular IP phone, the channel type is 'SIP'.")
resource = fields.Char(
string='Resource Name', copy=False,
help="Resource name for the channel type selected. For example, "
"if you use 'Dial(SIP/phone1)' in your Asterisk dialplan to ring "
"the SIP phone of this user, then the resource name for this user "
"is 'phone1'. For a SIP phone, the phone number is often used as "
"resource name, but not always.")
alert_info = fields.Char(
string='User-specific Alert-Info SIP Header',
help="Set a user-specific Alert-Info header in SIP request to "
"user's IP Phone for the click2dial feature. If empty, the "
"Alert-Info header will not be added. You can use it to have a "
"special ring tone for click2dial (a silent one !) or to "
"activate auto-answer for example.")
variable = fields.Char(
string='User-specific Variable',
help="Set a user-specific 'Variable' field in the Asterisk "
"Manager Interface 'originate' request for the click2dial "
"feature. If you want to have several variable headers, separate "
"them with '|'.")
asterisk_server_id = fields.Many2one(
'asterisk.server', string='Asterisk Server',
help="Asterisk server on which the user's phone is connected. "
"If you leave this field empty, it will use the first Asterisk "
"server of the user's company.")
@api.multi
@api.constrains('resource', 'internal_number', 'callerid')
def _check_validity(self):
for user in self:
strings_to_check = [
(_('Resource Name'), user.resource),
(_('Internal Number'), user.internal_number),
(_('Caller ID'), user.callerid),
]
for check_string in strings_to_check:
if check_string[1]:
try:
check_string[1].encode('ascii')
except UnicodeEncodeError:
raise ValidationError(_(
"The '%s' for the user '%s' should only have "
"ASCII caracters")
% (check_string[0], user.name))
@api.multi
def get_asterisk_server_from_user(self):
'''Returns an asterisk.server recordset'''
self.ensure_one()
# We check if the user has an Asterisk server configured
if self.asterisk_server_id:
ast_server = self.asterisk_server_id
else:
asterisk_servers = self.env['asterisk.server'].search(
[('company_id', '=', self.company_id.id)])
# If the user doesn't have an asterisk server,
# we take the first one of the user's company
if not asterisk_servers:
raise UserError(
_("No Asterisk server configured for the company '%s'.")
% self.company_id.name)
else:
ast_server = asterisk_servers[0]
return ast_server
class PhoneCommon(models.AbstractModel):
_inherit = 'phone.common'
@api.model
def click2dial(self, erp_number):
res = super(PhoneCommon, self).click2dial(erp_number)
if not erp_number:
raise UserError(_('Missing phone number'))
user, ast_server, ast_manager = \
self.env['asterisk.server']._connect_to_asterisk()
ast_number = self.convert_to_dial_number(erp_number)
# Add 'out prefix'
if ast_server.out_prefix:
_logger.debug('Out prefix = %s', ast_server.out_prefix)
ast_number = '%s%s' % (ast_server.out_prefix, ast_number)
_logger.debug('Number to be sent to Asterisk = %s', ast_number)
# The user should have a CallerID
if not user.callerid:
raise UserError(_('No callerID configured for the current user'))
variable = []
if user.asterisk_chan_type == 'SIP':
# We can only have one alert-info header in a SIP request
if user.alert_info:
variable.append(
'SIPAddHeader=Alert-Info: %s' % user.alert_info)
elif ast_server.alert_info:
variable.append(
'SIPAddHeader=Alert-Info: %s' % ast_server.alert_info)
if user.variable:
for user_variable in user.variable.split('|'):
variable.append(user_variable.strip())
channel = '%s/%s' % (user.asterisk_chan_type, user.resource)
if user.dial_suffix:
channel += '/%s' % user.dial_suffix
try:
ast_manager.Originate(
channel,
context=ast_server.context,
extension=ast_number,
priority=unicode(ast_server.extension_priority),
timeout=unicode(ast_server.wait_time * 1000),
caller_id=user.callerid,
account=user.cdraccount,
variable=variable)
except Exception, e:
_logger.error(
"Error in the Originate request to Asterisk server %s",
ast_server.ip_address)
_logger.error(
"Here are the details of the error: '%s'", unicode(e))
raise UserError(
_("Click to dial with Asterisk failed.\nHere is the error: "
"'%s'")
% unicode(e))
finally:
ast_manager.Logoff()
res['dialed_number'] = ast_number
return res