test/crm
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: added telephony medium field to capture Exotel, Twilio or Manual call log

Browse Source
This commit is contained in:
Shariq Ansari 2025-01-11 16:32:18 +05:30
parent 3991c819ba
commit 5c2aa522fc
4 changed files with 176 additions and 146 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
crm/fcrm/doctype/crm_call_log/crm_call_log.json
View File

@ -9,6 +9,8 @@
"editable_grid": 1, "editable_grid": 1,
"engine": "InnoDB", "engine": "InnoDB",
"field_order": [ "field_order": [
"telephony_medium",
"section_break_gyqe",
"id", "id",
"from", "from",
"status", "status",
@ -24,7 +26,9 @@
"caller", "caller",
"recording_url", "recording_url",
"end_time", "end_time",
"note" "note",
"section_break_kebz",
"links"
], ],
"fields": [ "fields": [
{ {
@ -75,6 +79,7 @@
"label": "To" "label": "To"
}, },
{ {
"description": "Call duration in seconds",
"fieldname": "duration", "fieldname": "duration",
"fieldtype": "Duration", "fieldtype": "Duration",
"in_list_view": 1, "in_list_view": 1,
@ -123,11 +128,31 @@
"fieldtype": "Dynamic Link", "fieldtype": "Dynamic Link",
"label": "Reference Name", "label": "Reference Name",
"options": "reference_doctype" "options": "reference_doctype"
},
{
"fieldname": "section_break_kebz",
"fieldtype": "Section Break"
},
{
"fieldname": "links",
"fieldtype": "Table",
"label": "Links",
"options": "Dynamic Link"
},
{
"fieldname": "telephony_medium",
"fieldtype": "Select",
"label": "Telephony Medium",
"options": "\nManual\nTwilio\nExotel"
},
{
"fieldname": "section_break_gyqe",
"fieldtype": "Section Break"
} }
], ],
"index_web_pages_for_search": 1, "index_web_pages_for_search": 1,
"links": [], "links": [],
"modified": "2024-03-16 13:23:09.201843", "modified": "2025-01-11 16:27:56.992950",
"modified_by": "Administrator", "modified_by": "Administrator",
"module": "FCRM", "module": "FCRM",
"name": "CRM Call Log", "name": "CRM Call Log",

100
crm/integrations/twilio/api.py
View File

@ -1,44 +1,45 @@
from werkzeug.wrappers import Response
import json import json
import frappe import frappe
from frappe import _ from frappe import _
from .twilio_handler import Twilio, IncomingCall, TwilioCallDetails from werkzeug.wrappers import Response
from .twilio_handler import IncomingCall, Twilio, TwilioCallDetails
from .utils import parse_mobile_no from .utils import parse_mobile_no
@frappe.whitelist() @frappe.whitelist()
def is_enabled(): def is_enabled():
return frappe.db.get_single_value("Twilio Settings", "enabled") return frappe.db.get_single_value("Twilio Settings", "enabled")
@frappe.whitelist() @frappe.whitelist()
def generate_access_token(): def generate_access_token():
"""Returns access token that is required to authenticate Twilio Client SDK. """Returns access token that is required to authenticate Twilio Client SDK."""
"""
twilio = Twilio.connect() twilio = Twilio.connect()
if not twilio: if not twilio:
return {} return {}
from_number = frappe.db.get_value('Twilio Agents', frappe.session.user, 'twilio_number') from_number = frappe.db.get_value("Twilio Agents", frappe.session.user, "twilio_number")
if not from_number: if not from_number:
return { return {
"ok": False, "ok": False,
"error": "caller_phone_identity_missing", "error": "caller_phone_identity_missing",
"detail": "Phone number is not mapped to the caller" "detail": "Phone number is not mapped to the caller",
} }
token=twilio.generate_voice_access_token(identity=frappe.session.user) token = twilio.generate_voice_access_token(identity=frappe.session.user)
return { return {"token": frappe.safe_decode(token)}
'token': frappe.safe_decode(token)
}
@frappe.whitelist(allow_guest=True) @frappe.whitelist(allow_guest=True)
def voice(**kwargs): def voice(**kwargs):
"""This is a webhook called by twilio to get instructions when the voice call request comes to twilio server. """This is a webhook called by twilio to get instructions when the voice call request comes to twilio server."""
"""
def _get_caller_number(caller): def _get_caller_number(caller):
identity = caller.replace('client:', '').strip() identity = caller.replace("client:", "").strip()
user = Twilio.emailid_from_identity(identity) user = Twilio.emailid_from_identity(identity)
return frappe.db.get_value('Twilio Agents', user, 'twilio_number') return frappe.db.get_value("Twilio Agents", user, "twilio_number")
args = frappe._dict(kwargs) args = frappe._dict(kwargs)
twilio = Twilio.connect() twilio = Twilio.connect()
@ -54,7 +55,8 @@ def voice(**kwargs):
call_details = TwilioCallDetails(args, call_from=from_number) call_details = TwilioCallDetails(args, call_from=from_number)
create_call_log(call_details) create_call_log(call_details)
return Response(resp.to_xml(), mimetype='text/xml') return Response(resp.to_xml(), mimetype="text/xml")
@frappe.whitelist(allow_guest=True) @frappe.whitelist(allow_guest=True)
def twilio_incoming_call_handler(**kwargs): def twilio_incoming_call_handler(**kwargs):
@ -63,23 +65,24 @@ def twilio_incoming_call_handler(**kwargs):
create_call_log(call_details) create_call_log(call_details)
resp = IncomingCall(args.From, args.To).process() resp = IncomingCall(args.From, args.To).process()
return Response(resp.to_xml(), mimetype='text/xml') return Response(resp.to_xml(), mimetype="text/xml")
def create_call_log(call_details: TwilioCallDetails): def create_call_log(call_details: TwilioCallDetails):
call_log = frappe.get_doc({**call_details.to_dict(), call_log = frappe.get_doc(
'doctype': 'CRM Call Log', {**call_details.to_dict(), "doctype": "CRM Call Log", "telephony_medium": "Twilio"}
'medium': 'Twilio' )
})
call_log.reference_docname, call_log.reference_doctype = get_lead_or_deal_from_number(call_log) call_log.reference_docname, call_log.reference_doctype = get_lead_or_deal_from_number(call_log)
call_log.flags.ignore_permissions = True call_log.flags.ignore_permissions = True
call_log.save() call_log.save()
frappe.db.commit() frappe.db.commit()
def update_call_log(call_sid, status=None): def update_call_log(call_sid, status=None):
"""Update call log status. """Update call log status."""
"""
twilio = Twilio.connect() twilio = Twilio.connect()
if not (twilio and frappe.db.exists("CRM Call Log", call_sid)): return if not (twilio and frappe.db.exists("CRM Call Log", call_sid)):
return
call_details = twilio.get_call_info(call_sid) call_details = twilio.get_call_info(call_sid)
call_log = frappe.get_doc("CRM Call Log", call_sid) call_log = frappe.get_doc("CRM Call Log", call_sid)
@ -94,6 +97,7 @@ def update_call_log(call_sid, status=None):
call_log.save() call_log.save()
frappe.db.commit() frappe.db.commit()
@frappe.whitelist(allow_guest=True) @frappe.whitelist(allow_guest=True)
def update_recording_info(**kwargs): def update_recording_info(**kwargs):
try: try:
@ -105,6 +109,7 @@ def update_recording_info(**kwargs):
except: except:
frappe.log_error(title=_("Failed to capture Twilio recording")) frappe.log_error(title=_("Failed to capture Twilio recording"))
@frappe.whitelist(allow_guest=True) @frappe.whitelist(allow_guest=True)
def update_call_status_info(**kwargs): def update_call_status_info(**kwargs):
try: try:
@ -113,53 +118,54 @@ def update_call_status_info(**kwargs):
update_call_log(parent_call_sid, status=args.CallStatus) update_call_log(parent_call_sid, status=args.CallStatus)
call_info = { call_info = {
'ParentCallSid': args.ParentCallSid, "ParentCallSid": args.ParentCallSid,
'CallSid': args.CallSid, "CallSid": args.CallSid,
'CallStatus': args.CallStatus, "CallStatus": args.CallStatus,
'CallDuration': args.CallDuration, "CallDuration": args.CallDuration,
'From': args.From, "From": args.From,
'To': args.To, "To": args.To,
} }
client = Twilio.get_twilio_client() client = Twilio.get_twilio_client()
client.calls(args.ParentCallSid).user_defined_messages.create( client.calls(args.ParentCallSid).user_defined_messages.create(content=json.dumps(call_info))
content=json.dumps(call_info)
)
except: except:
frappe.log_error(title=_("Failed to update Twilio call status")) frappe.log_error(title=_("Failed to update Twilio call status"))
def get_datetime_from_timestamp(timestamp): def get_datetime_from_timestamp(timestamp):
from datetime import datetime from datetime import datetime
from zoneinfo import ZoneInfo from zoneinfo import ZoneInfo
if not timestamp: return None if not timestamp:
return None
datetime_utc_tz_str = timestamp.strftime('%Y-%m-%d %H:%M:%S%z') datetime_utc_tz_str = timestamp.strftime("%Y-%m-%d %H:%M:%S%z")
datetime_utc_tz = datetime.strptime(datetime_utc_tz_str, '%Y-%m-%d %H:%M:%S%z') datetime_utc_tz = datetime.strptime(datetime_utc_tz_str, "%Y-%m-%d %H:%M:%S%z")
system_timezone = frappe.utils.get_system_timezone() system_timezone = frappe.utils.get_system_timezone()
converted_datetime = datetime_utc_tz.astimezone(ZoneInfo(system_timezone)) converted_datetime = datetime_utc_tz.astimezone(ZoneInfo(system_timezone))
return frappe.utils.format_datetime(converted_datetime, 'yyyy-MM-dd HH:mm:ss') return frappe.utils.format_datetime(converted_datetime, "yyyy-MM-dd HH:mm:ss")
@frappe.whitelist() @frappe.whitelist()
def add_note_to_call_log(call_sid, note): def add_note_to_call_log(call_sid, note):
"""Add note to call log. based on child call sid. """Add note to call log. based on child call sid."""
"""
twilio = Twilio.connect() twilio = Twilio.connect()
if not twilio: return if not twilio:
return
call_details = twilio.get_call_info(call_sid) call_details = twilio.get_call_info(call_sid)
sid = call_sid if call_details.direction == 'inbound' else call_details.parent_call_sid sid = call_sid if call_details.direction == "inbound" else call_details.parent_call_sid
frappe.db.set_value("CRM Call Log", sid, "note", note) frappe.db.set_value("CRM Call Log", sid, "note", note)
frappe.db.commit() frappe.db.commit()
def get_lead_or_deal_from_number(call):
"""Get lead/deal from the given number.
"""
def find_record(doctype, mobile_no, where=''): def get_lead_or_deal_from_number(call):
"""Get lead/deal from the given number."""
def find_record(doctype, mobile_no, where=""):
mobile_no = parse_mobile_no(mobile_no) mobile_no = parse_mobile_no(mobile_no)
query = f""" query = f"""
SELECT name, mobile_no SELECT name, mobile_no
FROM `tab{doctype}` FROM `tab{doctype}`
@ -170,12 +176,12 @@ def get_lead_or_deal_from_number(call):
return data[0].name if data else None return data[0].name if data else None
doctype = "CRM Deal" doctype = "CRM Deal"
number = call.get('to') if call.type == 'Outgoing' else call.get('from') number = call.get("to") if call.type == "Outgoing" else call.get("from")
doc = find_record(doctype, number) or None doc = find_record(doctype, number) or None
if not doc: if not doc:
doctype = "CRM Lead" doctype = "CRM Lead"
doc = find_record(doctype, number, 'AND converted is not True') doc = find_record(doctype, number, "AND converted is not True")
if not doc: if not doc:
doc = find_record(doctype, number) doc = find_record(doctype, number)

182
crm/integrations/twilio/twilio_handler.py
View File

@ -1,16 +1,17 @@
from twilio.rest import Client as TwilioClient
from twilio.jwt.access_token import AccessToken
from twilio.jwt.access_token.grants import VoiceGrant
from twilio.twiml.voice_response import VoiceResponse, Dial
from .utils import get_public_url, merge_dicts
import frappe import frappe
from frappe import _ from frappe import _
from frappe.utils.password import get_decrypted_password from frappe.utils.password import get_decrypted_password
from twilio.jwt.access_token import AccessToken
from twilio.jwt.access_token.grants import VoiceGrant
from twilio.rest import Client as TwilioClient
from twilio.twiml.voice_response import Dial, VoiceResponse
from .utils import get_public_url, merge_dicts
class Twilio: class Twilio:
"""Twilio connector over TwilioClient. """Twilio connector over TwilioClient."""
"""
def __init__(self, settings): def __init__(self, settings):
""" """
:param settings: `Twilio Settings` doctype :param settings: `Twilio Settings` doctype
@ -24,22 +25,19 @@ class Twilio:
@classmethod @classmethod
def connect(self): def connect(self):
"""Make a twilio connection. """Make a twilio connection."""
"""
settings = frappe.get_doc("Twilio Settings") settings = frappe.get_doc("Twilio Settings")
if not (settings and settings.enabled): if not (settings and settings.enabled):
return return
return Twilio(settings=settings) return Twilio(settings=settings)
def get_phone_numbers(self): def get_phone_numbers(self):
"""Get account's twilio phone numbers. """Get account's twilio phone numbers."""
"""
numbers = self.twilio_client.incoming_phone_numbers.list() numbers = self.twilio_client.incoming_phone_numbers.list()
return [n.phone_number for n in numbers] return [n.phone_number for n in numbers]
def generate_voice_access_token(self, identity: str, ttl=60*60): def generate_voice_access_token(self, identity: str, ttl=60 * 60):
"""Generates a token required to make voice calls from the browser. """Generates a token required to make voice calls from the browser."""
"""
# identity is used by twilio to identify the user uniqueness at browser(or any endpoints). # identity is used by twilio to identify the user uniqueness at browser(or any endpoints).
identity = self.safe_identity(identity) identity = self.safe_identity(identity)
@ -49,7 +47,7 @@ class Twilio:
# Create a Voice grant and add to token # Create a Voice grant and add to token
voice_grant = VoiceGrant( voice_grant = VoiceGrant(
outgoing_application_sid=self.application_sid, outgoing_application_sid=self.application_sid,
incoming_allow=True, # Allow incoming calls incoming_allow=True, # Allow incoming calls
) )
token.add_grant(voice_grant) token.add_grant(voice_grant)
return token.to_jwt() return token.to_jwt()
@ -60,14 +58,13 @@ class Twilio:
Twilio Client JS fails to make a call connection if identity has special characters like @, [, / etc) Twilio Client JS fails to make a call connection if identity has special characters like @, [, / etc)
https://www.twilio.com/docs/voice/client/errors (#31105) https://www.twilio.com/docs/voice/client/errors (#31105)
""" """
return identity.replace('@', '(at)') return identity.replace("@", "(at)")
@classmethod @classmethod
def emailid_from_identity(cls, identity: str): def emailid_from_identity(cls, identity: str):
"""Convert safe identity string into emailID. """Convert safe identity string into emailID."""
""" return identity.replace("(at)", "@")
return identity.replace('(at)', '@')
def get_recording_status_callback_url(self): def get_recording_status_callback_url(self):
url_path = "/api/method/crm.integrations.twilio.api.update_recording_info" url_path = "/api/method/crm.integrations.twilio.api.update_recording_info"
return get_public_url(url_path) return get_public_url(url_path)
@ -77,20 +74,19 @@ class Twilio:
return get_public_url(url_path) return get_public_url(url_path)
def generate_twilio_dial_response(self, from_number: str, to_number: str): def generate_twilio_dial_response(self, from_number: str, to_number: str):
"""Generates voice call instructions to forward the call to agents Phone. """Generates voice call instructions to forward the call to agents Phone."""
"""
resp = VoiceResponse() resp = VoiceResponse()
dial = Dial( dial = Dial(
caller_id=from_number, caller_id=from_number,
record=self.settings.record_calls, record=self.settings.record_calls,
recording_status_callback=self.get_recording_status_callback_url(), recording_status_callback=self.get_recording_status_callback_url(),
recording_status_callback_event='completed' recording_status_callback_event="completed",
) )
dial.number( dial.number(
to_number, to_number,
status_callback_event='initiated ringing answered completed', status_callback_event="initiated ringing answered completed",
status_callback=self.get_update_call_status_callback_url(), status_callback=self.get_update_call_status_callback_url(),
status_callback_method='POST' status_callback_method="POST",
) )
resp.append(dial) resp.append(dial)
return resp return resp
@ -98,21 +94,20 @@ class Twilio:
def get_call_info(self, call_sid): def get_call_info(self, call_sid):
return self.twilio_client.calls(call_sid).fetch() return self.twilio_client.calls(call_sid).fetch()
def generate_twilio_client_response(self, client, ring_tone='at'): def generate_twilio_client_response(self, client, ring_tone="at"):
"""Generates voice call instructions to forward the call to agents computer. """Generates voice call instructions to forward the call to agents computer."""
"""
resp = VoiceResponse() resp = VoiceResponse()
dial = Dial( dial = Dial(
ring_tone=ring_tone, ring_tone=ring_tone,
record=self.settings.record_calls, record=self.settings.record_calls,
recording_status_callback=self.get_recording_status_callback_url(), recording_status_callback=self.get_recording_status_callback_url(),
recording_status_callback_event='completed' recording_status_callback_event="completed",
) )
dial.client( dial.client(
client, client,
status_callback_event='initiated ringing answered completed', status_callback_event="initiated ringing answered completed",
status_callback=self.get_update_call_status_callback_url(), status_callback=self.get_update_call_status_callback_url(),
status_callback_method='POST' status_callback_method="POST",
) )
resp.append(dial) resp.append(dial)
return resp return resp
@ -122,12 +117,13 @@ class Twilio:
twilio_settings = frappe.get_doc("Twilio Settings") twilio_settings = frappe.get_doc("Twilio Settings")
if not twilio_settings.enabled: if not twilio_settings.enabled:
frappe.throw(_("Please enable twilio settings before making a call.")) frappe.throw(_("Please enable twilio settings before making a call."))
auth_token = get_decrypted_password("Twilio Settings", "Twilio Settings", 'auth_token') auth_token = get_decrypted_password("Twilio Settings", "Twilio Settings", "auth_token")
client = TwilioClient(twilio_settings.account_sid, auth_token) client = TwilioClient(twilio_settings.account_sid, auth_token)
return client return client
class IncomingCall: class IncomingCall:
def __init__(self, from_number, to_number, meta=None): def __init__(self, from_number, to_number, meta=None):
self.from_number = from_number self.from_number = from_number
@ -145,17 +141,18 @@ class IncomingCall:
if not attender: if not attender:
resp = VoiceResponse() resp = VoiceResponse()
resp.say(_('Agent is unavailable to take the call, please call after some time.')) resp.say(_("Agent is unavailable to take the call, please call after some time."))
return resp return resp
if attender['call_receiving_device'] == 'Phone': if attender["call_receiving_device"] == "Phone":
return twilio.generate_twilio_dial_response(self.from_number, attender['mobile_no']) return twilio.generate_twilio_dial_response(self.from_number, attender["mobile_no"])
else: else:
return twilio.generate_twilio_client_response(twilio.safe_identity(attender['name'])) return twilio.generate_twilio_client_response(twilio.safe_identity(attender["name"]))
def get_twilio_number_owners(phone_number): def get_twilio_number_owners(phone_number):
"""Get list of users who is using the phone_number. """Get list of users who is using the phone_number.
>>> get_twilio_number_owners('+11234567890') >>> get_twilio_number_owners("+11234567890")
{ {
'owner1': {'name': '..', 'mobile_no': '..', 'call_receiving_device': '...'}, 'owner1': {'name': '..', 'mobile_no': '..', 'call_receiving_device': '...'},
'owner2': {....} 'owner2': {....}
@ -163,105 +160,106 @@ def get_twilio_number_owners(phone_number):
""" """
# remove special characters from phone number and get only digits also remove white spaces # remove special characters from phone number and get only digits also remove white spaces
# keep + sign in the number at start of the number # keep + sign in the number at start of the number
phone_number = ''.join([c for c in phone_number if c.isdigit() or c == '+']) phone_number = "".join([c for c in phone_number if c.isdigit() or c == "+"])
user_voice_settings = frappe.get_all( user_voice_settings = frappe.get_all(
'Twilio Agents', "Twilio Agents", filters={"twilio_number": phone_number}, fields=["name", "call_receiving_device"]
filters={'twilio_number': phone_number},
fields=["name", "call_receiving_device"]
) )
user_wise_voice_settings = {user['name']: user for user in user_voice_settings} user_wise_voice_settings = {user["name"]: user for user in user_voice_settings}
user_general_settings = frappe.get_all( user_general_settings = frappe.get_all(
'User', "User", filters=[["name", "IN", user_wise_voice_settings.keys()]], fields=["name", "mobile_no"]
filters = [['name', 'IN', user_wise_voice_settings.keys()]],
fields = ['name', 'mobile_no']
) )
user_wise_general_settings = {user['name']: user for user in user_general_settings} user_wise_general_settings = {user["name"]: user for user in user_general_settings}
return merge_dicts(user_wise_general_settings, user_wise_voice_settings) return merge_dicts(user_wise_general_settings, user_wise_voice_settings)
def get_active_loggedin_users(users): def get_active_loggedin_users(users):
"""Filter the current loggedin users from the given users list """Filter the current loggedin users from the given users list"""
""" rows = frappe.db.sql(
rows = frappe.db.sql(""" """
SELECT `user` SELECT `user`
FROM `tabSessions` FROM `tabSessions`
WHERE `user` IN %(users)s WHERE `user` IN %(users)s
""", {'users': users}) """,
{"users": users},
)
return [row[0] for row in set(rows)] return [row[0] for row in set(rows)]
def get_the_call_attender(owners, caller=None): def get_the_call_attender(owners, caller=None):
"""Get attender details from list of owners """Get attender details from list of owners"""
""" if not owners:
if not owners: return return
current_loggedin_users = get_active_loggedin_users(list(owners.keys())) current_loggedin_users = get_active_loggedin_users(list(owners.keys()))
if len(current_loggedin_users) > 1 and caller: if len(current_loggedin_users) > 1 and caller:
deal_owner = frappe.db.get_value('CRM Deal', {'mobile_no': caller}, 'deal_owner') deal_owner = frappe.db.get_value("CRM Deal", {"mobile_no": caller}, "deal_owner")
if not deal_owner: if not deal_owner:
deal_owner = frappe.db.get_value('CRM Lead', {'mobile_no': caller, 'converted': False}, 'lead_owner') deal_owner = frappe.db.get_value(
"CRM Lead", {"mobile_no": caller, "converted": False}, "lead_owner"
)
for user in current_loggedin_users: for user in current_loggedin_users:
if user == deal_owner: if user == deal_owner:
current_loggedin_users = [user] current_loggedin_users = [user]
for name, details in owners.items(): for name, details in owners.items():
if ((details['call_receiving_device'] == 'Phone' and details['mobile_no']) or if (details["call_receiving_device"] == "Phone" and details["mobile_no"]) or (
(details['call_receiving_device'] == 'Computer' and name in current_loggedin_users)): details["call_receiving_device"] == "Computer" and name in current_loggedin_users
):
return details return details
class TwilioCallDetails: class TwilioCallDetails:
def __init__(self, call_info, call_from = None, call_to = None): def __init__(self, call_info, call_from=None, call_to=None):
self.call_info = call_info self.call_info = call_info
self.account_sid = call_info.get('AccountSid') self.account_sid = call_info.get("AccountSid")
self.application_sid = call_info.get('ApplicationSid') self.application_sid = call_info.get("ApplicationSid")
self.call_sid = call_info.get('CallSid') self.call_sid = call_info.get("CallSid")
self.call_status = self.get_call_status(call_info.get('CallStatus')) self.call_status = self.get_call_status(call_info.get("CallStatus"))
self._call_from = call_from or call_info.get('From') self._call_from = call_from or call_info.get("From")
self._call_to = call_to or call_info.get('To') self._call_to = call_to or call_info.get("To")
def get_direction(self): def get_direction(self):
if self.call_info.get('Caller').lower().startswith('client'): if self.call_info.get("Caller").lower().startswith("client"):
return 'Outgoing' return "Outgoing"
return 'Incoming' return "Incoming"
def get_from_number(self): def get_from_number(self):
return self._call_from or self.call_info.get('From') return self._call_from or self.call_info.get("From")
def get_to_number(self): def get_to_number(self):
return self._call_to or self.call_info.get('To') return self._call_to or self.call_info.get("To")
@classmethod @classmethod
def get_call_status(cls, twilio_status): def get_call_status(cls, twilio_status):
"""Convert Twilio given status into system status. """Convert Twilio given status into system status."""
""" twilio_status = twilio_status or ""
twilio_status = twilio_status or '' return " ".join(twilio_status.split("-")).title()
return ' '.join(twilio_status.split('-')).title()
def to_dict(self): def to_dict(self):
"""Convert call details into dict. """Convert call details into dict."""
"""
direction = self.get_direction() direction = self.get_direction()
from_number = self.get_from_number() from_number = self.get_from_number()
to_number = self.get_to_number() to_number = self.get_to_number()
caller = '' caller = ""
receiver = '' receiver = ""
if direction == 'Outgoing': if direction == "Outgoing":
caller = self.call_info.get('Caller') caller = self.call_info.get("Caller")
identity = caller.replace('client:', '').strip() identity = caller.replace("client:", "").strip()
caller = Twilio.emailid_from_identity(identity) if identity else '' caller = Twilio.emailid_from_identity(identity) if identity else ""
else: else:
owners = get_twilio_number_owners(to_number) owners = get_twilio_number_owners(to_number)
attender = get_the_call_attender(owners, from_number) attender = get_the_call_attender(owners, from_number)
receiver = attender['name'] if attender else '' receiver = attender["name"] if attender else ""
return { return {
'type': direction, "type": direction,
'status': self.call_status, "status": self.call_status,
'id': self.call_sid, "id": self.call_sid,
'from': from_number, "from": from_number,
'to': to_number, "to": to_number,
'receiver': receiver, "receiver": receiver,
'caller': caller, "caller": caller,
} }

11
crm/integrations/twilio/utils.py
View File

@ -1,7 +1,7 @@
from frappe.utils import get_url from frappe.utils import get_url
def get_public_url(path: str=None): def get_public_url(path: str | None = None):
return get_url().split(":8", 1)[0] + path return get_url().split(":8", 1)[0] + path
@ -13,11 +13,12 @@ def merge_dicts(d1: dict, d2: dict):
) )
... {'name1': {'age': 20, 'phone': '+xxx'}, 'name2': {'age': 30, 'phone': '+yyy'}} ... {'name1': {'age': 20, 'phone': '+xxx'}, 'name2': {'age': 30, 'phone': '+yyy'}}
""" """
return {k:{**v, **d2.get(k, {})} for k, v in d1.items()} return {k: {**v, **d2.get(k, {})} for k, v in d1.items()}
def parse_mobile_no(mobile_no: str): def parse_mobile_no(mobile_no: str):
"""Parse mobile number to remove spaces, brackets, etc. """Parse mobile number to remove spaces, brackets, etc.
>>> parse_mobile_no('+91 (766) 667 6666') >>> parse_mobile_no("+91 (766) 667 6666")
... '+917666676666' ... "+917666676666"
""" """
return ''.join([c for c in mobile_no if c.isdigit() or c == '+']) return "".join([c for c in mobile_no if c.isdigit() or c == "+"])