Kodduu Python
1.02K subscribers
319 photos
29 videos
212 links
Научись программировать на Python на интересных примерах

Самый быстрый курс https://stepik.org/a/187914
Самый нескучный курс https://stepik.org/a/185238

Во вопросам сотрудничества: @AlexErf
Download Telegram
🎮 Разработка многопользовательской игры на Flask и WebSockets


from flask import Flask, render_template, request
from flask_socketio import SocketIO, emit

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your_secret_key'
socketio = SocketIO(app)

players = set()

@app.route('/')
def index():
return render_template('index.html')

@socketio.on('connect')
def handle_connect():
players.add(request.sid)
emit('player_count', {'count': len(players)}, broadcast=True)

@socketio.on('disconnect')
def handle_disconnect():
players.discard(request.sid)
emit('player_count', {'count': len(players)}, broadcast=True)

@socketio.on('move')
def handle_move(data):
emit('move', data, broadcast=True, include_self=False)

if __name__ == '__main__':
socketio.run(app, debug=True)

# Пример использования: создайте файл index.html с минимальной логикой подключения


📌 Этот код создаёт базовую структуру для многопользовательской игры, используя Flask и WebSockets. Сервер обрабатывает подключение игроков и синхронизирует их действия в реальном времени. Подобный подход можно использовать для создания простых интерактивных игр и приложений с обменом данных между клиентами.

Подпишись 👉🏻 @KodduuPython 🤖
📊 Визуализация данных IoT в реальном времени с Plotly


import random
import time
import datetime
import plotly.graph_objects as go
from plotly.subplots import make_subplots

def get_sensor_data() -> float:
"""Имитирует получение данных с датчика."""
return random.uniform(20.0, 25.0)

def update_plot(fig, x_data, y_data):
"""Обновляет график новыми данными."""
fig.data[0].x = x_data
fig.data[0].y = y_data
fig.update_layout(transition_duration=500)
fig.show()

def visualize_iot_data():
"""Визуализирует данные с IoT-устройств в реальном времени."""
fig = make_subplots(rows=1, cols=1)
fig.add_trace(go.Scatter(x=[], y=[], mode='lines+markers', name='Температура'))

x_data, y_data = [], []

try:
while True:
x_data.append(datetime.datetime.now().strftime('%H:%M:%S'))
y_data.append(get_sensor_data())
update_plot(fig, x_data, y_data)
time.sleep(1)
except KeyboardInterrupt:
print("Визуализация остановлена")
except Exception as e:
print(f"Ошибка: {e}")

# Пример использования
visualize_iot_data()


📌 Этот код демонстрирует, как собирать и визуализировать данные с IoT-устройств в реальном времени. Он использует Plotly для создания динамического графика, обновляющегося каждую секунду с новыми данными датчика. Такой подход удобен для мониторинга параметров в реальном времени.

Подпишись 👉🏻 @KodduuPython 🤖
2
Forwarded from AIGENTTO
Отказ от Cursor 🙅‍♂️

Луну обогнули 🌔, вернёмся на землю.

Я писал про кейс перевода всех сайтов на OpenClaw. Теперь переводим разработку больших продуктов.

Было так: Cursor пишет код как ассистент разработчика. Говоришь ему "построй план", корректируешь, и с нуля быстро можешь написать довольно масштабную систему.

После этого начинается просто корректировка, новые фичи и фиксы багов. И как правило, в большой системе, обросшей фичами, внедрение нового занимает время, особенно если команда не Full Stack (отдельно Front и отдельно Back разработка).

Но я попробовал и вынес дальнейшую разработку двух больших систем в OpenClaw.

По сути, на VPS с OpenClaw (полный sudo доступ) подтягиваем нужные репозитории. И в чате ему говорим: в репо A добавь X, почини Y, найди баги. Вроде все то же самое, как и в Cursor, но нет, все мгновенно оказывается в Prod, и в цикле там уже включено UI тестирование, потому что есть playwright + browser.

Dev Pipeline теперь это не 3 отдельные среды (prod + test + dev), а Front+Back Dev + QA + DevOps со средним time-2-market в 3 недели, а просто три ветки prod/test/dev, все три постоянно подняты, и time-2-market падает до часов.

Этап дизайна в Figma тоже уходит. Зачем просить дизайнера, который все время занят, сделать новый дизайн, если можно просто сказать боту: "Накидай три версии дизайна новой фичи сразу с самой фичей прямо в dev-ветке", и уже через 5 минут смотреть, выбирать и корректировать.

Написание больших систем с нуля все еще удобнее в Cursor, но дальнейшее развитие работает на OpenClaw на 4+.

Напомню, мы используем свою оптимизированную версию OpenClaw - DarwinClaw, которая скоро должна стать доступной для всех.

#КЕЙС

Подпишись 👉🏻 @aigentto 🤖
Создание AI для NPC с алгоритмом A*


from heapq import heappush, heappop
from typing import List, Tuple

def heuristic(a: Tuple[int, int], b: Tuple[int, int]) -> float:
# Вычисляем манхэттенское расстояние
return abs(a[0] - b[0]) + abs(a[1] - b[1])

def a_star_search(start: Tuple[int, int], goal: Tuple[int, int], grid: List[List[int]]) -> List[Tuple[int, int]]:
open_set = []
heappush(open_set, (0, start))
came_from = {}
g_score = {start: 0}
f_score = {start: heuristic(start, goal)}

while open_set:
_, current = heappop(open_set)

if current == goal:
path = []
while current in came_from:
path.append(current)
current = came_from[current]
path.append(start) # Добавляем начальную позицию в путь
return path[::-1]

for dx, dy in [(-1, 0), (1, 0), (0, -1), (0, 1)]:
neighbor = (current[0] + dx, current[1] + dy)
tentative_g_score = g_score[current] + 1

if 0 <= neighbor[0] < len(grid) and 0 <= neighbor[1] < len(grid[0]) and grid[neighbor[0]][neighbor[1]] == 0:
if neighbor not in g_score or tentative_g_score < g_score[neighbor]:
came_from[neighbor] = current
g_score[neighbor] = tentative_g_score
f_score[neighbor] = tentative_g_score + heuristic(neighbor, goal)
if neighbor not in [i[1] for i in open_set]:
heappush(open_set, (f_score[neighbor], neighbor))

return []

# Пример использования
grid = [
[0, 1, 0, 0, 0],
[0, 1, 0, 1, 0],
[0, 0, 0, 1, 0],
[0, 1, 1, 0, 0],
[0, 0, 0, 0, 0]
]
start = (0, 0)
goal = (4, 4)
path = a_star_search(start, goal, grid)
print("Путь:", path)


📌 Алгоритм A* используется для поиска оптимального пути на карте, что делает его идеальным для реализации AI в играх.

Подпишись 👉🏻 @KodduuPython 🤖
☁️ Микросервис для прогноза погоды на Flask и OpenWeatherMap API


from flask import Flask, jsonify, request
import requests
import os

app = Flask(__name__)

API_KEY = os.getenv('OPENWEATHERMAP_API_KEY') # Получение ключа API из переменных окружения
BASE_URL = 'http://api.openweathermap.org/data/2.5/weather'

def get_weather_data(city: str) -> dict:
try:
response = requests.get(BASE_URL, params={'q': city, 'appid': API_KEY, 'units': 'metric'})
response.raise_for_status()
return response.json()
except requests.HTTPError as http_err:
return {'error': f'HTTP error occurred: {http_err}'}
except Exception as err:
return {'error': f'Other error occurred: {err}'}

@app.route('/weather', methods=['GET'])
def weather():
city = request.args.get('city')
if not city:
return jsonify({'error': 'Город не указан'}), 400

weather_data = get_weather_data(city)
if 'error' in weather_data:
return jsonify(weather_data), 500

return jsonify({
'city': city,
'temperature': weather_data['main']['temp'],
'description': weather_data['weather'][0]['description']
})

if __name__ == '__main__':
app.run(debug=True)


📌 Этот код создает микросервис на Flask, который предоставляет прогноз погоды по заданному городу, используя API OpenWeatherMap. Запросы к API обрабатываются с учетом ошибок, а данные возвращаются в виде JSON. Используйте переменные окружения для хранения ключа API, чтобы повысить безопасность.

Подпишись 👉🏻 @KodduuPython 🤖
🔗 Композирование функций для сложных операций


from typing import Callable

def compose(*functions: Callable) -> Callable:
def composed_function(x):
for func in reversed(functions): # Применение функций в обратном порядке
x = func(x)
return x
return composed_function

# Пример простых функций
def double(x: int) -> int:
return x * 2

def increment(x: int) -> int:
return x + 1

# Композирование функций
process = compose(double, increment)

# Пример использования
result = process(3) # Результат: (3 + 1) * 2 = 8
print(result)


📌 Композирование позволяет объединять простые функции в более сложные, улучшая читаемость и поддержку кода. В данном примере функции double и increment комбинируются в цепочку, применяемую к значению 3. Это особенно полезно в обработке данных, где часто требуется выполнить серию преобразований.

Подпишись 👉🏻 @KodduuPython 🤖
1
⚡️ Ускоряем Python с Cython


# Пример Cython кода для ускорения вычислений
def sum_of_squares(int n):
cdef int i
cdef long total = 0
for i in range(n):
total += i * i
return total

# Сохраните этот код в файле с расширением .pyx и скомпилируйте с помощью setup.py
# Пример setup.py для компиляции

from setuptools import setup
from Cython.Build import cythonize

setup(
ext_modules=cythonize("example.pyx")
)

# Выполните команду: python setup.py build_ext --inplace
# Пример использования
import example

result = example.sum_of_squares(1000000)
print(result)


📌 Cython позволяет преобразовывать Python-код в C-код, что значительно ускоряет выполнение ресурсоемких операций. Этот пример показывает, как использовать Cython для оптимизации функции, суммирующей квадраты чисел. Преобразование и компиляция кода выполняются легко, предоставляя значительный прирост производительности для вычислительно интенсивных задач.

🛠 pip install cython

Подпишись 👉🏻 @KodduuPython 🤖
1
⚡️ Ускорение Python-кода с помощью Cython


# Пример использования Cython для ускорения вычислений
# Сначала создайте файл example.pyx с этим содержимым:

def calculate_sum(n: int) -> int:
cdef int i, total = 0
for i in range(n):
total += i
return total

# Затем создайте setup.py для компиляции:

from setuptools import setup
from Cython.Build import cythonize

setup(
ext_modules=cythonize("example.pyx"),
)

# Для компиляции выполните:
# python setup.py build_ext --inplace

# Пример использования скомпилированного модуля:
import example

result = example.calculate_sum(1000000)
print(result)


📌 Cython позволяет компилировать Python-код в C, значительно увеличивая его производительность. В этом примере мы используем Cython для ускорения функции суммирования чисел. Подобный подход пригодится в проектах, где критически важна скорость выполнения, например, в научных вычислениях или обработке больших объемов данных.

🛠 pip install Cython

Подпишись 👉🏻 @KodduuPython 🤖
📊 Оптимизация базы данных: ORM vs. прямые SQL-запросы


import sqlite3
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

# Настройка для прямых SQL-запросов
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()
cursor.execute('CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)')

# Настройка для ORM с использованием SQLAlchemy
Base = declarative_base()

class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

def add_user_sql(name: str):
try:
cursor.execute('INSERT INTO users (name) VALUES (?)', (name,))
conn.commit()
except sqlite3.Error as e:
print(f"SQL Error: {e}")

def add_user_orm(name: str):
try:
user = User(name=name)
session.add(user)
session.commit()
except Exception as e:
session.rollback()
print(f"ORM Error: {e}")

# Пример использования
add_user_sql('Alice')
add_user_orm('Bob')

# Закрытие соединений
cursor.close()
conn.close()
session.close()


📌 В этом примере показано использование прямых SQL-запросов и ORM для добавления записей в базу данных SQLite. Прямые SQL-запросы дают больше контроля и могут быть быстрее, но менее удобны. ORM, как SQLAlchemy, упрощает работу с базой данных, автоматизируя многие процессы, но может добавить некоторую задержку из-за абстракции.

Подпишись 👉🏻 @KodduuPython 🤖
Forwarded from AIGENTTO
Полезный тул от NVIDIA для penetration тестирования вашего ИИ бота

Гоняет всякие промты, чтобы найти дыры (персональные данные и прочее) в вашем AI боте:
https://github.com/NVIDIA/garak/

Подпишись 👉🏻 @aigentto 🤖
🔐 Замыкания: Управление состоянием без классов


def create_counter(start: int = 0):
count = start

def increment():
nonlocal count
count += 1
return count

return increment

# Пример использования
counter = create_counter()
print(counter()) # Выведет: 1
print(counter()) # Выведет: 2
print(counter()) # Выведет: 3


📌 Замыкания позволяют создавать функции с собственным состоянием, что может быть полезно для управления счетчиками, кэшами и другими задачами, требующими сохранение состояния между вызовами. В приведенном примере замыкание create_counter создает функцию increment, которая помнит текущее значение счетчика. Это позволяет обойтись без использования классов, сохраняя код компактным и понятным.

Подпишись 👉🏻 @KodduuPython 🤖
🔒 Автоматизация сетевых операций с Paramiko


import paramiko
from paramiko import SSHException, AuthenticationException, BadHostKeyException

def execute_remote_command(hostname: str, username: str, password: str, command: str) -> str:
try:
with paramiko.SSHClient() as client:
client.set_missing_host_key_policy(paramiko.WarningPolicy())
client.connect(hostname, username=username, password=password)

stdin, stdout, stderr = client.exec_command(command)
result = stdout.read().decode()
return result
except AuthenticationException:
return "Ошибка аутентификации"
except BadHostKeyException:
return "Неправильный ключ хоста"
except SSHException as e:
return f"SSH ошибка: {e}"
except Exception as e:
return f"Ошибка: {e}"

# Пример использования
if __name__ == "__main__":
response = execute_remote_command(
hostname='192.168.1.100',
username='admin',
password='password',
command='ls'
)
print(response)


📌 Этот скрипт демонстрирует, как выполнять команды на удаленном сервере через SSH с помощью Paramiko. Он автоматически уведомляет о ключах хоста и обрабатывает ошибки подключения. Это полезно для автоматизации задач администрирования и управления серверами.

🛠 pip install paramiko

Подпишись 👉🏻 @KodduuPython 🤖
🧠 Быстрый старт с машинным обучением в Scikit-learn


from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

# Загрузка датасета
data = load_iris()
X, y = data.data, data.target

# Разделение данных на обучающую и тестовую выборки
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Создание и обучение модели
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)

# Предсказание и оценка модели
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)

print(f"Точность модели: {accuracy:.2f}")

# Обработка ошибок
try:
# Пример использования модели на новых данных
new_data = [[5.1, 3.5, 1.4, 0.2]]
prediction = model.predict(new_data)
print(f"Предсказанный класс: {prediction[0]}")
except Exception as e:
print(f"Ошибка при предсказании: {e}")


📌 Этот код демонстрирует, как быстро создать и обучить модель машинного обучения с использованием Scikit-learn. Мы используем датасет ирисов для классификации, обучая модель случайного леса и оцениваем её точность. Такой подход позволяет быстро прототипировать решения и интегрировать их в приложения, что делает Python предпочтительным инструментом для задач машинного обучения по сравнению с C++.

Подпишись 👉🏻 @KodduuPython 🤖
🧪 Быстрое тестирование кода с Pytest


# Установка Pytest
# 🛠 pip install pytest

# Функция для тестирования
def add(a: int, b: int) -> int:
return a + b

# Простой тест для функции add
def test_add():
assert add(2, 3) == 5
assert add(-1, 1) == 0
assert add(0, 0) == 0


📌 Pytest — это мощный инструмент для тестирования, который упрощает написание и выполнение тестов. В этом примере показана простая функция сложения и тесты для проверки её корректности. Pytest автоматически обнаруживает и запускает тесты, предоставляя детализированные отчеты об ошибках, что помогает быстро улучшать качество кода.

Подпишись 👉🏻 @KodduuPython 🤖
Forwarded from AIGENTTO
Мониторинг логов с автоматическим исправлением проблем

И так, раньше надо было использовать различные отдельные инструменты для мониторинга логов (Zabbix и прочее), чтобы в крайнем случае сделать reboot отдельных компонентов или всей системы.

Но оказалось, можно отдать задачу ReAct боту типа OpenClaw (мы используем версию B2B DarwinClaw). Он может проверять логи хоть каждые 2 минуты. Тонны токенов здесь не тратятся, он также просто делает поиск по ключевым словам. И важно, что он сам может подобрать ключевые слова и обновлять этот набор исходя из типа логов.

Дальше больше - он может не просто сделать reboot, он может исправить проблему прямо в коде, или сделать откат в git, если поймет, что это проблема новой версии 🔥

Ну то есть, по сути это полноценный support+разраб 24x7, работающий за миску токенов. Да, важно прописать ему некоторые правила (когда то-то делай так, когда это делай сяк), по сути, вашу политику работы с Prod.

#КЕЙС

Прочие полезные кейсы:
👉 WP умер, перевёл все сайты на DarwinClaw
👉 Сбор любой публичной статистики
👉 Отказ от Cursor IDE

Подпишись 👉🏻 @aigentto 🤖
### Чат-бот для психологической поддержки с использованием NLP 🤖


import random
import spacy

nlp = spacy.load("en_core_web_sm")

def get_response(user_input: str) -> str:
doc = nlp(user_input)

# Проверка на наличие ключевых лексем
for token in doc:
if token.lemma_ == "sad":
return random.choice([
"Мне жаль это слышать. Хочешь поговорить об этом?",
"Грустно слышать. Может, расскажешь подробнее?",
])
elif token.lemma_ == "happy":
return random.choice([
"Это замечательно! Что делает тебя счастливым?",
"Рад слышать, что ты счастлив!",
])

return "Расскажи, как ты себя чувствуешь?"

# Пример использования
def chat():
print("Чат-бот: Привет! Как ты себя чувствуешь сегодня?")
while True:
user_input = input("Вы: ")
if user_input.lower() in ["выход", "пока"]:
print("Чат-бот: Пока! Заботься о себе.")
break
print(f"Чат-бот: {get_response(user_input)}")

chat()


Этот чат-бот использует spaCy для анализа пользовательских сообщений. Он может определить эмоциональные состояния, такие как грусть или счастье, и предлагает соответствующие ответы. Это простой пример создания чат-бота, который может стать основой для более сложных систем взаимодействия.

Подпишись 👉🏻 @KodduuPython 🤖
🔒 Декораторы для улучшения безопасности API


from functools import wraps

def require_authentication(func):
@wraps(func)
def wrapper(*args, **kwargs):
user_authenticated = kwargs.get('authenticated', False)
if not user_authenticated:
raise PermissionError("Требуется аутентификация для доступа к этой функции.")
return func(*args, **kwargs)
return wrapper

# Пример использования
@require_authentication
def get_user_data(user_id: int, authenticated=False) -> dict:
# Имитация получения данных пользователя
return {"user_id": user_id, "name": "Alice"}

try:
print(get_user_data(user_id=1, authenticated=True)) # Успешный доступ
print(get_user_data(user_id=2, authenticated=False)) # Ошибка доступа
except PermissionError as e:
print(e)


📌 Этот декоратор добавляет проверку аутентификации для функции, что полезно для защиты API от неавторизованного доступа. Он оборачивает функцию, проверяет наличие флага аутентификации и выбрасывает исключение, если доступ запрещен. Это может быть полезно, например, в веб-приложениях, где безопасность имеет первостепенное значение.

Подпишись 👉🏻 @KodduuPython 🤖
⚡️ Реактивное программирование с RxPy


import rx
from rx import operators as ops
from rx.scheduler import ThreadPoolScheduler
import requests

def fetch_url(url: str) -> str:
try:
response = requests.get(url)
response.raise_for_status()
return f"Fetched {url} with status {response.status_code}"
except requests.RequestException as e:
return f"Error fetching {url}: {str(e)}"

urls = [
"http://example.com",
"http://nonexistent.url",
"http://example.org"
]

# Настройка многопоточного планировщика
scheduler = ThreadPoolScheduler(max_workers=2)

# Создание реактивного потока
rx.from_(urls).pipe(
ops.map(lambda url: fetch_url(url)),
ops.do_action(lambda result: print(result))
).subscribe(
on_error=lambda e: print(f"Error occurred: {e}"),
on_completed=lambda: print("Fetching completed!")
)


📌 Этот код демонстрирует использование RxPy для асинхронного веб-скрапинга. Он создает поток URL-адресов, которые обрабатываются параллельно с помощью многопоточного планировщика. Каждый URL запрашивается, и результат выводится в консоль. Такой подход полезен для обработки большого количества запросов или выполнения асинхронных задач.

Подпишись 👉🏻 @KodduuPython 🤖
Forwarded from AIGENTTO
Два бота OpenClaw

Самим OpenClaw я обычно делаю апдейт его конфигов, но иногда это изменение что-то ломает. Поэтому у меня крутится всегда два бота, один чинит/меняет конфиги другого. А другой первого. Система абсолютно устойчива.

Подпишись 👉🏻 @aigentto 🤖
🚀 Улучшение производительности веб-приложений с помощью Asyncio


import asyncio
import aiohttp

async def fetch_data(url: str) -> dict:
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status == 200:
try:
return await response.json()
except aiohttp.ContentTypeError:
print("Получен ответ, не являющийся JSON")
else:
response.raise_for_status()

async def main():
urls = [
'https://api.example.com/data1',
'https://api.example.com/data2',
'https://api.example.com/data3',
]
tasks = [fetch_data(url) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
print(result)

# Запуск асинхронной программы
asyncio.run(main())


📌 Этот код демонстрирует, как использовать асинхронные операции для выполнения нескольких HTTP-запросов параллельно. С помощью библиотек aiohttp и asyncio можно значительно сократить время отклика при работе с сетевыми запросами, так как они выполняются без блокировки основного потока. Такая техника особенно полезна для веб-приложений, которые часто обращаются к внешним API.

Подпишись 👉🏻 @KodduuPython 🤖
👍2
🗄️ Оптимизация запросов к базам данных с использованием SQLAlchemy


from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship, joinedload

Base = declarative_base()

class Author(Base):
__tablename__ = 'authors'
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)

class Book(Base):
__tablename__ = 'books'
id = Column(Integer, primary_key=True)
title = Column(String, nullable=False)
author_id = Column(Integer, ForeignKey('authors.id'))
author = relationship('Author', back_populates='books')

Author.books = relationship('Book', order_by=Book.id, back_populates='author')

# Настройка базы данных
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

# Пример использования: Избегание избыточных запросов с помощью joinedload
def get_books_with_authors():
with Session() as session:
try:
books = session.query(Book).options(joinedload(Book.author)).all()
return [(book.title, book.author.name) for book in books]
except Exception as e:
print(f"Ошибка: {e}")

# Создание примеров данных
with Session() as session:
author = Author(name='J.K. Rowling')
book = Book(title='Harry Potter', author=author)
session.add(author)
session.add(book)
session.commit()

# Вызов функции
print(get_books_with_authors())


📌 Этот код демонстрирует использование SQLAlchemy для оптимизации запросов к базе данных с помощью joinedload. Эта техника уменьшает количество SQL-запросов за счет загрузки связанных данных в одно обращение, что особенно полезно при работе с большими объемами данных.

Подпишись 👉🏻 @KodduuPython 🤖
👍2