Differences

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

--- язык_программирования_python [2026/01/12 12:17]
val [Язык программирования Python]
+++ язык_программирования_python [2026/07/13 15:52] (current)
val [ari_chat.py]
@@ Line 1: / Line 1: @@
 ====== Язык программирования Python ======
+  * [[Все что должен знать DevOps про Python]]
   * [[https://habr.com/ru/articles/277679/|Пишем shell скрипты на Python и можно ли заменить им Bash]]
@@ Line 32: / Line 34: @@
 <code>
 (venv1) # python
+</code><code>
+>>> import os
+>>> print ("Hello " + os.environ.get('USER') + "!")
 </code>
@@ Line 112: / Line 117: @@
 (venv1) :~/pywebd$ #export PYWEBD_PORT=4443
 (venv1) :~/pywebd$ #export PYWEBD_DOC_ROOT=/var/www/html/
 (venv1) :~/pywebd$ #PYWEBD_PORT=4080 PYWEBD_DOC_ROOT=/var/www/ python app.py
@@ Line 120: / Line 126: @@
 (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>
@@ Line 378: / Line 401: @@
 </code>
+==== wav_and_ogg.py ====
+<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>
+==== speech_and_text.py ====
+<code>
+import os
+import sys
+import requests
+API_KEY = os.getenv("API_KEY")
+FOLDER_ID = os.getenv("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("FOLDER_ID"),
+        auth=os.getenv("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>
+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"
+)
+# Структура: {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} на 5 секунд...")
+    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": 5,
+        "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(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)
+        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"Сохранение контекста и удаление медиа файлов")
+        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