import websockets
import asyncio
import logging
# Настройка логирования
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
async def listen():
url = "ws://simple-websocket-server-echo.glitch.me/"
logging.info("Подключение к серверу WebSocket...")
async with websockets.connect(url) as ws:
logging.info("Успешно подключено к серверу.")
while True:
msg = await ws.recv() # Ожидаем получение сообщения
logging.info(f"Получено сообщение: {msg}") # Логируем сообщение
async def main():
# Создание задачи для прослушивания
listener_task = asyncio.create_task(listen())
# Здесь вы можете добавить другие задачи параллельно, если необходимо
await listener_task # Ожидаем завершения задачи, чтобы программа не завершилась
# Запуск основной функции
asyncio.run(main())import uuid # Библиотека для генерации RqUID
# Получаем эти данные на странице разработчиков Сбера
client_id = 'xxxxx-xxxxx-xxxxx-xxxxx-xxxxx'
secret = 'xxxxx-xxxxx-xxxxx-xxxxx-xxxxx'
# Данные авторизации пользователя
auth = 'MDk4N2Y4ZTxxxxxxxxxxxxxxxDQ0Nw=='
# Генерируем токен
def get_token(user_auth):
'''
Функция генерации токена
user_auth - данные авторизации пользователя
'''
rq_uid = str(uuid.uuid4())
# эндпоинт GigaChat для генерации токена
url = "https://ngw.devices.sberbank.ru:9443/api/v2/oauth"
payload='scope=GIGACHAT_API_PERS'
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': 'application/json',
'RqUID': rq_uid,
'Authorization': f'Basic {user_auth}'
}
response = requests.request("POST", url, headers=headers, data=payload)
token = response.json()["access_token"]
# print(token)
return tokenЧитайте про позиционные (кортежи) и именованные (словари) аргументы.
def func(*args, **qwargs):
if args:
sum = 0
for i in args:
sum += i
print(sum)
if qwargs:
sum2 = 0
for v in qwargs.values():
sum2 += v
print(sum2)
func(1, 2) #3
func(a=1, b=2) #3