Skip to content

Instantly share code, notes, and snippets.

@igmar

igmar/extensions.lua

Created November 13, 2012 15:55
Show Gist options
  • Save igmar/4066527 to your computer and use it in GitHub Desktop.
Save igmar/4066527 to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. igmar created this gist Nov 13, 2012.
    378 changes: 378 additions & 0 deletions extensions.lua
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,378 @@
    require("lsqlite3")

    -- Igmar: Wanneer closen we dat DB object eigenlijk ?
    db = sqlite3.open('/etc/asterisk/users.sqlite')

    --CONSOLE = "Console/dsp" -- Console interface for demo
    --CONSOLE = "DAHDI/1"
    --CONSOLE = "Phone/phone0"

    TRUNK = "DAHDI/G1"
    TRUNKMSD = 1

    APIKEY = "oursecretapikey"

    -- The default comment is here

    function e_drop_call(context, entension)
    app.hangup(29)
    end

    function e_hotdesk_login(context, extension)
    -- Find out the peername
    name = channel.CALLERID("name"):get()
    peername = channel.CHANNEL("peername"):get()
    extension = string.sub(extension, 2)

    status = get_extension_status(extension)
    if status == false then
    app.Playback("acces-denied")
    app.Verbose(1, "Extension " .. extension .. " disabled")
    app.Hangup(29)
    end

    -- Already logged in ?
    location = extension_to_location(extension)
    if location ~= nil then
    app.Verbose(1, "Extension " .. extension .. " already logged in")
    app.Playback("agent-alreadyon")
    app.Hangup(29)
    return
    end

    app.Playback("please-enter-the&access-code")
    app.Verbose(1, "Logging in at the PBX context : " .. context .. " extension " .. extension .. " peername " .. peername)

    userstatus = "FALSE"
    userpin = get_extension_pin(extension)

    -- Ask the PIN
    app.Read("pin", "", 4)
    pin = channel["pin"]:get()

    if pin == userpin then
    app.Verbose(1, "PIN OK")
    -- Update the location field
    res, error = db:exec(string.format("UPDATE ast_users SET location = '%s' WHERE extension = %s", peername, extension))
    if res > 0 then
    app.Verbose(1, string.format("UPDATE on ast_users failed : %d", res))
    app.Hangup(29)
    return
    end
    app.Playback("agent-loginok")
    else
    app.Playback("acces-denied")
    app.Hangup(29)
    return
    end

    -- Put the user in the queue
    queue = get_extension_queue(extension)
    if queue ~= nil then
    app.Verbose(1, string.format("Adding member to queue : %s", queue))
    app.AddQueueMember(queue)
    end
    end

    function e_hotdesk_logout(context, extension)
    name = channel.CALLERID("name"):get()
    peername = channel.CHANNEL("peername"):get()

    extension = location_to_extension(peername)
    if extension == nil then
    app.Verbose(1, "Location " .. peername .. " is not logged in")
    app.Hangup(29)
    return
    end

    app.Verbose(1, "Logging out at the PBX context : " .. context .. " extension " .. extension .. " peername " .. peername)

    res, error = db:exec(string.format("UPDATE ast_users SET location = null WHERE extension = %s", extension))
    if res > 0 then
    app.Verbose(1, string.format("UPDATE on ast_users failed"))
    app.Hangup(29)
    return
    end

    app.Playback("agent-loggedoff")
    queue = get_extension_queue(extension)
    app.Verbose(1, string.format("Removing member from queue : %s", queue))
    app.RemoveQueueMember(queue)
    app.Answer()
    end

    function e_outgoing_sip(context, extension)
    name = channel.CALLERID("name"):get()
    peername = channel.CHANNEL("peername"):get()

    app.Verbose("Called outgoing " .. extension .. " in context " .. context)

    -- Already logged in ?
    my_extension = location_to_extension(peername)
    if my_extension == nil then
    app.Verbose(1, "Location " .. peername .. " not logged in")
    app.Playback("acces-denied")
    app.Hangup(29)
    end

    app.Dial(TRUNK .. "/" .. extension)
    end


    function e_outgoing_emergency(context, extension)
    name = channel.CALLERID("name"):get()
    peername = channel.CHANNEL("peername"):get()

    app.Verbose("Emergency dial " .. extension .. " in context " .. context)

    app.Dial(TRUNK .. "/112")
    end

    function e_internal_sip(context, extension)
    app.Verbose("Called internal extension " .. extension .. " in context " .. context)
    -- Is this device logged on ?
    name = channel.CALLERID("name"):get()
    peername = channel.CHANNEL("peername"):get()
    callee = channel.CALLERID("num"):get()
    my_extension = nil

    app.Verbose(1, string.format("Device name : %s, callee %s", name, callee))
    callee_extension = location_to_extension(callee)
    callee_name = location_to_name(callee)

    -- peername is nil on a direct call transfer. Handle this
    if peername ~= nil then
    -- What is my own extension number ? Find it.
    my_extension = location_to_extension(peername)
    if my_extension == nil then
    app.Verbose(1, "Location " .. peername .. " not logged in")
    app.Playback("acces-denied")
    app.Hangup(26)
    end

    if my_extension == extension then
    app.Verbose(1, "We called outselves. Bad")
    app.Playback("all-your-base")
    app.Hangup(26)
    end
    end

    dialed_location = extension_to_location(extension)
    if dialed_location == nil then
    app.Verbose(1, "Extension " .. extension .. " not active")
    app.Playback("extension&T-is-not-available")
    app.Hangup(20)
    return
    end

    if callee_extension ~= nil then
    app.Verbose(1, string.format("Callee extension : %s", callee_extension));
    channel.CALLERID("all"):set(string.format("%s <%s>", callee_name, callee_extension))
    end

    app.Dial("SIP/" .. dialed_location)
    end

    function e_dial_custnr(context, extension)
    peername = channel.CHANNEL("peername"):get()
    app.Verbose(1, string.format("Peername : %s", peername))

    my_extension = location_to_extension(peername)
    if my_extension == nil then
    app.Verbose(1, "Location " .. peername .. " not logged in")
    app.Playback("acces-denied")
    app.Hangup(29)
    end

    local https = require('ssl.https')

    -- Ditch the initial *
    extension = string.sub(extension, 2)
    app.Verbose(1, "Entering dial_custnr_phone")
    -- Build the URL
    url = string.format("https://X.X.X.X/rest/?apikey=%s&cmd=custinfo&arg1=%s", APIKEY, extension)
    app.Verbose(1, string.format("Requesting %s", url))
    r, c, h, s = https.request(url)
    for s1, s2 in string.gmatch(r, "([%w%s]+):([%w%s]+)") do
    name = s1
    number = s2
    end
    app.Verbose(1, string.format("Data for customer %s : %s %s", extension, name, number))
    app.Dial(TRUNK .. "/" .. number)

    end

    function e_dial_employee(context, extension)
    peername = channel.CHANNEL("peername"):get()
    app.Verbose(1, string.format("Peername : %s", peername))

    my_extension = location_to_extension(peername)
    if my_extension == nil then
    app.Verbose(1, "Location " .. peername .. " not logged in")
    app.Playback("acces-denied")
    app.Hangup(29)
    end

    local https = require('ssl.https')
    extension = string.sub(extension, 3)
    app.Verbose(1, string.format("Dialing employee %s", extension))
    url = string.format("https://X.X.X.X/rest/?apikey=%s&cmd=emplinfo&arg1=%s", APIKEY, extension)
    app.Verbose(1, string.format("Requesting %s", url))
    r, c, h, s = https.request(url)
    for s1, s2 in string.gmatch(r, "([%w%s]+):([%w%s]+)") do
    name = s1
    number = s2
    end
    app.Verbose(1, string.format("Data for employee %s : %s %s", extension, name, number))
    app.Dial(TRUNK .. "/" .. number)
    end

    function e_incoming(context, extension)
    -- Check if we are open
    if in_office_hours() == false then
    app.Playback("/var/lib/asterisk/sounds/custom/beantwoorder")
    app.Hangup(16)
    end

    local https = require('ssl.https')
    callee = channel.CALLERID("num"):get()
    app.Verbose(1, string.format("Received call for %s from %s", extension, callee))
    url = string.format("https://X.X.X.X/rest/?apikey=%s&cmd=numinfo&arg1=%s", APIKEY, callee)
    app.Verbose(1, string.format("Requesting %s", url))
    r, c, h, s = https.request(url)
    name = nil
    number = callee
    for s1, s2 in string.gmatch(r, "([%w%s]+):([%w%s]+)") do
    name = s1
    end
    if name == nil then
    name = ""
    end
    if number == nil then
    number = callee
    end
    channel.CALLERID("all"):set(string.format("%s <%s>", name, '0' .. number))
    app.Queue("q1", "tTrn", "", "", 12)
    app.Queue("q2", "tTr", "", "", 28)
    app.Hangup(17)
    end


    -- Helper functions

    function extension_to_location(extension)
    for row in db:nrows(string.format("SELECT * FROM ast_users WHERE extension = '%s' LIMIT 1", extension)) do
    if row['location'] == nil then
    return nil
    end
    return string.format("%s", row['location'])
    end

    return nil
    end

    function location_to_extension(location)
    for row in db:nrows(string.format("SELECT * FROM ast_users WHERE location = '%s' LIMIT 1", location)) do
    if row['extension'] == nil then
    return nil
    end
    return string.format("%s", row['extension'])
    end

    return nil
    end

    function location_to_name(location)
    for row in db:nrows(string.format("SELECT * FROM ast_users WHERE location = '%s' LIMIT 1", location)) do
    if row['first_name'] == nil then
    return nil
    end
    return string.format("%s", row['first_name'])
    end

    return nil
    end

    function get_extension_status(extension)
    for row in db:nrows(string.format("SELECT * FROM ast_users WHERE extension = '%s' LIMIT 1", extension)) do
    if row['status'] == 1 then
    return true
    else
    return false
    end
    end
    end

    function get_extension_pin(extension)
    for row in db:nrows(string.format("SELECT * FROM ast_users WHERE extension = '%s' LIMIT 1", extension)) do
    return string.format("%s", row['pin'])
    end
    end

    function get_extension_queue(extension)
    for row in db:nrows(string.format("SELECT * FROM ast_users WHERE extension = '%s' LIMIT 1", extension)) do
    return string.format("%s", row['queue'])
    end
    end

    function in_office_hours()
    t = os.date("*t")

    -- Werkdagen + zaterdag en zondag
    if t['wday'] == 1 or t['wday'] == 7 or t['hour'] >= 17 or t['hour'] < 8 or (t['hour'] == 8 and t['min'] < 30) then
    return false
    end

    -- Koninginnedag
    if t['month'] == 4 and t['day'] == 30 then
    return false
    end

    -- Bevrijdingsdag
    if t['month'] == 5 and t['day'] == 5 then
    return false
    end

    -- Kerstdagen
    if t['month'] == 12 and t['day'] == 25 then
    return false
    end
    if t['month'] == 12 and t['day'] == 26 then
    return false
    end

    -- Oudejaarsdag
    if t['month'] == 12 and t['day'] == 31 then
    return false
    end

    return true
    end

    extensions = {
    -- Incoming calls
    ["unauthenticated"] = {
    ["_."] = e_drop_call
    };

    ["sipphones"] = {
    ["_#NXX"] = e_hotdesk_login,
    ["_#000"] = e_hotdesk_logout,
    ["_0X."] = e_outgoing_sip,
    ["_NXX"] = e_internal_sip,
    ["112"] = e_outgoing_emergency,
    ["911"] = e_outgoing_emergency,
    ["_*."] = e_dial_custnr,
    ["_**."] = e_dial_employee
    };

    ["bri-incoming"] = {
    ["_X."] = e_incoming

    };

    ["default"] = {
    include = { "unauthenticated" }
    };
    }