اتصال برنامه به Asterisk با AMI و ARI؛ از کجا شروع کنیم؟

اگر هدف فقط پیدا‌کردن تماس پایان‌یافته و خروجی گزارش است، راهنمای CDR ایزابل ممکن است کافی باشد. اگر قبلاً اتصال CRM داشته‌اید و رکوردها گم یا تکراری شده‌اند، راهنمای ثبت‌نشدن تماس در CRM موضوع جداست. اینجا یک نقطه شروع فنی برای توسعه تازه می‌سازیم.

پیکربندی و رفتار API با کد رسمی Asterisk 18.20.2 تطبیق داده شده است. برنامه‌ها برای Python 3.11 روی Linux نوشته شده‌اند و نمونه ARI از websocket-client 1.8.0 استفاده می‌کند. این اعداد مبنای بررسی‌اند، نه توصیه نصب نسخه قدیمی روی مرکز عملیاتی؛ نسخه و ماژول‌های نصب‌شده باید با نمونه تطبیق داده شوند.

یک Asterisk آزمایشیِ مستقل و بدون ترانک بیرونی، یک گوشی یا سافت‌فون PJSIP ثبت‌شده و دسترسی مجاز به تنظیمات و کنسول لازم است. برنامه Python روی همان میزبان اجرا می‌شود؛ در این طرح، localhost داخل کانتینر جدا یا لپ‌تاپ دیگر به PBX اشاره نمی‌کند. آزمون با شماره و اطلاعات ساختگی انجام شود.

از تنظیمات فعلی و context داخلی آزمایشی نسخه پشتیبان بگیرید. مسیر معمول فایل‌ها /etc/asterisk است؛ مسیر واقعی نصب را بررسی کنید. در فایل موجود بخش [general] را دوباره نسازید و فایل کامل را با تکه‌کد جایگزین نکنید. در نصب‌های دارای پنل مانند ایزابل، روش تغییر و فایل قابل ویرایش باید با همان پنل هماهنگ باشد؛ این نمونه، دستور ویرایش فایل‌های تولیدشده پنل نیست. تفاوت ایزابل و استریسک این مرز را توضیح می‌دهد.

core show version
module show like res_ari
module show like res_http_websocket
core show application Stasis
core show function TIMEOUT

برای بخش ARI، ماژول‌های res_ari، res_ari_events، res_ari_channels، res_http_websocket و Stasis و تابع TIMEOUT باید در نصب آزمایشی موجود و بارگذاری شده باشند. اگر نیستند، بسته و پیکربندی همان نسخه را اصلاح کنید؛ ماژول ناشناخته از نسخه دیگر کپی نکنید. اجرای نمونه Python به دسترسی root نیاز ندارد.

در manager.conf آزمایشگاه، تنظیم عمومی و حساب تازه زیر را ادغام کنید. عبارت REPLACE را با رمز تصادفی مستقل و طولانی از حروف و اعداد ASCII جایگزین کنید؛ همین رمز را هنگام اجرای برنامه وارد خواهید کرد. حساب آماده یا رمز مدیر پنل را استفاده نکنید.

[general]
enabled = yes
webenabled = no
port = 5038
bindaddr = 127.0.0.1

[nm_watch]
secret = REPLACE_WITH_UNIQUE_AMI_SECRET
deny = 0.0.0.0/0.0.0.0
permit = 127.0.0.1/255.255.255.255
read = call
write =
eventfilter = Event: (Newchannel|Newstate|Hangup)

read=call دسته رخدادهای تماس را مجاز می‌کند و write خالی به حساب اجازه Originate، قطع تماس یا اجرای فرمان CLI نمی‌دهد. Login، Ping و Logoff عملیات پایه همین اتصال‌اند. eventfilter این مبنا یک عبارت منظم روی متن رخداد است و جای کنترل دسترسی کاربران یا محدودیت یک مشتری را نمی‌گیرد؛ این حساب برای مرکز آزمایشی است، نه دسترسی کاربر نهایی به همه تماس‌های شرکت.

manager reload
manager show settings
manager show user nm_watch
manager show command Ping

آدرس شنود باید 127.0.0.1 و پورت همین تمرین 5038 باشد؛ مجوز حساب را هم بررسی کنید. تغییر پیکربندی ممکن است مجوز نشست قبلی را همان لحظه عوض نکند؛ اتصال قبلی نمونه را ببندید و با نشست تازه آزمایش کنید. TCP این نمونه رمزنگاری ندارد و صرفاً محلی است.

کد زیر را در ami_watch.py ذخیره و با python3.11 ami_watch.py اجرا کنید. رمز در ورودی مخفی پرسیده می‌شود. پس از تأیید ورود و Pong، برنامه تا ۶۰ ثانیه رخدادها را می‌خواند؛ در همین بازه یک تماس داخلی آزمایشی برقرار و قطع کنید. برنامه برای شما شماره‌گیری نمی‌کند.

"""Bounded local lab listener; not a production connector or a PBX test result."""
import getpass
import json
import socket
import time
import uuid


class Reader:
    def __init__(self, sock):
        self.sock, self.buffer = sock, b""

    def take(self, separator, deadline):
        while separator not in self.buffer:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return None
            self.sock.settimeout(min(1, remaining))
            try:
                chunk = self.sock.recv(4096)
            except socket.timeout:
                continue
            if not chunk:
                raise ConnectionError("AMI closed the connection")
            self.buffer += chunk
            if len(self.buffer) > 262144:
                raise ValueError("AMI frame exceeds this lab limit")
        item, self.buffer = self.buffer.split(separator, 1)
        return item

    def message(self, deadline):
        raw = self.take(b"\r\n\r\n", deadline)
        if raw is None:
            return None
        fields = {}
        for line in raw.decode("utf-8", errors="replace").split("\r\n"):
            key, colon, value = line.partition(":")
            if colon:
                fields.setdefault(key.lower(), []).append(value.lstrip(" "))
        return fields


def first(message, key):
    return message.get(key, [""])[0]


def show_event(message):
    if first(message, "event") in {"Newchannel", "Newstate", "Hangup"}:
        keys = ["event", "uniqueid", "linkedid", "channel", "channelstatedesc"]
        print(json.dumps({k: first(message, k) for k in keys}), flush=True)


def action(sock, reader, name, **fields):
    action_id = uuid.uuid4().hex
    values = {"Action": name, "ActionID": action_id, **fields}
    sock.sendall(("\r\n".join(f"{k}: {v}" for k, v in values.items())
                  + "\r\n\r\n").encode("ascii"))
    deadline = time.monotonic() + 8
    while (message := reader.message(deadline)) is not None:
        show_event(message)
        if first(message, "actionid") == action_id and "response" in message:
            if first(message, "response") != "Success":
                raise PermissionError("AMI action rejected")
            return message
    raise TimeoutError("No matching AMI response")


def main():
    secret = getpass.getpass("AMI secret for nm_watch: ")
    if not secret or not secret.isascii() or any(c in secret for c in "\r\n"):
        raise ValueError("Use the configured single-line ASCII secret")
    with socket.create_connection(("127.0.0.1", 5038), timeout=5) as sock:
        reader = Reader(sock)
        banner = reader.take(b"\r\n", time.monotonic() + 5)
        if not banner or not banner.startswith(b"Asterisk Call Manager/"):
            raise ValueError("Unexpected AMI greeting")
        action(sock, reader, "Login", Username="nm_watch", Secret=secret, Events="call")
        try:
            response = action(sock, reader, "Ping")
            if first(response, "ping") != "Pong":
                raise ValueError("Missing Pong")
            print("Authenticated; listening for up to 60 seconds.", flush=True)
            deadline = time.monotonic() + 60
            while (message := reader.message(deadline)) is not None:
                show_event(message)
        finally:
            try:
                sock.settimeout(1)
                sock.sendall(b"Action: Logoff\r\n\r\n")
            except OSError:
                pass


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("Stopped.")
    except (OSError, ValueError) as error:
        raise SystemExit(f"AMI stopped: {type(error).__name__}") from None

پیام AMI با خط خالی، یعنی دو جفت CRLF، تمام می‌شود؛ یک recv الزاماً یک پیام نیست. کد، فریم ناقص را نگه می‌دارد و پاسخ را با ActionID همان فرمان تطبیق می‌دهد؛ رخدادِ بین پاسخ‌ها گم نمی‌شود. کلیدهای تکراری نیز به‌صورت فهرست حفظ می‌شوند. این نمونه فقط فیلدهای مشخص را چاپ می‌کند و یک کتابخانه کامل AMI نیست.

Newchannel یعنی یک کانال ساخته شده؛ Newstate تغییر وضعیت کانال و Hangup پایان همان کانال را گزارش می‌کند. Uniqueid شناسه کانال است؛ Linkedid برای ربط برخی بخش‌های مرتبط کمک می‌کند. یک گفت‌وگو ممکن است چند کانال و چند رخداد داشته باشد، پس هر سطر را یک تماس فروش نشمارید. پیام ورود یا Pong نیز فقط مسیر اتصال را تأیید می‌کند، نه موفقیت ثبت در CRM.

در http.conf و ari.conf آزمایشگاه، بخش‌های زیر را با تنظیمات فعلی ادغام کنید. پورت 8088 باید آزاد باشد و در این نمونه prefix برای HTTP تعریف نشده باشد. رمز ARI از رمز AMI جدا باشد. رابط مدیریتی را برای اجرای این تمرین روی اینترنت یا همه آدرس‌ها باز نکنید.

[general]
enabled = yes
bindaddr = 127.0.0.1
bindport = 8088

[general]
enabled = yes

[nm_ari_watch]
type = user
read_only = yes
password = REPLACE_WITH_DIFFERENT_ARI_SECRET
password_format = plain

در کنسول مرکز آزمایشیِ بدون تماس فعال، core reload را اجرا و سپس http show status را بررسی کنید؛ وضعیت واقعی شنود و مسیر /ari باید مشخص باشد. اگر خطای ماژول یا پیکربندی وجود دارد، همان را رفع کنید و تغییر را موفق فرض نکنید. در مرکز فعال، Reload و تغییر listener باید با مسئول نگهداری هماهنگ شود.

read_only=yes در این نسخه، درخواست‌های تغییر مانند POST و DELETE را رد می‌کند، اما GET /channels می‌تواند فهرست کانال‌های فعال مرکز را برگرداند. نام app در WebSocket هم مجوزِ محدود به یک کاربر یا داخلی نیست. این حساب را در مرورگر کاربر، کد فرانت‌اند یا محصول عمومی قرار ندهید. CORS جای احراز هویت و مجوز نیست.

روی همان میزبان، در پوشه آزمایش یک محیط مستقل Python بسازید و وابستگی نمونه را نصب کنید. بسته موردنیاز websocket-client است؛ نام ماژول import در کد websocket است.

python3.11 -m venv .venv
.venv/bin/python -m pip install websocket-client==1.8.0
.venv/bin/python ari_watch.py

پیش از فرمان آخر، کد زیر را در ari_watch.py ذخیره کنید. برنامه ابتدا GET /ari/channels می‌فرستد و سپس به WebSocket برنامه nm-demo وصل می‌شود. Authorization در header است؛ رمز در URL یا آرگومان اجرای برنامه قرار نمی‌گیرد. trace کامل HTTP/WebSocket را با اطلاعات واقعی فعال نکنید.

"""Read-only HTTP request and bounded ARI event listener for an isolated lab."""
import base64
import getpass
import json
import time
import urllib.request
import websocket


def main():
    secret = getpass.getpass("ARI password for nm_ari_watch: ")
    if not secret or not secret.isascii() or any(c in secret for c in "\r\n"):
        raise ValueError("Use the configured single-line ASCII password")
    token = base64.b64encode(("nm_ari_watch:" + secret).encode("ascii")).decode("ascii")
    auth = "Basic " + token
    request = urllib.request.Request(
        "http://127.0.0.1:8088/ari/channels", headers={"Authorization": auth})
    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
    with opener.open(request, timeout=5) as response:
        channels = json.load(response)
    print("Active channels:", len(channels))
    ws = websocket.create_connection(
        "ws://127.0.0.1:8088/ari/events?app=nm-demo",
        header=["Authorization: " + auth], subprotocols=["ari"],
        suppress_origin=True, http_no_proxy=["127.0.0.1"], timeout=5)
    try:
        print("WebSocket connected; dial the lab extension within 60 seconds.", flush=True)
        deadline = time.monotonic() + 60
        while (remaining := deadline - time.monotonic()) > 0:
            ws.settimeout(remaining)
            try:
                raw = ws.recv()
            except websocket.WebSocketTimeoutException:
                break
            if not raw:
                raise ConnectionError("WebSocket closed")
            event = json.loads(raw)
            print(json.dumps({"type": event.get("type"),
                              "channel_id": event.get("channel", {}).get("id")}), flush=True)
            if event.get("type") == "ApplicationReplaced":
                raise RuntimeError("Another client took over nm-demo")
    finally:
        ws.close()


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("Stopped.")
    except (OSError, ValueError, RuntimeError, websocket.WebSocketException) as error:
        raise SystemExit(f"ARI stopped: {type(error).__name__}") from None

صفر بودن Active channels هنگام نبود تماس، نتیجه طبیعیِ فهرست خالی است. دریافت پاسخ HTTP به‌تنهایی اتصال WebSocket و رسیدن رخداد را تأیید نمی‌کند. پس از پیام WebSocket connected، مرحله بعد را در همان بازه ۶۰ ثانیه انجام دهید. این متن، انتظار آزمون را توضیح می‌دهد؛ خروجی واقعیِ اجرای نت‌مسترز ارائه نشده است.

در extensions.conf آزمایشگاه، context تازه زیر را اضافه کنید. فقط endpoint آزمایشی PJSIP باید context=nm-ari-lab داشته باشد؛ تنظیم قبلی آن را ثبت کنید. شماره 700 نمونه است و باید آزاد باشد. این context هیچ مسیر ترانک یا تماس بیرونی ندارد. تغییر context داخلی فعال شرکت، بخشی از این تمرین نیست.

[nm-ari-lab]
exten => 700,1,Answer()
 same => n,Set(TIMEOUT(absolute)=20)
 same => n,Stasis(nm-demo)
 same => n,Hangup()

تغییر context را در تنظیم endpoint واقعی خود اعمال و در محیط آزمایشی بدون تماس، core reload کنید. سپس با dialplan show 700@nm-ari-lab و pjsip show endpoint نام-endpoint، مسیر و context مؤثر را بررسی کنید. برنامه ARI را اجرا کنید و فقط پس از اتصال WebSocket، از گوشی آزمایشی 700 را بگیرید.

باید رخداد StasisStart با channel_id مربوط دیده شود. خودتان تماس را قطع کنید یا پایان timeout را انتظار بکشید؛ خروج کانال از برنامه با StasisEnd قابل بررسی است. این مسیر Answer می‌کند و ممکن است سکوت شنیده شود، چون نمونه هیچ Playback یا پل رسانه‌ای نمی‌سازد. سکوت در این آزمایش، تست کیفیت صوت نیست.

اگر nm-demo فعال یا ثبت نشده باشد، Stasis می‌تواند با وضعیت FAILED برگردد و مسیر به Hangup بعدی برسد. دو نسخه برنامه با نام یکسان را هم‌زمان اجرا نکنید؛ اتصال تازه می‌تواند برنامه قبلی را با رخداد ApplicationReplaced کنار بزند. نمونه در این حالت متوقف می‌شود تا دو پردازش مالک یک برنامه فرض نشوند.

• نمونه‌ها یک نشست محدود دارند و پس از قطع، خودکار دوباره وصل نمی‌شوند. در برنامه پایدار، timeout و heartbeat تعریف کنید؛ اتصال مجدد با فاصله افزایشی و سقف مشخص باشد تا خرابی مرکز، حلقه اتصال سریع نسازد.
• پس از اتصال مجدد، دریافت رخدادهای از دست‌رفته را تضمین‌شده فرض نکنید. فاصله قطعی را ثبت و وضعیت فعلی را با منبع مجاز تطبیق دهید؛ درخواست snapshot هم باید مجوز لازم خودش را داشته باشد.
• ActionID برای ربط درخواست و پاسخ است، نه شناسه مشتری یا شناسه یکتای تمام مکالمه. برای ثبت در پایگاه داده، شناسه مرکز، کانال و معنای رخداد را نگه دارید و ارسال دوباره را بدون ساخت سابقه ناخواسته مدیریت کنید.
• پردازش سنگین و درخواست CRM را در حلقه دریافت رخداد معطل نکنید؛ صف محدود، کنترل فشار و ثبت خطای قابل پیگیری داشته باشید. موفقیت دریافت، موفقیت ذخیره مقصد نیست.
• برای کنترل واقعی تماس در ARI، تغییر مجوز کافی نیست؛ رفتار Answer، Bridge، Playback، پایان کانال، خطای برنامه و مسیر جایگزین باید طراحی و آزمایش شوند. این نمونه عمداً کنترل‌کننده تماس نیست.

اگر برنامه روی میزبان دیگری اجرا می‌شود، محل اجرا، شبکه مجاز و TLS یا تونل مناسب با اعتبارسنجی مشخص لازم است. برداشتن محدودیت loopback یا خاموش‌کردن بررسی گواهی را جای طراحی اتصال نگذارید. رمزها در مدیریت امن تنظیمات نگهداری شوند و فایل‌های پیکربندی فقط در اختیار حساب‌های مجاز باشند.

پذیرش آموزشی یعنی هر مسیر جدا نتیجه داشته باشد: AMI ورود و رخدادِ تماس معلوم؛ ARI درخواست خواندنی و ورود/خروج کانال آزمایشی؛ سپس توقف برنامه، پایان تماس و لغو دسترسی نمونه. رمز غلط را هم یک‌بار در محیط تحت اختیار بررسی کنید؛ این آزمون را به تلاش تکراری روی حساب دیگر تبدیل نکنید.

برای جمع‌کردن تمرین، تماس آزمایشی را تمام و هر دو برنامه را متوقف کنید؛ context قبلی endpoint را برگردانید و فقط حساب‌ها و context تازه را حذف کنید. وضعیت عمومی listenerها را مطابق ثبت اولیه بازیابی کنید؛ اگر جزء دیگری از HTTP یا AMI استفاده می‌کند، آن را بی‌بررسی غیرفعال نکنید. پس از اعمال، با نشست تازه بررسی کنید حساب حذف‌شده دیگر وارد نمی‌شود.

برای ساخت قابلیت واقعی شرکت، خدمات توسعه اختصاصی تلفن دامنه اجرا و نگهداری را توضیح می‌دهد. اگر هدف نمایش پرونده، سابقه یا شماره‌گیری از نرم‌افزار مشتریان است، اتصال ویپ به CRM باید با امکانات هر دو سیستم تعریف شود. نمونه این صفحه هنوز پایگاه داده، اتصال CRM، بازیابی رخداد و آزمون عملی مرکز را انجام نمی‌دهد.