• Пишем shell скрипты на Python и можно ли заменить им Bash

• Basic Web Frontend/Backend Demo
• Использование Celery с Django для обработки фоновых задач

• Что такое venv и virtualenv в Python, и как их использовать
• Install packages in a virtual environment using pip and venv

• Poetry — прекрасная альтернатива pip (шпаргалка)
• UV. Обзор пакетного менеджера Python

# python3 -V

# apt install python3-venv

$ python3 -m venv ~/venv1

$ source ~/venv1/bin/activate

(venv1) $ which python
(venv1) $ which ansible

(venv1) $ deactivate

(venv1) # python

>>> import os
>>> print ("Hello " + os.environ.get('USER') + "!")

• https://ru.wikipedia.org/wiki/IPython

(venv1) # pip install ipython

• try-jupyter/notebooks
• Pyplot tutorial

• Сравнение WSGI-серверов:uvicorn и Gunicorn

• Виртуальная среда Python

• Python декораторы на максималках. Универсальный рецепт по написанию и аннотированию от мала до велика
• Publishing a Directory with Flask

• Создание Web-приложения Flask и деплой с помощью Docker Compose & Dockerfile
• Running Your Flask Application Over HTTPS
• configparser — Configuration file parser

(venv1) :~$ pip install Flask

(venv1) :~$ mkdir -p pywebd/; cd $_

(venv1) :~/pywebd$ cat app.py

from flask import Flask, send_from_directory
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__)

@app.route('/')
def index():
   return send_from_directory(pywebd_doc_root, 'index.html')

@app.route('/<path:path>')
def sendstuff(path):
        print(path)
        return send_from_directory(pywebd_doc_root, path)

if __name__ == "__main__":

  if 'PYWEBD_PORT' in os.environ: pywebd_port = os.environ.get('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)

# mkdir -p /etc/pywebd/

# cat /etc/pywebd/pywebd.conf

[default]
DocumentRoot = /var/www/
Listen = 4080

# #cp /root/wild.crt /etc/pywebd/pywebd.crt
# #cp /root/wild.key /etc/pywebd/pywebd.key

• Ресурсы Web сервера на shell

(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

(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

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

(venv1) :~/pywebd$ # gunicorn -c gunicorn.conf.py app:app

• https://python-poetry.org/docs/basic-usage/
• Poetry — прекрасная альтернатива pip (шпаргалка)

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

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)

/pywebd2 # cat .env

PYWEBD_MESSAGE="Hello World!!!"

• Тестируем на Python: unittest и pytest. Инструкция для начинающих

• Регулярные выражения в Python от простого к сложному. Подробности, примеры, картинки, упражнения
• Online Regular expression tester with syntax highlighting

/pywebd2 # cat tests/test_ver_format.py

import app, re

assert re.match(r'^ver\d+\.\d+$', app.CONST_VER),"Incorrect VER format"

/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

• TypeError: __call__() missing 1 required positional argument: 'send'

/pywebd2 # ###poetry run gunicorn app:app --bind 0.0.0.0:6080 -k uvicorn.workers.UvicornWorker

• Running Python File in Terminal

• Модуль subprocess

• Понимание итераторов в Python
• Find dictionary matching value in list - Python
• Функция next() в Python, следующий элемент итератора

gate# apt install python3-xmltodict/stable

gate# cat /etc/zabbix/dhcp-pools.py

#!/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])

# cat /usr/local/bin/asterisk.chansip.discovery.py

#!/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))

Вариант от DeepSeek

# cat /usr/local/bin/asterisk.chansip.discovery.ds.py

#!/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))

• 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

(venv1) server:~# pip install requests

(venv1) server:~# cat zab_set_map_name.py

#!/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"
UNAME = "Admin"
PWORD = "zabbix"

print("\nLogin user {} to Zabbiz API".format(UNAME))
r = requests.post(ZABBIX_API_URL,
                  json={
                      "jsonrpc": "2.0",
                      "method": "user.login",
                      "params": {
                          "username": UNAME,
                          "password": PWORD},
                      "id": 1
                  })

print(json.dumps(r.json(), indent=4, sort_keys=True))

AUTHTOKEN = r.json()["result"]

#print("Rename Map with ID", MAPID, "to", MAPNAME)
print(f"Rename Map with ID {MAPID} to {MAPNAME}")

r = requests.post(ZABBIX_API_URL,
                  headers={'Authorization': 'Bearer ' + AUTHTOKEN},
                  json={
                    "jsonrpc": "2.0",
                    "method": "map.update",
                    "params": {
                        "sysmapid": MAPID,
                        "name": MAPNAME
                    },
                    "id": 2
                  })

print(json.dumps(r.json(), indent=4, sort_keys=True))

print("\nLogout user")
r = requests.post(ZABBIX_API_URL,
                  headers={'Authorization': 'Bearer ' + AUTHTOKEN},
                  json={
                      "jsonrpc": "2.0",
                      "method": "user.logout",
                      "params": {},
                      "id": 2,
                  })

print(json.dumps(r.json(), indent=4, sort_keys=True))

(venv1) server:~# chmod +x zab_set_map_name.py

(venv1) server:~# ./zab_set_map_name.py 2 "ISP 1"

• 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/

• SeanPesce/https_server.py

dir$ python3 -m http.server 80

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)

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)

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})

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 = {}

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 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",
        "maxDuration": 5,
        "terminateOn": "#",
        "ifExists": "overwrite"
    }
    requests.post(record_url, params=record_payload, auth=AUTH)

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)

        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]

        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)

        import request_to_agent
        context = get_channel_context(channel_id)
        agent_response = request_to_agent.request_to_agent(user_request, SYSTEM_PROMPT, context)
        print(agent_response)

        #agent_response = user_request

        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", {})
        target_uri = playback_data.get('target_uri', '')
        channel_id = target_uri.split("channel:")[1]

        print(f"Воспроизведение завершено. Возвращаемся в режим записи...")
        start_recording(channel_id)

        '''
        print(f"Вешаем трубку для channel_id {channel_id}")
        hangup_url = f"{BASE_URL}/channels/{channel_id}"
        requests.delete(hangup_url, auth=AUTH)
        '''

    elif event_type == "StasisEnd":
        channel_id = event.get("channel", {}).get("id")
        print(f"Звонок завершен пользователем (Канал {channel_id}). Очистка ресурсов...")
        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()