Differences

This shows you the differences between two versions of the page.

--- язык_программирования_python [2025/05/24 17:58]
val [Zabbix API приложение]
+++ язык_программирования_python [2026/09/06 13:05] (current)
val [ari_chat.py]
@@ Line 1: / Line 1: @@
 ====== Язык программирования Python ======
+  * [[Все что должен знать DevOps про Python]]
+  * [[https://habr.com/ru/articles/277679/|Пишем shell скрипты на Python и можно ли заменить им Bash]]
+  * [[https://github.com/rupertsworld/frontend-backend|Basic Web Frontend/Backend Demo]]
+  * [[https://code.tutsplus.com/ru/using-celery-with-django-for-background-task-processing--cms-28732t|Использование Celery с Django для обработки фоновых задач]]
 ===== Виртуальная среда Python =====
   * [[https://blog.sedicomm.com/2021/06/29/chto-takoe-venv-i-virtualenv-v-python-i-kak-ih-ispolzovat/|Что такое venv и virtualenv в Python, и как их использовать]]
-  * [[https://packaging.python.org/en/latest/guides/installing-using-pip-and-virtual-environments/|https://packaging.python.org/en/latest/guides/installing-using-pip-and-virtual-environments/]]
+  * [[https://packaging.python.org/en/latest/guides/installing-using-pip-and-virtual-environments/|Install packages in a virtual environment using pip and venv]]
-  * [[https://www.workroom-productions.com/serving-a-directory-with-flask/|Publishing a Directory with Flask]]
+  * [[https://habr.com/ru/articles/593529/|Poetry — прекрасная альтернатива pip (шпаргалка)]]
+  * [[https://habr.com/ru/articles/828016/|UV. Обзор пакетного менеджера Python]]
 <code>
+# python3 -V
 # apt install python3-venv
@@ Line 19: / Line 29: @@
 </code>
+===== Интерактивная оболочка REPL (Read-Eval-Print Loop) =====
+<code>
+(venv1) # python
+</code><code>
+>>> import os
+>>> print ("Hello " + os.environ.get('USER') + "!")
+</code>
+  * [[https://ru.wikipedia.org/wiki/IPython]]
+<code>
+(venv1) # pip install ipython
+</code>
+  * [[https://jupyter.org/try-jupyter/notebooks/?path=notebooks/Intro.ipynb|try-jupyter/notebooks]]
+  * [[https://matplotlib.org/stable/tutorials/pyplot.html|Pyplot tutorial]]
 ===== Web приложение =====
+  * [[https://webadventures.ru/sravnenie-wsgi-serverov-uvicorn-i-gunicorn/|Сравнение WSGI-серверов:uvicorn и Gunicorn]]
+==== Flask Gunicorn ====
+  * [[#Виртуальная среда Python]]
+  * [[https://habr.com/ru/articles/750312/|Python декораторы на максималках. Универсальный рецепт по написанию и аннотированию от мала до велика]]
+  * [[https://www.workroom-productions.com/serving-a-directory-with-flask/|Publishing a Directory with Flask]]
   * [[https://python.ivan-shamaev.ru/run-install-deploy-flask-web-app-docker-dockerfile-compose/|Создание Web-приложения Flask и деплой с помощью Docker Compose & Dockerfile]]
   * [[https://blog.miguelgrinberg.com/post/running-your-flask-application-over-https|Running Your Flask Application Over HTTPS]]
   * [[https://docs.python.org/3/library/configparser.html|configparser — Configuration file parser]]
-  * [[#Виртуальная среда Python]]
 <code>
@@ Line 37: / Line 72: @@
 import os
 import configparser
+config = configparser.ConfigParser()
+config.read('/etc/pywebd/pywebd.conf')
+if 'PYWEBD_DOC_ROOT' in os.environ:
+  pywebd_doc_root = os.environ.get('PYWEBD_DOC_ROOT')
+else:
+  pywebd_doc_root = config['default']['DocumentRoot']
 app = Flask(__name__)
@@ Line 50: / Line 93: @@
 if __name__ == "__main__":
-    config = configparser.ConfigParser()
-    config.read('/etc/pywebd/pywebd.conf')
-#    pywebd_port = os.environ.get('PYWEBD_PORT',config['default']['Listen'])
-#    pywebd_doc_root = os.environ.get('PYWEBD_DOC_ROOT',config['default']['DocumentRoot'])
-    if 'PYWEBD_PORT' in os.environ:
-      pywebd_port = os.environ.get('PYWEBD_PORT')
-    else:
-      pywebd_port = config['default']['Listen']
-    if 'PYWEBD_DOC_ROOT' in os.environ:
-      pywebd_doc_root = os.environ.get('PYWEBD_DOC_ROOT')
-    else:
-       pywebd_doc_root = config['default']['DocumentRoot']
-    app.run(host="0.0.0.0", port=pywebd_port, debug=True)
+  if 'PYWEBD_PORT' in os.environ: pywebd_port = os.environ.get('PYWEBD_PORT')
-#    app.run(ssl_context=('/etc/pywebd/pywebd.crt', '/etc/pywebd/pywebd.key'), debug=True, host='0.0.0.0', port=pywebd_port)
+  else: pywebd_port = config['default']['Listen']
+  app.run(host="0.0.0.0", port=pywebd_port, debug=True)
+#  app.run(ssl_context=('/etc/pywebd/pywebd.crt', '/etc/pywebd/pywebd.key'), debug=True, host='0.0.0.0', port=pywebd_port)
 </code><code>
 # mkdir -p /etc/pywebd/
@@ Line 84: / Line 117: @@
 (venv1) :~/pywebd$ #export PYWEBD_PORT=4443
 (venv1) :~/pywebd$ #export PYWEBD_DOC_ROOT=/var/www/html/
-(venv1) :~/pywebd$ #PYWEBD_PORT=4443 PYWEBD_DOC_ROOT=/var/www/html/ python app.py
+(venv1) :~/pywebd$ #PYWEBD_PORT=4080 PYWEBD_DOC_ROOT=/var/www/ python app.py
 (venv1) :~/pywebd$ python app.py
 (venv1) :~/pywebd$ pip freeze | tee requirements.txt
+(venv1) :~/pywebd$ #pip install gunicorn
+(venv1) :~/pywebd$ #gunicorn app:app --bind 0.0.0.0:8000
+(venv1) :~/pywebd$ #cat gunicorn.conf.py
+</code><code>
+import os
+import configparser
+myconfig = configparser.ConfigParser()
+myconfig.read('/etc/pywebd/pywebd.conf')
+if 'PYWEBD_PORT' in os.environ: pywebd_port = os.environ.get('PYWEBD_PORT')
+else: pywebd_port = myconfig['default']['Listen']
+#certfile = "/etc/pywebd/pywebd.crt"
+#keyfile = "/etc/pywebd/pywebd.key"
+bind = '0.0.0.0:' + pywebd_port
+</code><code>
+(venv1) :~/pywebd$ # gunicorn -c gunicorn.conf.py app:app
 </code>
-===== CRUD Rest API приложение =====
+==== FastAPI Uvicorn ====
-  * [[https://dev.to/francescoxx/python-crud-rest-api-using-flask-sqlalchemy-postgres-docker-docker-compose-3kh4|Python CRUD Rest API using Flask, SQLAlchemy, Postgres, Docker, Docker Compose]]
+  * [[https://python-poetry.org/docs/basic-usage/]]
+  * [[https://habr.com/ru/articles/593529/|Poetry — прекрасная альтернатива pip (шпаргалка)]]
+<code>
+server# ###docker run -ti --rm --name pywebd2 python:3.11-alpine sh
+/ # ### pip install poetry
+server# apt install python3-poetry; export PYTHON_KEYRING_BACKEND=keyring.backends.null.Keyring #https://stackoverflow.com/questions/74438817/poetry-failed-to-unlock-the-collection
+/ # poetry new pywebd2
+/ # cd pywebd2
+/pywebd2 # poetry add fastapi dotenv uvicorn ###gunicorn
+/pywebd2 # more pyproject.toml
+/pywebd2 # cat app.py
+</code><code>
+from fastapi import FastAPI
+from dotenv import load_dotenv
+import os
+CONST_VER = "ver1.1"
+hostname = os.uname().nodename
+load_dotenv()
+pywebd_message = os.environ.get('PYWEBD_MESSAGE')
+app = FastAPI()
+@app.get("/")
+async def home():
+   return {"data": f"{pywebd_message} from {hostname} app {CONST_VER}"}
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)
+</code><code>
+/pywebd2 # cat .env
+</code><code>
+PYWEBD_MESSAGE="Hello World!!!"
+</code>
+  * [[https://tproger.ru/articles/testiruem-na-python-unittest-i-pytest-instrukcija-dlja-nachinajushhih|Тестируем на Python: unittest и pytest. Инструкция для начинающих]]
+  * [[https://habr.com/ru/articles/349860/|Регулярные выражения в Python от простого к сложному. Подробности, примеры, картинки, упражнения]]
+  * [[https://regex101.com/|Online Regular expression tester with syntax highlighting]]
+<code>
+/pywebd2 # cat tests/test_ver_format.py
+</code><code>
+import app, re
+assert re.match(r'^ver\d+\.\d+$', app.CONST_VER),"Incorrect VER format"
+</code><code>
+/pywebd2 # poetry run python3 -m unittest
+/pywebd2 # #export PYWEBD_MESSAGE="Another message"
+/pywebd2 # poetry run python3 app.py
+/pywebd2 # poetry run uvicorn app:app --reload --host 0.0.0.0 --port 4080
+</code>
+  * [[https://github.com/benoitc/gunicorn/issues/2154|TypeError: __call__() missing 1 required positional argument: 'send']]
+<code>
+/pywebd2 # ###poetry run gunicorn app:app --bind 0.0.0.0:6080 -k uvicorn.workers.UvicornWorker
+</code>
+===== Zabbix LLD приложение =====
+  * [[https://askubuntu.com/questions/244378/running-python-file-in-terminal|Running Python File in Terminal]]
+  * [[https://pythonworld.ru/moduli/modul-subprocess.html|Модуль subprocess]]
+  * [[https://habr.com/ru/articles/488112/|Понимание итераторов в Python]]
+  * [[https://www.geeksforgeeks.org/python-find-dictionary-matching-value-in-list/|Find dictionary matching value in list - Python]]
+  * [[https://docs-python.ru/tutorial/vstroennye-funktsii-interpretatora-python/funktsija-next/|Функция next() в Python, следующий элемент итератора]]
+==== Пример 1 ====
+<code>
+gate# apt install python3-xmltodict/stable
+gate# cat /etc/zabbix/dhcp-pools.py
+</code><code>
+#!/usr/bin/env python3
+# Example usage:
+# ./dhcp-pools.py
+# ./dhcp-pools.py LAN1 defined|used
+from sys import argv
+import subprocess, xmltodict, json
+p = subprocess.Popen("/usr/bin/dhcpd-pools -f x", stdout=subprocess.PIPE, shell=True)
+(output, err) = p.communicate()
+p_status = p.wait()
+o = xmltodict.parse(output)
+#print (o)
+if len(argv)==1:
+  r=[]
+  for i in o['dhcpstatus']['shared-network']:
+    s={}
+    s["{#POOLNAME}"]=i['location']
+    r.insert(1,s)
+  print(json.dumps(r))
+else:
+  LANNAME=argv[1]
+  USDEF=argv[2]
+  res = next((sub for sub in o['dhcpstatus']['shared-network'] if sub['location'] == LANNAME), None)
+  print(res[USDEF])
+</code>
+==== Пример 2 ====
+<code>
+# cat /usr/local/bin/asterisk.chansip.discovery.py
+</code><code>
+#!/usr/bin/env python3
+import subprocess, io, json
+result=[]
+proc = subprocess.Popen('/usr/sbin/asterisk -x "sip show users" | tail -n +2 | cut -d" " -f1', stdout=subprocess.PIPE, shell=True)
+for line in io.TextIOWrapper(proc.stdout, encoding="utf-8"):
+  #print(line.rstrip())
+  s={}
+  s["{#CHANNAME}"]="SIP/"+line.rstrip()
+  result.insert(1,s)
+print(json.dumps(result))
+</code>
+Вариант от DeepSeek
+<code>
+# cat /usr/local/bin/asterisk.chansip.discovery.ds.py
+</code><code>
+#!/usr/bin/env python3
+import subprocess
+import json
+# Запускаем команду и сразу получаем вывод
+output = subprocess.check_output(
+    '/usr/sbin/asterisk -x "sip show users" | tail -n +2 | cut -d" " -f1',
+    shell=True,
+    text=True
+)
+# Формируем список словарей в одну строку
+result = [{"{#CHANNAME}": f"SIP/{line.strip()}"} for line in output.splitlines() if line.strip()]
+# Выводим результат в формате JSON
+print(json.dumps(result))
+</code>
 ===== Zabbix API приложение =====
+  * [[https://www.zabbix.com/documentation/current/en/manual/api]]
   * [[https://sbcode.net/zabbix/zabbix-api-python-example/]]
+  * [[https://forum.checkmk.com/t/rest-api-python-question-how-to-use-variables-in-json-post-in-key-and-value/34652]]
 <code>
 (venv1) server:~# pip install requests
-(venv1) server:~# cat zab_get_problem.py
+(venv1) server:~# cat zab_set_map_name.py
 </code><code>
+#!/usr/bin/env python3
 import requests
 import json
+from sys import argv
+if len(argv) != 3 :
+    print("You must set argument!!!\nExample: python zab_set_map_name.py 2 \"ISP 1\"")
+    quit()
+MAPID = argv[1]
+MAPNAME = argv[2]
 ZABBIX_API_URL = "http://127.0.0.1/zabbix/api_jsonrpc.php"
@@ Line 112: / Line 333: @@
 PWORD = "zabbix"
+print("\nLogin user {} to Zabbiz API".format(UNAME))
 r = requests.post(ZABBIX_API_URL,
                   json={
@@ Line 126: / Line 348: @@
 AUTHTOKEN = r.json()["result"]
-print(AUTHTOKEN)
+#print("Rename Map with ID", MAPID, "to", MAPNAME)
+print(f"Rename Map with ID {MAPID} to {MAPNAME}")
-# Retrieve a list of problems
-print("\nRetrieve a list of problems")
 r = requests.post(ZABBIX_API_URL,
                   headers={'Authorization': 'Bearer ' + AUTHTOKEN},
                   json={
-                      "jsonrpc": "2.0",
+                    "jsonrpc": "2.0",
-                      "method": "problem.get",
+                    "method": "map.update",
-                      "params": {},
+                    "params": {
-                      "id": 2,
+                        "sysmapid": MAPID,
+                        "name": MAPNAME
+                    },
+                    "id": 2
                   })
 print(json.dumps(r.json(), indent=4, sort_keys=True))
-#Logout user
 print("\nLogout user")
 r = requests.post(ZABBIX_API_URL,
@@ Line 154: / Line 377: @@
 print(json.dumps(r.json(), indent=4, sort_keys=True))
 </code><code>
-(venv1) server:~# python3 zab_get_problem.py
+(venv1) server:~# chmod +x zab_set_map_name.py
+(venv1) server:~# ./zab_set_map_name.py 2 "ISP 1"
 </code>
+===== CRUD Rest API приложение =====
+  * [[https://dev.to/francescoxx/python-crud-rest-api-using-flask-sqlalchemy-postgres-docker-docker-compose-3kh4|Python CRUD Rest API using Flask, SQLAlchemy, Postgres, Docker, Docker Compose]]
 ===== Дополнительные материалы =====
+  * [[https://pypi.org/project/ansible-output-parser/]]
+  * [[https://www.cyberciti.biz/faq/python-run-external-command-and-get-output/]]
+  * [[https://selectel.ru/blog/tutorials/how-to-develop-fastapi-application/]]
 ==== Доступ к каталогу по http ====
@@ Line 166: / Line 401: @@
 </code>
+==== wav_and_ogg.py ====
+  * [[Перекодировка звука#Перекодировка в ogg формат]]
+<code>
+$ pip install pydub pip #audioop-lts
+</code><code>
+import sys
+from pydub import AudioSegment
+def wav_to_ogg(file_path_wo_ext):
+    audio = AudioSegment.from_wav(f"{file_path_wo_ext}.wav")
+    audio.export(f"{file_path_wo_ext}.ogg", format="ogg", codec="libopus", bitrate="24k")
+    print(f"Файл успешно конвертирован и сохранен как: {file_path_wo_ext}.ogg")
+def ogg_to_wav(file_path_wo_ext):
+    audio = AudioSegment.from_ogg(f"{file_path_wo_ext}.ogg")
+    audio = audio.set_channels(1)
+    audio = audio.set_frame_rate(8000)
+    audio = audio.set_sample_width(2)
+    audio.export(f"{file_path_wo_ext}.wav", format="wav")
+    print(f"Файл успешно конвертирован и сохранен как: {file_path_wo_ext}.wav")
+if __name__ == "__main__":
+    target_file = sys.argv[1]
+#    wav_to_ogg(target_file)
+#    ogg_to_wav(target_file)
+</code><code>
+$ python wav_and_ogg.py speech
+</code>
+==== speech_and_text.py ====
+  * [[Yandex AI]]
+<code>
+import os
+import sys
+import requests
+API_KEY = os.getenv("YANDEX_API_KEY")
+FOLDER_ID = os.getenv("YANDEX_FOLDER_ID")
+def speech_to_text(file_path):
+    url = f"https://stt.api.cloud.yandex.net/speech/v1/stt:recognize?folderId={FOLDER_ID}"
+    headers = {
+        "Authorization": f"Api-Key {API_KEY}",
+        "Content-Type": "audio/ogg; codecs=opus",
+    }
+    params = {"lang": "ru-RU"}
+    with open(file_path, "rb") as f:
+        audio_data = f.read()
+    response = requests.post(
+       url, headers=headers, params=params, data=audio_data, timeout=15
+    )
+    result_json = response.json()
+    text = result_json.get("result", "")
+    return text
+def text_to_speech(text, output_file_path):
+    url = "https://tts.api.cloud.yandex.net/speech/v1/tts:synthesize"
+    headers = {
+        "Authorization": f"Api-Key {API_KEY}"
+    }
+    data = {
+        "text": text,
+        "lang": "ru-RU",
+        "voice": "marina",
+        "format": "oggopus",
+        "folderId": FOLDER_ID
+    }
+    response = requests.post(url, headers=headers, data=data, timeout=15)
+    if response.status_code == 200:
+        with open(output_file_path, "wb") as f:
+            f.write(response.content)
+        print(f"Файл успешно сохранен: {output_file_path}")
+    else:
+        print(f"Ошибка API: {response.status_code} - {response.text}")
+if __name__ == "__main__":
+    file_name = sys.argv[1]
+#    text = speech_to_text(file_name)
+#    print(text)
+#    text = sys.argv[2]
+#    text_to_speech(text, file_name)
+</code>
+==== request_to_agent.py ====
+<code>
+import sys
+import os
+from yandex_ai_studio_sdk import AIStudio
+def request_to_agent(user_request, system_prompt, context_messages=None):
+    sdk = AIStudio(
+        folder_id=os.getenv("YANDEX_FOLDER_ID"),
+        auth=os.getenv("YANDEX_API_KEY")
+    )
+    messages = [{"role": "system", "text": system_prompt}]
+    if context_messages:
+        messages.extend(context_messages)
+    messages.append({"role": "user", "text": user_request})
+    #model = sdk.chat.completions("yandexgpt")
+    model = sdk.chat.completions("aliceai-llm")
+    model = model.configure(temperature=0.3, max_tokens=1500)
+    response = model.run(messages)
+    text = response.choices[0].text
+    return text
+if __name__ == "__main__":
+    system_prompt = (
+        "Ты — ассистент.\n"
+        "Отвечай кратко.\n"
+    )
+    chat_context = []
+    #chat_context = [
+    #    {"role": "user", "text": "Привет! Меня зовут Алексей, я программист."},
+    #    {"role": "assistant", "text": "Привет, Алексей! Рад знакомству. Чем могу помочь?"}
+    #]
+    while True:
+        try:
+            user_request = input("Введите ваш вопрос: ")
+        except (KeyboardInterrupt, EOFError):
+            print("\nСессия завершена!")
+            sys.exit(0)
+        reply_text = request_to_agent(user_request, system_prompt, chat_context)
+        print(reply_text)
+        chat_context.append({"role": "user", "text": user_request})
+        chat_context.append({"role": "assistant", "text": reply_text})
+</code>
+==== ari_rec.py ====
+<code>
+root#
+mkdir -p /var/spool/asterisk/recording/
+chown -R asterisk:asterisk /var/spool/asterisk/recording/
+chmod -R 775 /var/spool/asterisk/recording/
+$ pip install websocket-client requests
+</code><code>
+import json
+import requests
+import websocket
+ARI_USER = "asterisk"
+ARI_PASS = "asterisk"
+BASE_URL = "http://localhost:8088/ari"
+AUTH = (ARI_USER, ARI_PASS)
+def detect_speech(channel_id):
+    print(f"Запускаем TALK_DETECT для канала {channel_id}...")
+    talk_url = f"{BASE_URL}/channels/{channel_id}/variable"
+    # 2000 мс тишины для завершения фразы, 500 мс речи для начала
+    talk_payload = {
+        "variable": "TALK_DETECT(set)",
+        "value": "2000,500"
+    }
+    response = requests.post(talk_url, json=talk_payload, auth=AUTH)
+    print(f"Статус-код ответа: {response.status_code}")
+    print(f"Текст ответа: {response.text}")
+def undetect_speech(channel_id):
+    print(f"Отключаем TALK_DETECT для канала {channel_id}...")
+    talk_url = f"{BASE_URL}/channels/{channel_id}/variable"
+    talk_payload = {
+        "variable": "TALK_DETECT(set)",
+        "value": "remove"
+    }
+    response = requests.post(talk_url, json=talk_payload, auth=AUTH)
+    print(f"Статус-код ответа: {response.status_code}")
+    print(f"Текст ответа: {response.text}")
+def start_recording(channel_id):
+    print(f"Включаем запись для канала {channel_id}")
+    record_url = f"{BASE_URL}/channels/{channel_id}/record"
+    recording_name = f"rec_{channel_id}"
+    record_payload = {
+        "name": recording_name,
+        "format": "wav",
+        "ifExists": "overwrite",
+        #"maxDuration": 29,
+        #"terminateOn": "#",
+        #"beep": True,
+    }
+    requests.post(record_url, params=record_payload, auth=AUTH)
+def stop_recording(channel_id):
+    print(f"Останавливаем запись для канала {channel_id}")
+    recording_name = f"rec_{channel_id}"
+    stop_url = f"{BASE_URL}/recordings/live/{recording_name}/stop"
+    requests.post(stop_url, auth=AUTH)
+def on_message(wsapp, message):
+    event = json.loads(message)
+    event_type = event.get("type")
+    # Сценарий 1: Звонок поступил -> Отвечаем и воспроизводим beep
+    if event_type == "StasisStart":
+        channel_id = event.get("channel", {}).get("id")
+        print(f"Отвечаем на звонок {channel_id}")
+        requests.post(f"{BASE_URL}/channels/{channel_id}/answer", auth=AUTH)
+        #start_recording(channel_id)
+        # or
+        detect_speech(channel_id)
+    # Сценарий N
+    elif event_type == "ChannelTalkingStarted":
+        channel_id = event.get("channel", {}).get("id")
+        print(f"Обнаружена речь в канале {channel_id}! Начинаем запись...")
+        start_recording(channel_id)
+    # Сценарий N
+    elif event_type == "ChannelTalkingFinished":
+        channel_id = event.get("channel", {}).get("id")
+        print(f"Обнаружено молчание в канале {channel_id}")
+        stop_recording(channel_id)
+        undetect_speech(channel_id)
+    # Сценарий N: Запись завершилась -> Проигрываем её обратно
+    elif event_type == "RecordingFinished":
+        recording_data = event.get("recording", {})
+        recording_name = recording_data.get("name")
+        target_uri = recording_data.get("target_uri", "")
+        channel_id = target_uri.split("channel:")[1]
+        print(f"Запись {recording_name} готова! Проигрываем её обратно в канал {channel_id}...")
+        play_url = f"{BASE_URL}/channels/{channel_id}/play"
+        play_payload = {
+            "media": f"recording:{recording_name}"
+        }
+        response = requests.post(play_url, params=play_payload, auth=AUTH)
+        print("Ответ ARI на воспроизведение:", response.status_code)
+    # Сценарий N: Воспроизведение завершено (бип или записанный голос)
+    elif event_type == "PlaybackFinished":
+        playback_data = event.get("playback", {})
+        media_uri = playback_data.get("media_uri", "")
+        target_uri = playback_data.get('target_uri', '')
+        channel_id = target_uri.split("channel:")[1]
+        #start_recording(channel_id)
+        # or
+        detect_speech(channel_id)
+    # Сценарий N: Абонент повесил трубку -> Удаляем файл
+    elif event_type == "StasisEnd":
+        channel_id = event.get("channel", {}).get("id")
+        recording_name = f"rec_{channel_id}"
+        print(f"Звонок {channel_id} завершен абонентом. Удаляем файл записи...")
+        delete_url = f"{BASE_URL}/recordings/stored/{recording_name}"
+        requests.delete(delete_url, auth=AUTH)
+def on_error(ws, error):
+    print(f"Произошла ошибка: {error}")
+wsapp = websocket.WebSocketApp(
+    f"ws://localhost:8088/ari/events?api_key={ARI_USER}:{ARI_PASS}&app=my-first-app",
+    on_message=on_message,
+    on_error=on_error
+)
+wsapp.run_forever()
+</code>
+==== ari_chat.py ====
+<code>
+import json
+import requests
+import websocket
+import os
+ARI_USER = "asterisk"
+ARI_PASS = "asterisk"
+BASE_URL = "http://localhost:8088/ari"
+AUTH = (ARI_USER, ARI_PASS)
+REC_PATH = "/var/spool/asterisk/recording/"
+SYSTEM_PROMPT = (
+    "Ты — ассистент.\n"
+    "Отвечай кратко.\n"
+    "Не спрашивай когда напомнить.\n"
+)
+COMPRESS_PROMPT = (
+    "Ты — инструмент оптимизации контекста. Проанализируй историю диалога. "
+    "Удали выполненные дела, приветствия и воду. Сформулируй ОДНО лаконичное "
+    "сообщение от лица пользователя, в котором отражена только актуальная суть "
+    "и невыполненные задачи/оставшиеся вопросы на текущий момент."
+)
+# Структура: {channel_id: [{"role": "user", "text": "..."}, {"role": "assistant", "text": "..."}]}
+channel_contexts = {}
+channel_caller_numbers = {}
+def get_channel_context(channel_id):
+    if channel_id not in channel_contexts:
+        channel_contexts[channel_id] = []
+    return channel_contexts[channel_id]
+def update_channel_context(channel_id, user_request, agent_response):
+    context = get_channel_context(channel_id)
+    context.append({"role": "user", "text": user_request})
+    context.append({"role": "assistant", "text": agent_response})
+    return context
+def load_context(caller_number):
+    file_path = f"{REC_PATH}{caller_number}.json"
+    if os.path.exists(file_path):
+        with open(file_path, 'r', encoding='utf-8') as f:
+            return json.load(f)
+    return []
+def save_context(caller_number, context):
+    file_path = f"{REC_PATH}{caller_number}.json"
+    with open(file_path, 'w', encoding='utf-8') as f:
+        json.dump(context, f, ensure_ascii=False, indent=2)
+def start_recording(channel_id):
+    print(f"Включаем запись для канала {channel_id} на 29 секунд...")
+    record_url = f"{BASE_URL}/channels/{channel_id}/record"
+    recording_name = f"rec_{channel_id}"
+    record_payload = {
+        "name": recording_name,
+        "format": "wav",
+        "ifExists": "overwrite",
+        "maxDuration": 29,
+        "terminateOn": "#",
+        "beep": True,
+    }
+    requests.post(record_url, params=record_payload, auth=AUTH)
+def get_caller_number(channel_id):
+    url = f"{BASE_URL}/channels/{channel_id}"
+    response = requests.get(url, auth=AUTH, timeout=5)
+    channel_info = response.json()
+    caller_number = channel_info.get("caller", {}).get("number")
+    return caller_number
+def on_message(wsapp, message):
+    event = json.loads(message)
+    event_type = event.get("type")
+    # Сценарий 1: Звонок поступил
+    if event_type == "StasisStart":
+        channel_id = event.get("channel", {}).get("id")
+        print(f"Отвечаем на звонок {channel_id}")
+        requests.post(f"{BASE_URL}/channels/{channel_id}/answer", auth=AUTH)
+        channel_caller_numbers[channel_id]=get_caller_number(channel_id)
+        channel_contexts[channel_id] = load_context(channel_caller_numbers[channel_id])
+        start_recording(channel_id)
+    # Сценарий 2: Запись вопроса завершилась -> Проигрываем ответ
+    elif event_type == "RecordingFinished":
+        recording_data = event.get("recording", {})
+        recording_name = recording_data.get("name")
+        target_uri = recording_data.get("target_uri", "")
+        channel_id = target_uri.split("channel:")[1]
+        caller_number = channel_caller_numbers[channel_id]
+        print(f"Номер звонящего в RecordingFinished: {caller_number}")
+        import wav_and_ogg
+        import speech_and_text
+        wav_and_ogg.wav_to_ogg(f"{REC_PATH}{recording_name}")
+        user_request = speech_and_text.speech_to_text(f"{REC_PATH}{recording_name}.ogg")
+        print(f"user_request: {user_request}")
+        if user_request:
+            context = get_channel_context(channel_id)
+            import request_to_agent
+            agent_response = request_to_agent.request_to_agent(user_request, SYSTEM_PROMPT, context)
+            #agent_response = user_request
+            print(agent_response)
+            update_channel_context(channel_id, user_request, agent_response)
+            print("Текущий контекст:")
+            from pprint import pprint
+            pprint(channel_contexts[channel_id])
+            speech_and_text.text_to_speech(agent_response,f"{REC_PATH}{recording_name}.response.ogg")
+            wav_and_ogg.ogg_to_wav(f"{REC_PATH}{recording_name}.response")
+            print(f"Запись {recording_name} готова! Проигрываем её обратно в канал {channel_id}...")
+            play_url = f"{BASE_URL}/channels/{channel_id}/play"
+            play_payload = {
+                #"media": f"recording:{recording_name}"
+                "media": f"recording:{recording_name}.response"
+            }
+            response = requests.post(play_url, params=play_payload, auth=AUTH)
+            print("Ответ ARI на воспроизведение:", response.status_code)
+    # Сценарий 3: Воспроизведение завершено
+    elif event_type == "PlaybackFinished":
+        playback_data = event.get("playback", {})
+        media_uri = playback_data.get("media_uri", "")
+        target_uri = playback_data.get('target_uri', '')
+        channel_id = target_uri.split("channel:")[1]
+        start_recording(channel_id)
+    # Сценарий 4: Абонент повесил трубку
+    elif event_type == "StasisEnd":
+        channel_id = event.get("channel", {}).get("id")
+        context = get_channel_context(channel_id)
+        caller_number = channel_caller_numbers[channel_id]
+        print(f"Номер звонящего в StasisEnd: номер {caller_number} канал {channel_id}")
+        print(f"Сжатие контекста")
+        import request_to_agent
+        agent_response = request_to_agent.request_to_agent("Сожми контекст", COMPRESS_PROMPT, context)
+        print(agent_response)
+        context = [{"role": "user", "text": agent_response}]
+        print(f"Сохранение контекста и удаление медиа файлов")
+        save_context(caller_number,context)
+        import glob
+        recording_name = f"rec_{channel_id}"
+        file_mask = f"{REC_PATH}{recording_name}*"
+        matched_files = glob.glob(file_mask)
+        for file_path in matched_files:
+            os.remove(file_path)
+            print(f"Файл удален: {file_path}")
+def on_error(ws, error):
+    print(f"Произошла ошибка: {error}")
+wsapp = websocket.WebSocketApp(
+    f"ws://localhost:8088/ari/events?api_key={ARI_USER}:{ARI_PASS}&app=my-first-app",
+    on_message=on_message,
+    on_error=on_error
+)
+wsapp.run_forever()
+</code>
+==== ari_chat_talk_detect.py ====
+<code>
+import os
+import json
+import requests
+import websocket
+ARI_USER = "asterisk"
+ARI_PASS = "asterisk"
+BASE_URL = "http://localhost:8088/ari"
+AUTH = (ARI_USER, ARI_PASS)
+REC_PATH = "/var/spool/asterisk/recording/"
+SYSTEM_PROMPT = (
+    "Ты — ассистент.\n"
+    "Отвечай кратко.\n"
+    "Не спрашивай когда напомнить.\n"
+)
+COMPRESS_PROMPT = (
+    "Ты — инструмент оптимизации контекста. Проанализируй историю диалога. "
+    "Удали выполненные дела, приветствия и воду. Сформулируй ОДНО лаконичное "
+    "сообщение от лица пользователя, в котором отражена только актуальная суть "
+    "и невыполненные задачи/оставшиеся вопросы на текущий момент."
+)
+# Структура: {channel_id: [{"role": "user", "text": "..."}, {"role": "assistant", "text": "..."}]}
+channel_contexts = {}
+channel_caller_numbers = {}
+def load_context(caller_number):
+    file_path = f"{REC_PATH}{caller_number}.json"
+    if os.path.exists(file_path):
+        with open(file_path, 'r', encoding='utf-8') as f:
+            return json.load(f)
+    return []
+def save_context(caller_number, context):
+    file_path = f"{REC_PATH}{caller_number}.json"
+    with open(file_path, 'w', encoding='utf-8') as f:
+        json.dump(context, f, ensure_ascii=False, indent=2)
+def get_caller_number(channel_id):
+    url = f"{BASE_URL}/channels/{channel_id}"
+    response = requests.get(url, auth=AUTH, timeout=5)
+    channel_info = response.json()
+    caller_number = channel_info.get("caller", {}).get("number")
+    return caller_number
+def get_channel_context(channel_id):
+    if channel_id not in channel_contexts:
+        channel_contexts[channel_id] = []
+    return channel_contexts[channel_id]
+def update_channel_context(channel_id, user_request, agent_response):
+    context = get_channel_context(channel_id)
+    context.append({"role": "user", "text": user_request})
+    context.append({"role": "assistant", "text": agent_response})
+    return context
+def detect_speech(channel_id):
+    print(f"Запускаем TALK_DETECT для канала {channel_id}...")
+    talk_url = f"{BASE_URL}/channels/{channel_id}/variable"
+    # 1000 мс тишины для завершения фразы, 500 мс речи для начала
+    talk_payload = {
+        "variable": "TALK_DETECT(set)",
+        "value": "1000,500"
+    }
+    response = requests.post(talk_url, json=talk_payload, auth=AUTH)
+    print(f"Статус-код ответа: {response.status_code}")
+    print(f"Текст ответа: {response.text}")
+def undetect_speech(channel_id):
+    print(f"Отключаем TALK_DETECT для канала {channel_id}...")
+    talk_url = f"{BASE_URL}/channels/{channel_id}/variable"
+    talk_payload = {
+        "variable": "TALK_DETECT(set)",
+        "value": "remove"
+    }
+    response = requests.post(talk_url, json=talk_payload, auth=AUTH)
+    print(f"Статус-код ответа: {response.status_code}")
+    print(f"Текст ответа: {response.text}")
+def play_beep(channel_id):
+    print(f"Проигрываем сигнал перед записью в канал {channel_id}...")
+    play_url = f"{BASE_URL}/channels/{channel_id}/play"
+    play_payload = {"media": "sound:beep"}
+    requests.post(play_url, params=play_payload, auth=AUTH)
+def start_recording(channel_id):
+    print(f"Включаем запись для канала {channel_id}")
+    record_url = f"{BASE_URL}/channels/{channel_id}/record"
+    recording_name = f"rec_{channel_id}"
+    record_payload = {
+        "name": recording_name,
+        "format": "wav",
+        "ifExists": "overwrite",
+        #"maxDuration": 29,
+        #"terminateOn": "#",
+        #"beep": True,
+    }
+    requests.post(record_url, params=record_payload, auth=AUTH)
+def stop_recording(channel_id):
+    print(f"Останавливаем запись для канала {channel_id}")
+    recording_name = f"rec_{channel_id}"
+    stop_url = f"{BASE_URL}/recordings/live/{recording_name}/stop"
+    requests.post(stop_url, auth=AUTH)
+def on_message(wsapp, message):
+    event = json.loads(message)
+    event_type = event.get("type")
+    # Сценарий 1: Звонок поступил -> Отвечаем и воспроизводим beep
+    if event_type == "StasisStart":
+        channel_id = event.get("channel", {}).get("id")
+        print(f"Отвечаем на звонок {channel_id}")
+        requests.post(f"{BASE_URL}/channels/{channel_id}/answer", auth=AUTH)
+        channel_caller_numbers[channel_id]=get_caller_number(channel_id)
+        channel_contexts[channel_id] = load_context(channel_caller_numbers[channel_id])
+        play_beep(channel_id)
+        #start_recording(channel_id)
+        # or
+        #detect_speech(channel_id)
+    # Сценарий N
+    elif event_type == "ChannelTalkingStarted":
+        channel_id = event.get("channel", {}).get("id")
+        print(f"Обнаружена речь в канале {channel_id}! Начинаем запись...")
+        start_recording(channel_id)
+    # Сценарий N
+    elif event_type == "ChannelTalkingFinished":
+        channel_id = event.get("channel", {}).get("id")
+        print(f"Обнаружено молчание в канале {channel_id}")
+        stop_recording(channel_id)
+        undetect_speech(channel_id)
+    # Сценарий N: Запись завершилась -> Проигрываем её обратно
+    elif event_type == "RecordingFinished":
+        recording_data = event.get("recording", {})
+        recording_name = recording_data.get("name")
+        target_uri = recording_data.get("target_uri", "")
+        channel_id = target_uri.split("channel:")[1]
+        import wav_and_ogg
+        import speech_and_text
+        wav_and_ogg.wav_to_ogg(f"{REC_PATH}{recording_name}")
+        user_request = speech_and_text.speech_to_text(f"{REC_PATH}{recording_name}.ogg")
+        print(f"user_request: {user_request}")
+        if user_request:
+            context = get_channel_context(channel_id)
+            import request_to_agent
+            agent_response = request_to_agent.request_to_agent(user_request, SYSTEM_PROMPT, context)
+            #agent_response = user_request
+            print(agent_response)
+            update_channel_context(channel_id, user_request, agent_response)
+            print("Текущий контекст:")
+            from pprint import pprint
+            pprint(channel_contexts[channel_id])
+            speech_and_text.text_to_speech(agent_response,f"{REC_PATH}{recording_name}.response.ogg")
+            wav_and_ogg.ogg_to_wav(f"{REC_PATH}{recording_name}.response")
+            print(f"Запись {recording_name} готова! Проигрываем её обратно в канал {channel_id}...")
+            play_url = f"{BASE_URL}/channels/{channel_id}/play"
+            play_payload = {
+                #"media": f"recording:{recording_name}"
+                "media": f"recording:{recording_name}.response"
+            }
+            response = requests.post(play_url, params=play_payload, auth=AUTH)
+            print("Ответ ARI на воспроизведение:", response.status_code)
+    # Сценарий N: Воспроизведение завершено (бип или записанный голос)
+    elif event_type == "PlaybackFinished":
+        playback_data = event.get("playback", {})
+        media_uri = playback_data.get("media_uri", "")
+        target_uri = playback_data.get('target_uri', '')
+        channel_id = target_uri.split("channel:")[1]
+        #start_recording(channel_id)
+        # or
+        #detect_speech(channel_id)
+        if media_uri == "sound:beep":
+            print(f"Сигнал завершен. Начинаем запись для канала {channel_id}...")
+            detect_speech(channel_id)
+        elif "recording:rec_" in media_uri:
+            print(f"Воспроизведение записи завершено. Идем на новый круг для канала {channel_id}...")
+            play_beep(channel_id)
+    # Сценарий N: Абонент повесил трубку -> Удаляем файл
+    elif event_type == "StasisEnd":
+        channel_id = event.get("channel", {}).get("id")
+        #print(f"Звонок {channel_id} завершен абонентом. Удаляем файл записи...")
+        #recording_name = f"rec_{channel_id}"
+        #delete_url = f"{BASE_URL}/recordings/stored/{recording_name}"
+        #requests.delete(delete_url, auth=AUTH)
+        context = get_channel_context(channel_id)
+        caller_number = channel_caller_numbers[channel_id]
+        print(f"Номер звонящего в StasisEnd: номер {caller_number} канал {channel_id}")
+        print(f"Сжатие контекста")
+        import request_to_agent
+        agent_response = request_to_agent.request_to_agent("Сожми контекст", COMPRESS_PROMPT, context)
+        print(agent_response)
+        context = [{"role": "user", "text": agent_response}]
+        print(f"Сохранение контекста и удаление медиа файлов")
+        save_context(caller_number,context)
+        import glob
+        recording_name = f"rec_{channel_id}"
+        file_mask = f"{REC_PATH}{recording_name}*"
+        matched_files = glob.glob(file_mask)
+        for file_path in matched_files:
+            os.remove(file_path)
+            print(f"Файл удален: {file_path}")
+def on_error(ws, error):
+    print(f"Произошла ошибка: {error}")
+wsapp = websocket.WebSocketApp(
+    f"ws://localhost:8088/ari/events?api_key={ARI_USER}:{ARI_PASS}&app=my-first-app",
+    on_message=on_message,
+    on_error=on_error
+)
+wsapp.run_forever()
+</code>
+==== Черновик ====

Методические материалы Лохтурова Вячеслава

User Tools

Site Tools

Differences

Page Tools