🔄 Оптимизация SQL запросов с SQLAlchemy

from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError

def get_optimized_data(engine, query: str):
    try:
        with engine.connect() as connection:
            # Выполнение SQL-запроса и получение всех результатов
            result = connection.execute(text(query))
            return result.fetchall()
    except SQLAlchemyError as e:
        print(f"Ошибка выполнения запроса: {e}")
        return None

# Настройка подключения к базе данных
engine = create_engine('sqlite:///example.db')

# Пример оптимизированного запроса
optimized_query = """
SELECT * FROM users
WHERE age > 30
ORDER BY last_login DESC
LIMIT 10
"""

# Получение данных
data = get_optimized_data(engine, optimized_query)
for row in data:
    print(row)

📌 Этот код показывает, как использовать SQLAlchemy для выполнения оптимизированных SQL-запросов. Функция get_optimized_data выполняет запросы с обработкой ошибок, что повышает устойчивость приложения. Использование индексов и ограничений (`LIMIT`) помогает улучшить производительность при работе с большими объемами данных.

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

📊 Автоматизация анализа данных в реальном времени с Apache Kafka

from kafka import KafkaProducer, KafkaConsumer
from json import dumps, loads
import logging

logging.basicConfig(level=logging.INFO)

def create_producer(broker: str):
    return KafkaProducer(
        bootstrap_servers=[broker],
        value_serializer=lambda x: dumps(x).encode('utf-8')
    )

def create_consumer(broker: str, topic: str, group_id: str):
    return KafkaConsumer(
        topic,
        bootstrap_servers=[broker],
        auto_offset_reset='earliest',
        group_id=group_id,
        value_deserializer=lambda x: loads(x.decode('utf-8'))
    )

def send_data(producer, topic: str, data: dict):
    try:
        producer.send(topic, value=data)
        producer.flush()
        logging.info(f"Отправлены данные: {data}")
    except Exception as e:
        logging.error(f"Ошибка отправки данных: {e}")

def consume_data(consumer):
    for message in consumer:
        try:
            logging.info(f"Получены данные: {message.value}")
        except Exception as e:
            logging.error(f"Ошибка обработки сообщения: {e}")

# Пример использования
broker_address = 'localhost:9092'
producer = create_producer(broker_address)
consumer = create_consumer(broker_address, 'test-topic', 'test-group')

# Отправка данных
send_data(producer, 'test-topic', {'key': 'value'})

# Обработка данных
consume_data(consumer)

📌 Этот код демонстрирует простую реализацию продюсера и потребителя в Apache Kafka на Python. С помощью Kafka вы можете отправлять и получать сообщения в реальном времени, что идеально подходит для анализа потоковых данных. Код показывает, как использовать Kafka для создания систем мониторинга и оповещения.

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

Будет еще много типов агентов на BlackBoard

На BlackBoard можно создать любое количество и любые типы агентов.

В планах добавить:
🕷️ spider-agent — для поиска в интернете
🧭 lead-agent — для координации микро-роев агентов
🔌 API-agent — для доступа к внешним API
🗃️ sql-agent — для доступа к БД
💬 stream-user-agent — для стриминга ответа пользователю
🧠 memory-agent — для памяти всех переписок
🧩 control-agent — для контроля выполнения мета-задач
🛡️ security-agent — для контроля безопасности переписки агентов

Подпишись 👉🏻 @aigentto 🤖

📅 Автоматизированные уведомления о событиях в календаре

import os
import datetime
import requests
from google.oauth2 import service_account
from googleapiclient.discovery import build

def get_upcoming_events(service, calendar_id='primary'):
    now = datetime.datetime.utcnow().isoformat() + 'Z'
    events_result = service.events().list(calendarId=calendar_id, timeMin=now,
                                          maxResults=10, singleEvents=True,
                                          orderBy='startTime').execute()
    return events_result.get('items', [])

def send_telegram_message(chat_id, message, bot_token):
    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
    payload = {'chat_id': chat_id, 'text': message}
    response = requests.post(url, data=payload)
    response.raise_for_status()

def main():
    SERVICE_ACCOUNT_FILE = 'path/to/credentials.json'
    SCOPES = ['https://www.googleapis.com/auth/calendar.readonly']
    
    credentials = service_account.Credentials.from_service_account_file(
        SERVICE_ACCOUNT_FILE, scopes=SCOPES)
    service = build('calendar', 'v3', credentials=credentials)

    events = get_upcoming_events(service)
    if not events:
        print('No upcoming events found.')
        return

    bot_token = os.getenv('TELEGRAM_BOT_TOKEN')
    chat_id = os.getenv('TELEGRAM_CHAT_ID')

    for event in events:
        start = event['start'].get('dateTime', event['start'].get('date'))
        message = f"Upcoming Event: {event['summary']} at {start}"
        send_telegram_message(chat_id, message, bot_token)

if __name__ == '__main__':
    main()

📌 Этот скрипт автоматически извлекает предстоящие события из Google Calendar и отправляет уведомления в Telegram. Он использует Google Calendar API для доступа к данным календаря и Telegram Bot API для доставки сообщений. Полезен для управления временем и повышения продуктивности.

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

🚨 Последняя распродажа следующих курсов, после этой распродажи доступ к этим курсам будет закрыт (останется только у тех, кто купил):

👉 Python: самый быстрый курс 🔥🔥🔥
👉 Python Data Science: самый быстрый курс 🔥
👉 JavaScript: самый быстрый курс 🔥🔥

⚡️⚡️⚡️ Только до 31 января ⚡️⚡️⚡️

Размышляем над тупиками нейросетей тут https://habr.com/ru/articles/989054/ и тут https://www.reddit.com/r/AIGENTTO/comments/1qnddav/agi_is_not_coming_yet/

Подпишись 👉🏻 @aigentto 🤖

📊 Реальный анализ данных с MongoDB и PyMongo

import pymongo
from pymongo.errors import ConnectionError, OperationFailure, PyMongoError
from datetime import datetime

def connect_to_mongo(uri: str):
    try:
        client = pymongo.MongoClient(uri)
        client.admin.command('ping')  # Проверяем подключение
        print("Успешное подключение к MongoDB")
        return client
    except (ConnectionError, OperationFailure) as e:
        print(f"Ошибка подключения: {e}")
        return None

def store_social_data(db, collection_name: str, data: dict):
    try:
        collection = db[collection_name]
        data['timestamp'] = datetime.now()  # Добавляем временную метку
        collection.insert_one(data)  # Сохраняем данные в коллекцию
        print(f"Данные сохранены в коллекцию {collection_name}")
    except PyMongoError as e:
        print(f"Ошибка сохранения данных: {e}")

# Пример использования
if __name__ == "__main__":
    mongo_uri = "mongodb://localhost:27017/"
    client = connect_to_mongo(mongo_uri)

    if client:
        db = client['social_data']
        sample_data = {
            "user": "john_doe",
            "platform": "Twitter",
            "content": "Python 3.9 вышел!",
            "likes": 150
        }
        store_social_data(db, "posts", sample_data)

📌 Этот код показывает, как подключиться к MongoDB и сохранить данные из социальных сетей в реальном времени. Используя PyMongo, создаем подключение к базе данных и сохраняем пример данных с временной меткой. Это полезно для маркетинга и аналитики, отслеживая активность в социальных сетях.

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

🎬 Создание системы рекомендаций с Surprise и scikit-learn

from surprise import Dataset, Reader, SVD
from surprise.model_selection import cross_validate
import pandas as pd

# Загрузка данных
data_dict = {'item': [1, 2, 3, 1, 2],
             'user': [1, 1, 1, 2, 2],
             'rating': [4, 5, 3, 2, 4]}
df = pd.DataFrame(data_dict)
reader = Reader(rating_scale=(1, 5))
data = Dataset.load_from_df(df[['user', 'item', 'rating']], reader)

# Создание и обучение модели SVD
model = SVD()
cross_validate(model, data, measures=['RMSE', 'MAE'], cv=2, verbose=True)

# Пример предсказания
trainset = data.build_full_trainset()
model.fit(trainset)
user_id = 1
item_id = 3
prediction = model.predict(user_id, item_id)
print(f"Предсказанный рейтинг для пользователя {user_id} и товара {item_id}: {prediction.est:.2f}")

📌 В этом примере мы создаем простую систему рекомендаций, используя библиотеку Surprise. Мы загружаем данные, создаем модель SVD и проводим кросс-валидацию для оценки её качества. Этот подход подходит для создания базовых рекомендательных систем, которые могут быть улучшены с помощью более сложных алгоритмов и дополнительных данных.

🛠 pip install surprise pandas

Этот код предоставляет основу для разработки рекомендательных систем, которые могут быть интегрированы в широкий спектр приложений, от интернет-магазинов до платформ потокового видео.

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

🚀 Последний шанс купить следующие курсы:

👉 Python: самый быстрый курс 🔥🔥🔥
👉 Python Data Science: самый быстрый курс 🔥
👉 JavaScript: самый быстрый курс 🔥🔥

⚡️⚡️⚡️ Только до 31 января ⚡️⚡️⚡️

🚀 Внедрение CI/CD для Python-проектов с GitHub Actions

name: CI/CD Pipeline

on:
  push:
    branches:
      - main
  pull_request:
    branches:
      - main

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v2

      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: 3.8

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt

      - name: Run tests
        run: pytest

  deploy:
    runs-on: ubuntu-latest
    needs: test
    steps:
      - name: Checkout code
        uses: actions/checkout@v2

      - name: Deploy to Production
        run: echo "Deploying application..."

📌 Этот YAML-файл описывает базовый CI/CD конвейер для Python-проекта с использованием GitHub Actions. Он включает две задачи: тестирование и развертывание. При каждом коммите в ветку main запускаются тесты с помощью pytest, и в случае их успешного прохождения выполняется развертывание. Это помогает автоматизировать процессы разработки и уменьшает вероятность ошибок при развертывании.

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

⚡️⚡️⚡️ Доступ к продаже следующих курсов закрывается сегодня ⚡️⚡️⚡️

👉 Python: самый быстрый курс 🔥🔥🔥
👉 Python Data Science: самый быстрый курс 🔥
👉 JavaScript: самый быстрый курс 🔥🔥

🗺️ Обработка и визуализация геоданных с GeoPandas

import geopandas as gpd
from shapely.geometry import Point

def load_geodata(file_path: str) -> gpd.GeoDataFrame:
    try:
        return gpd.read_file(file_path)
    except FileNotFoundError:
        print("Файл не найден.")
    except Exception as e:
        print(f"Ошибка при чтении файла: {e}")
    return gpd.GeoDataFrame()

def create_point_geodataframe(latitudes: list, longitudes: list) -> gpd.GeoDataFrame:
    if len(latitudes) != len(longitudes):
        print("Списки широт и долгот должны быть одинаковой длины.")
        return gpd.GeoDataFrame()
    geometry = [Point(xy) for xy in zip(longitudes, latitudes)]
    return gpd.GeoDataFrame(geometry=geometry, crs="EPSG:4326")

def filter_data_by_region(gdf: gpd.GeoDataFrame, region_name: str) -> gpd.GeoDataFrame:
    try:
        return gdf[gdf['region'] == region_name]
    except KeyError:
        print(f"Колонка 'region' отсутствует в данных.")
    return gpd.GeoDataFrame()

# Пример использования
if __name__ == "__main__":
    # Загрузка геоданных
    geodata = load_geodata("example.geojson")
    
    # Создание GeoDataFrame из координат
    latitudes = [37.7749, 34.0522]
    longitudes = [-122.4194, -118.2437]
    points = create_point_geodataframe(latitudes, longitudes)
    
    # Фильтрация по региону
    filtered_data = filter_data_by_region(geodata, "California")
    print(filtered_data)

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

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

🚀 Оптимизация кода с мемоизацией

from functools import lru_cache

@lru_cache(maxsize=None)
def fibonacci(n: int) -> int:
    """Рассчитывает n-е число Фибоначчи с мемоизацией."""
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# Пример использования
print(fibonacci(10))  # Вывод: 55

📌 Этот код демонстрирует использование мемоизации для оптимизации вычисления чисел Фибоначчи. Декоратор lru_cache из стандартной библиотеки автоматически сохраняет результаты вызова функции, предотвращая повторные вычисления для одинаковых аргументов. Такой подход может значительно ускорить выполнение рекурсивных функций и задач, требующих многократных вызовов с одинаковыми параметрами, например в анализе данных или при решении комбинаторных задач.

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

🚀 Функциональные и реактивные подходы в веб-разработке с Flask и React

from flask import Flask, jsonify, request
from functools import partial
from concurrent.futures import ThreadPoolExecutor

app = Flask(__name__)

# Функция для обработки данных
def process_data(data: dict, multiplier: int) -> dict:
    return {k: v * multiplier for k, v in data.items()}

# Обработчик запросов, использующий функциональные принципы
@app.route('/process', methods=['POST'])
def process_request():
    try:
        data = request.json
        if not data:
            raise ValueError("Отсутствуют JSON-данные")

        multiplier = data.get('multiplier', 1)
        if not isinstance(multiplier, int):
            raise ValueError("Multiplier должен быть целым числом")

        values = data.get('values')
        if not isinstance(values, dict):
            raise ValueError("Values должен быть словарем")

        process = partial(process_data, values, multiplier)
        
        with ThreadPoolExecutor() as executor:
            result = executor.submit(process).result()

        return jsonify(result), 200
    except Exception as e:
        return jsonify({'error': str(e)}), 400

# Пример использования: отправка POST-запроса с JSON-данными
# curl -X POST http://localhost:5000/process -H "Content-Type: application/json" -d '{"values": {"a": 1, "b": 2}, "multiplier": 3}'

📌 Этот код демонстрирует использование функциональных принципов в Flask для обработки данных. Применяя functools.partial и ThreadPoolExecutor, мы улучшаем производительность и отзывчивость серверной части приложения. Эти подходы позволяют легко масштабировать и тестировать серверные компоненты, что особенно полезно при высокой нагрузке.

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

🚀 Оптимизация моделей машинного обучения с Optuna

import optuna
from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split, cross_val_score

# Загрузка данных и разделение на обучающую и тестовую выборки
data = load_iris()
X_train, X_test, y_train, y_test = train_test_split(
    data.data, data.target, test_size=0.2, random_state=42
)

def objective(trial):
    n_estimators = trial.suggest_int('n_estimators', 10, 100)
    max_depth = trial.suggest_int('max_depth', 2, 32, log=True)
    min_samples_split = trial.suggest_int('min_samples_split', 2, 10)

    clf = RandomForestClassifier(
        n_estimators=n_estimators,
        max_depth=max_depth,
        min_samples_split=min_samples_split,
        random_state=42
    )
    
    # Кросс-валидация с параллельными вычислениями
    return cross_val_score(clf, X_train, y_train, n_jobs=-1, cv=3).mean()

study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=100)

print(f'Лучшие параметры: {study.best_params}')
print(f'Лучшая точность кросс-валидации: {study.best_value:.2f}')

# Пример использования с найденными гиперпараметрами
best_clf = RandomForestClassifier(**study.best_params)
best_clf.fit(X_train, y_train)
accuracy = best_clf.score(X_test, y_test)
print(f'Точность на тестовом наборе: {accuracy:.2f}')

📌 Этот код показывает, как использовать Optuna для автоматизации настройки гиперпараметров модели RandomForest. Optuna помогает находить оптимальные параметры, чтобы улучшить точность модели. Использование n_jobs=-1 позволяет выполнять кросс-валидацию параллельно, что ускоряет процесс.

🛠 pip install optuna scikit-learn

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

🌿 Симуляция клеточных автоматов для моделирования экосистем

import pygame
import random

# Константы
WINDOW_SIZE = 600
GRID_SIZE = 30
CELL_SIZE = WINDOW_SIZE // GRID_SIZE

# Инициализация Pygame
pygame.init()
window = pygame.display.set_mode((WINDOW_SIZE, WINDOW_SIZE))
clock = pygame.time.Clock()

def create_grid():
    return [[random.choice([0, 1]) for _ in range(GRID_SIZE)] for _ in range(GRID_SIZE)]

def update_grid(grid):
    new_grid = [[0] * GRID_SIZE for _ in range(GRID_SIZE)]
    for y in range(GRID_SIZE):
        for x in range(GRID_SIZE):
            alive_neighbors = sum([grid[y+dy][x+dx] for dx in [-1, 0, 1] if 0 <= x+dx < GRID_SIZE
                                   for dy in [-1, 0, 1] if 0 <= y+dy < GRID_SIZE and (dx != 0 or dy != 0)])
            if grid[y][x] == 1 and alive_neighbors in (2, 3):
                new_grid[y][x] = 1
            elif grid[y][x] == 0 and alive_neighbors == 3:
                new_grid[y][x] = 1
    return new_grid

def draw_grid(grid):
    for y in range(GRID_SIZE):
        for x in range(GRID_SIZE):
            color = (0, 255, 0) if grid[y][x] == 1 else (0, 0, 0)
            pygame.draw.rect(window, color, (x*CELL_SIZE, y*CELL_SIZE, CELL_SIZE, CELL_SIZE))

# Основной цикл
grid = create_grid()
running = True
while running:
    for event in pygame.event.get():
        if event.type == pygame.QUIT:
            running = False

    grid = update_grid(grid)
    window.fill((0, 0, 0))
    draw_grid(grid)
    pygame.display.flip()
    clock.tick(10)

pygame.quit()

📌 Этот код создает симуляцию клеточного автомата с использованием Pygame. Клеточные автоматы моделируют экосистемы через сетку клеток, где каждая клетка может быть "живой" или "мёртвой". Симуляция обновляется по правилам, что позволяет исследовать, как развиваются структуры во времени.

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

🔥 Создание системы частиц в Pygame

import pygame
import random
import sys

# Инициализация Pygame
pygame.init()
screen = pygame.display.set_mode((800, 600))
clock = pygame.time.Clock()

class Particle:
    def __init__(self, pos, vel, color, lifespan):
        self.pos = list(pos)
        self.vel = list(vel)
        self.color = color
        self.lifespan = lifespan

    def update(self):
        self.pos[0] += self.vel[0]
        self.pos[1] += self.vel[1]
        self.lifespan -= 1

    def draw(self, screen):
        if self.lifespan > 0:
            pygame.draw.circle(screen, self.color, (int(self.pos[0]), int(self.pos[1])), 3)

def create_particles(count, pos):
    particles = []
    for _ in range(count):
        particles.append(Particle(
            pos=pos,
            vel=(random.uniform(-1, 1), random.uniform(-1, 1)),
            color=(255, 255, 255),
            lifespan=random.randint(20, 100)
        ))
    return particles

particles = create_particles(100, (400, 300))

# Главный цикл
while True:
    for event in pygame.event.get():
        if event.type == pygame.QUIT:
            pygame.quit()
            sys.exit()

    screen.fill((0, 0, 0))

    for particle in particles[:]:
        particle.update()
        particle.draw(screen)
        if particle.lifespan <= 0:
            particles.remove(particle)

    pygame.display.flip()
    clock.tick(60)

📌 Этот код создает простую систему частиц в Pygame, где частицы возникают в центре экрана и со временем затухают. Система обновляет позиции частиц, изменяет их состояние и удаляет их, когда они "умирают". Такой код может быть использован для создания эффектов, таких как дым или огонь, в 2D-играх.

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

Наша новая статья ИИ-агенты без оркестрации и роутинга на habr.

Больше по теме в канале 👉🏻 @aigentto 🤖

Мясной слой для ИИ-агентов

Пока Elon Musk ещё не продает массово своих роботов Optimus 3, появилась замена. Теперь любой ИИ-агент может арендовать человека для выполнения задач в реальном мире 😄

Как я вижу, спрос со стороны людей пока превышает количество задач со стороны ИИ.

Но реально теперь можно собирать ИИ-агентов, у которых будут задачи в реальном мире, я точно попробую прикрутить мясной слой (как его назвали сами создатели сайта) к своим экспериментам с роями агентов на Blackboard.

Бывают такие задачи, например, позвонить куда-то и узнать что-то, или даже ответить на вопрос по данным, когда LLM не смог справиться.

API тут https://rentahuman.ai/mcp. Доступные люди тут https://rentahuman.ai/browse.

Подпишись 👉🏻 @aigentto 🤖

🔧 Функциональное программирование с functools и itertools

from itertools import chain
from collections import defaultdict

def process_data(data):
    # Объединяем элементы из вложенных списков
    combined = chain.from_iterable(data)
    
    # Группируем и суммируем значения по ключу
    result = defaultdict(int)
    for key, value in combined:
        result[key] += value
    
    return dict(result)

# Пример использования
data = [[('apple', 10), ('banana', 5)], [('apple', 3), ('banana', 2)], [('banana', 1)]]
result = process_data(data)
print(result)  # Вывод: {'apple': 13, 'banana': 8}

📌 Этот код демонстрирует использование itertools.chain для объединения списков и collections.defaultdict для упрощения группировки и суммирования данных. Такой подход полезен для эффективной обработки и агрегирования больших наборов данных.

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

💰 Реализация внутриигровых покупок с Flask

from flask import Flask, jsonify, request, abort
from dataclasses import dataclass, asdict

app = Flask(__name__)

# Модель покупки
@dataclass
class Purchase:
    item_id: int
    user_id: int
    amount: float

# Пример данных
purchases = []

@app.route('/purchase', methods=['POST'])
def create_purchase():
    if not request.json or 'item_id' not in request.json or 'user_id' not in request.json:
        abort(400, description="Неправильный запрос: отсутствует item_id или user_id.")
    if request.json.get('amount', 0.0) <= 0:
        abort(400, description="Неправильный запрос: сумма должна быть положительной.")
    purchase = Purchase(
        item_id=request.json['item_id'],
        user_id=request.json['user_id'],
        amount=request.json['amount']
    )
    purchases.append(purchase)
    return jsonify({'purchase': asdict(purchase)}), 201

# Flask сервер
if __name__ == '__main__':
    app.run(debug=True)

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

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