Tạo màn hình theo dõi trạng thái các agent trong Queue qua AMI

Tổng đài: Sangoma PBXact

Máy theo dõi: Windows

Script: Python 3.13

Chuẩn bị:

- Mở kết nối AMI trên PBX

sudo nano /etc/asterisk/manager.conf

Sửa nội dung default:

enabled = yes
port = 5038
bindaddr = 127.0.0.1
displayconnects=no ;only effects 1.6+

[admin]
secret = xxxxxx
deny=0.0.0.0/0.0.0.0
permit=127.0.0.1/255.255.255.0
read = system,call,log,verbose,command,agent,user,config,command,dtmf,reporting,cdr,dialplan,originate,message
write = system,call,log,verbose,command,agent,user,config,command,dtmf,reporting,cdr,dialplan,originate,message
writetimeout = 5000 

sửa lại bindaddr = 0.0.0.0

thêm dòng vào phía dưới permit=1.2.3.0/255.255.255.0 : với mỗi dải IP phải thêm 1 dòng permit tương ứng. Nếu bỏ 127.0.0.1 thì web quản trị sẽ báo lỗi ko kết nối đc asterisk (mất kết nối nội bộ trong PBX)

Sau đó chạy asterisk: manager reload để cập nhật 

Test bằng cách telnet tới IP PBX, port 5038

 

- Cài Python với thư viện 

Thư viện python:

- pip install panoramisk

- pip install pyst2 

- pip install colorama 

Source code, cần sửa lại IP và Password kết nối AMI:

import asyncio
from panoramisk.manager import Manager
from datetime import datetime
import colorama
colorama.init()
import csv
import os

manager = Manager(
    host='xxxxxxxxx',
    port=5038,
    username='admin',
    secret='xxxxxxxxx'
)

status_map = {
    '0': 'Unknown',
    '1': 'Ready',
    '2': 'In call',
    '3': 'Busy',
    '4': 'Invalid',
    '5': 'Unavailable',
    '6': 'Ringing',
    '7': 'Ringing',
    '8': 'On Hold'
}

sip_to_name = {
    "SIP/8860": "8860   HoanDV1",
    "SIP/8862": "8862   AnhTTM2",
    "SIP/8867": "8867   DontKnow",
    "SIP/8890": "8890   BachTQ2",
    "SIP/8891": "8891   HuyenPTT5",
    "SIP/8892": "8892   TungPT1",
    "SIP/8893": "8893   LyLK",
    "SIP/8897": "8897   TuanBV",
    "SIP/8899": "8899   TuanNN2",
    "PJSIP/60801": "60801  TrinhBT",
    "PJSIP/60802": "60802  HieuPT1",
    "PJSIP/60803": "60803  HungDV2"
}

log_file = 'agent_log.csv'
write_header = not os.path.exists(log_file)

async def get_agents():
    global write_header

    response = await manager.send_action({'Action': 'QueueStatus'})
    agents = {}

    for item in response:
        if item.get('Event') == 'QueueMember':
            location = item.get('Location', '')
            name = sip_to_name.get(location, location)
            queue = item.get('Queue')
            status_code = str(item.get('Status', '0'))
            status_text = status_map.get(status_code, 'Unknown')
            calls_taken = item.get('CallsTaken', '0')

            raw_last_call = item.get('LastCall', '0')
            try:
                ts = int(raw_last_call)
                if ts > 0:
                    last_call = datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")
                else:
                    last_call = "Chưa có"
            except ValueError:
                last_call = "Không hợp lệ"

            if queue not in agents:
                agents[queue] = []
            agents[queue].append({
                'name': name,
                'status': status_text,
                'code': status_code,
                'calls': calls_taken,
                'last': last_call
            })

    # In ra console
    current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    for queue, agent_list in agents.items():
        ready_count = sum(1 for ag in agent_list if ag['status'] == 'Ready')
        in_call_count = sum(1 for ag in agent_list if ag['status'] == 'In call')

        print(f"\n\033[93m{current_time} | Queue: {queue} | Ready: {ready_count} | In call: {in_call_count}\033[0m")
        print("\033[92mAgent  Name       Agent Status         Calls    Last Call\033[0m")
        print("-----------------------------------------------------------")
        for ag in agent_list:
            print(f"{ag['name']:<18}{ag['status']:<16}{ag['code']:<5}{ag['calls']:<9}{ag['last']}")

    # Ghi log CSV
    rows = []
    for queue, agent_list in agents.items():
        for ag in agent_list:
            rows.append([
                current_time,
                queue, ag['name'], ag['status'], ag['code'], ag['calls'], ag['last']
            ])
    
    with open(log_file, mode='a', newline='', encoding='utf-8') as file:
        writer = csv.writer(file)
        if write_header:
            writer.writerow(["Timestamp", "Queue", "Agent Name", "Status", "Status Code", "Calls Taken", "Last Call"])
            write_header = False
        writer.writerows(rows)

async def run():
    await manager.connect()
    try:
        while True:
            await get_agents()
            await asyncio.sleep(5)
    except KeyboardInterrupt:
        print("Đã dừng cập nhật.")
    finally:
        manager.close()

if __name__ == "__main__":
    asyncio.run(run())