Вышла новая статья на habr "Как извлечь ДНК автора? Креатив от LLM" 💡

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

🔍 Асинхронный веб-скрейпинг с Aiohttp и BeautifulSoup

import aiohttp
import asyncio
from bs4 import BeautifulSoup

async def fetch(session, url):
    async with session.get(url) as response:
        if response.status == 200:
            html = await response.text()
            # Парсинг HTML для извлечения заголовка
            soup = BeautifulSoup(html, 'html.parser')
            title = soup.title.string if soup.title else 'No Title Found'
            return title
        else:
            raise Exception(f"Ошибка при получении {url}: статус {response.status}")

async def main(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        titles = await asyncio.gather(*tasks, return_exceptions=True)
        return titles

# Пример использования
urls = ['https://www.python.org', 'https://www.example.com']
titles = asyncio.run(main(urls))
print(titles)

📌 Этот код выполняет асинхронный веб-скрейпинг, извлекая заголовки страниц с заданных URL. Используя aiohttp для асинхронных HTTP-запросов и BeautifulSoup для парсинга HTML, он эффективно обрабатывает множество страниц одновременно. Этот подход особенно полезен для обработки больших объемов данных без блокировки выполнения программы.

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

🖼️ Простая обработка изображений с PIL и OpenCV

from PIL import Image, ImageFilter
import cv2
import numpy as np

def process_image(image_path: str, output_path: str):
    try:
        # Открываем изображение с помощью PIL
        pil_image = Image.open(image_path)
        
        # Применяем размытие
        blurred_image = pil_image.filter(ImageFilter.BLUR)
        
        # Сохраняем результат
        blurred_image.save(output_path)
        
        # Чтение изображения с помощью OpenCV
        cv_image = cv2.imread(output_path)
        
        # Конвертируем в градации серого
        gray_image = cv2.cvtColor(cv_image, cv2.COLOR_BGR2GRAY)
        
        # Применяем детектор границ Canny
        edges = cv2.Canny(gray_image, 100, 200)
        
        # Сохраняем изображение с границами
        cv2.imwrite(output_path.replace('.jpg', '_edges.jpg'), edges)
        
    except Exception as e:
        print(f"Ошибка обработки изображения: {e}")

# Пример использования
process_image('input.jpg', 'output.jpg')

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

🛠 pip install pillow opencv-python-headless

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

🛠️ Быстрое RESTful API с FastAPI

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class Item(BaseModel):
    id: int
    name: str
    description: Optional[str] = None  # Описание элемента

# Пример данных
items_db = [
    {"id": 1, "name": "item1", "description": "Описание первого элемента"},
    {"id": 2, "name": "item2", "description": "Описание второго элемента"},
]

@app.get("/items/", response_model=List[Item])
async def read_items():
    return items_db

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    for item in items_db:
        if item["id"] == item_id:
            return item
    raise HTTPException(status_code=404, detail="Элемент не найден")

@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    for existing_item in items_db:
        if existing_item["id"] == item.id:
            raise HTTPException(status_code=400, detail="ID уже существует")
    items_db.append(item.dict())
    return item

# Пример использования:
# 1. Запустите сервер: uvicorn filename:app --reload
# 2. Посетите http://127.0.0.1:8000/items/ для просмотра всех элементов
📌 Этот код демонстрирует создание простого RESTful API с использованием FastAPI. API позволяет получать список элементов, искать элемент по ID и добавлять новые элементы с уникальными ID. FastAPI обеспечивает быструю разработку и безопасность API благодаря встроенной валидации и асинхронному выполнению.

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

📊 Простая агрегация данных с Pandas

import pandas as pd

def aggregate_sales_data(file_path: str) -> pd.DataFrame:
    try:
        data = pd.read_csv(file_path, parse_dates=['Date'])
        
        # Группировка данных по месяцам и суммирование продаж
        monthly_sales = (
            data.groupby(data['Date'].dt.to_period('M'))
            .agg(total_sales=('Sales', 'sum'))
            .reset_index()
        )
        return monthly_sales
    except FileNotFoundError:
        print("Файл не найден. Проверьте путь к файлу.")
        return pd.DataFrame()
    except Exception as e:
        print(f"Произошла ошибка: {e}")
        return pd.DataFrame()

# Пример использования
file_path = 'sales_data.csv'
monthly_sales = aggregate_sales_data(file_path)
print(monthly_sales)

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

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

* С новым годом *

import os
import time
import random

WIDTH = 80
HEIGHT = 20

# ASCII-art цифр (5 строк)
DIGITS = {
    "0": [
        " █████ ",
        " █   █ ",
        " █   █ ",
        " █   █ ",
        " █████ "
    ],
    "2": [
        " █████ ",
        "     █ ",
        " █████ ",
        " █     ",
        " █████ "
    ],
    "6": [
        " █████ ",
        " █     ",
        " █████ ",
        " █   █ ",
        " █████ "
    ]
}

# собрать строку 2026
def get_2026_art():
    art_lines = [""] * 5
    for digit in "2026":
        for i in range(5):
            art_lines[i] += DIGITS[digit][i] + "  "
    return art_lines

# создаем пустое поле для снега
def init_snow():
    return [[random.choice([False, True, False, False, False]) for _ in range(WIDTH)] for _ in range(HEIGHT)]

# обновление позиции снега
def update_snow(snow):
    new_snow = [[False]*WIDTH for _ in range(HEIGHT)]
    for y in range(HEIGHT-1):
        for x in range(WIDTH):
            if snow[y][x]:
                new_y = y + 1
                if new_y < HEIGHT:
                    new_snow[new_y][x] = True
    # случайные новые снежинки сверху
    for x in range(WIDTH):
        if random.random() < 0.02:
            new_snow[0][x] = True
    return new_snow

# отрисовка
def draw(snow, art):
    canvas = [[" "] * WIDTH for _ in range(HEIGHT)]
    # рисуем снег
    for y in range(HEIGHT):
        for x in range(WIDTH):
            if snow[y][x]:
                canvas[y][x] = "*"
    # рисуем 2026 примерно в центре
    start_x = (WIDTH - len(art[0])) // 2
    start_y = 5
    for i, line in enumerate(art):
        for j, c in enumerate(line):
            if c != " " and 0 <= start_y+i < HEIGHT and 0 <= start_x+j < WIDTH:
                canvas[start_y + i][start_x + j] = c
    return "\n".join("".join(row) for row in canvas)

def clear():
    os.system("cls" if os.name == "nt" else "clear")

# main loop
snow = init_snow()
art = get_2026_art()
while True:
    clear()
    print(draw(snow, art))
    snow = update_snow(snow)
    time.sleep(0.1)

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

Анимация по коду выше 👆👆👆

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

🔧 Автоматизация управления репозиториями с GitHub API

from github import Github
from github.GithubException import GithubException
import os

def create_repository(repo_name: str, private: bool = True) -> str:
    try:
        # Получаем токен доступа из переменных окружения
        token = os.getenv("GITHUB_ACCESS_TOKEN")
        if not token:
            raise ValueError("Токен доступа не установлен в переменных окружения.")

        # Подключаемся к GitHub с использованием токена
        g = Github(token)
        user = g.get_user()

        # Создаем новый репозиторий
        repo = user.create_repo(name=repo_name, private=private)
        return f"Репозиторий '{repo.name}' успешно создан по адресу {repo.html_url}"
    except GithubException as e:
        return f"Ошибка при создании репозитория: {e.data['message']}"
    except Exception as e:
        return f"Произошла ошибка: {str(e)}"

# Пример использования
if __name__ == "__main__":
    print(create_repository("my-new-repo", private=True))

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

🛠 pip install PyGithub

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

Если кто есть на reddit, лайкните пост плиз, мне надо набивать там карму 🙏

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

⚡ Быстрая обработка JSON с помощью ujson и orjson

import ujson
import orjson

def load_json_ujson(data: str):
    try:
        return ujson.loads(data)
    except ValueError as e:
        print(f"Ошибка загрузки JSON через ujson: {e}")
        return None

def load_json_orjson(data: bytes):
    try:
        return orjson.loads(data)
    except orjson.JSONDecodeError as e:
        print(f"Ошибка загрузки JSON через orjson: {e}")
        return None

def dump_json_ujson(data: dict) -> str:
    try:
        return ujson.dumps(data)
    except ValueError as e:
        print(f"Ошибка сериализации JSON через ujson: {e}")
        return ""

def dump_json_orjson(data: dict) -> bytes:
    try:
        return orjson.dumps(data)
    except orjson.JSONEncodeError as e:
        print(f"Ошибка сериализации JSON через orjson: {e}")
        return b""

# Пример использования
json_string = '{"name": "Alice", "age": 30}'
json_bytes = b'{"name": "Bob", "age": 25}'

ujson_data = load_json_ujson(json_string)
orjson_data = load_json_orjson(json_bytes)

print(dump_json_ujson(ujson_data))
print(dump_json_orjson(orjson_data))

📌 С помощью библиотек ujson и orjson можно значительно ускорить сериализацию и десериализацию JSON-данных по сравнению со стандартной библиотекой json. Эти библиотеки полезны для повышения производительности веб-приложений и скриптов, работающих с большими объёмами данных. Они обеспечивают более быструю обработку JSON, что критически важно для современного веб-разработчика.

🛠 pip install ujson orjson

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

Создал свой reddit, там будут посты на английском по теме Agentic AI. Будет больше специфики мировой и многие темы будут выходить раньше, чем в канале @aigentto 😎

Подключайтесь 👉 https://www.reddit.com/r/AIGENTTO/ 🤖

📊 Автоматизация отчетности с Pandas и Matplotlib

import pandas as pd
import matplotlib.pyplot as plt
from smtplib import SMTP
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.image import MIMEImage

def generate_report(data_file: str, email: str):
    try:
        # Чтение и обработка данных из Excel-файла
        df = pd.read_excel(data_file, engine='openpyxl')
        summary = df.describe()

        # Создание визуализации
        plt.figure(figsize=(10, 6))
        df.plot(kind='line')
        plt.title('Обзор данных о продажах')
        plt.xlabel('Дата')
        plt.ylabel('Продажи')
        plt.grid(True)
        plt.tight_layout()

        # Сохранение графика
        plot_file = 'sales_report.png'
        plt.savefig(plot_file)

        # Отправка отчета по электронной почте
        send_email(email, plot_file, summary.to_string())
    except Exception as e:
        print(f"Произошла ошибка: {e}")

def send_email(to_email: str, image_path: str, summary: str):
    msg = MIMEMultipart()
    msg['Subject'] = 'Автоматический отчет по продажам'
    msg['From'] = 'your_email@example.com'
    msg['To'] = to_email

    text = MIMEText(f"Добрый день,\n\nПожалуйста, найдите отчет по продажам ниже:\n\n{summary}", 'plain')
    msg.attach(text)

    with open(image_path, 'rb') as img:
        img_data = img.read()
        image = MIMEImage(img_data, name='Отчет о продажах')
        msg.attach(image)

    with SMTP('smtp.example.com', 587) as smtp:
        smtp.starttls()
        smtp.login('your_email@example.com', 'your_password')
        smtp.sendmail(msg['From'], msg['To'], msg.as_string())

# Пример использования
generate_report('sales_data.xlsx', 'recipient@example.com')

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

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

🤖 Создание чат-бота для Telegram с использованием TeleBot и OpenAI GPT

import telebot
import openai
import os

# Установка токенов из окружения
TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN')
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')

# Инициализация клиентов
bot = telebot.TeleBot(TELEGRAM_TOKEN)
openai.api_key = OPENAI_API_KEY

def generate_response(prompt: str) -> str:
    try:
        response = openai.Completion.create(
            engine="text-davinci-003",
            prompt=prompt,
            max_tokens=150
        )
        return response.choices[0].text.strip() if response.choices else "Ошибка: пустой ответ"
    except Exception as e:
        return f"Ошибка при генерации ответа: {e}"

@bot.message_handler(func=lambda message: True)
def handle_message(message):
    chat_id = message.chat.id
    user_message = message.text
    bot_response = generate_response(user_message)
    bot.send_message(chat_id, bot_response)

# Запуск бота
bot.polling(none_stop=True)

# Пример использования:
# Отправьте сообщение вашему боту в Telegram, и он ответит, используя OpenAI GPT.

📌 Этот код создает простого чат-бота для Telegram, использующего OpenAI GPT для генерации ответов. Бот принимает текст от пользователя, отправляет его в OpenAI API и возвращает ответ. Такой бот может быть полезен для автоматизации общения и развлечения.

🛠 Убедитесь, что установлены зависимости: pip install pyTelegramBotAPI openai

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

Прогнозирование погоды с использованием OpenWeatherMap API

import requests
from datetime import datetime

def get_weather(api_key: str, city: str) -> dict:
    url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric"
    try:
        response = requests.get(url)
        response.raise_for_status()
        data = response.json()
        weather = {
            "city": data["name"],
            "temperature": data["main"]["temp"],
            "description": data["weather"][0]["description"],
            "datetime": datetime.fromtimestamp(data["dt"]).strftime('%Y-%m-%d %H:%M:%S')
        }
        return weather
    except requests.exceptions.HTTPError as http_err:
        print(f"HTTP ошибка: {http_err}")
    except Exception as err:
        print(f"Произошла ошибка: {err}")

# Пример использования
api_key = "ваш_api_ключ"  # Замените на ваш ключ API
city = "Moscow"
weather_info = get_weather(api_key, city)
if weather_info:
    print(f"Погода в {weather_info['city']} на {weather_info['datetime']}:")
    print(f"{weather_info['temperature']}°C, {weather_info['description']}")

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

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

⚡️ Параллельная обработка данных с concurrent.futures

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

def fetch_url(url: str) -> str:
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return response.text
    except requests.RequestException as e:
        return f"Ошибка при получении {url}: {e}"

def fetch_all(urls: list) -> dict:
    results = {}
    with ThreadPoolExecutor(max_workers=5) as executor:
        future_to_url = {executor.submit(fetch_url, url): url for url in urls}
        for future in as_completed(future_to_url):
            url = future_to_url[future]
            try:
                results[url] = future.result()
            except Exception as exc:
                results[url] = f"Ошибка: {exc}"
    return results

# Пример использования
urls = [
    "https://www.example.com",
    "https://www.python.org",
    "https://www.github.com",
]
results = fetch_all(urls)
for url, content in results.items():
    print(f"URL: {url}, Длина контента: {len(content)}")

📌 Этот код демонстрирует, как использовать ThreadPoolExecutor из модуля concurrent.futures для параллельной загрузки веб-страниц. Функция fetch_all принимает список URL-адресов и возвращает словарь, где ключи — это адреса, а значения — загруженные данные или сообщения об ошибках. Это может быть полезно для ускорения сетевых операций, когда нужно быстро обработать множество запросов.

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

📚 Основы обработки текста с NLTK

import nltk
from nltk.tokenize import word_tokenize
from nltk.corpus import stopwords
from collections import Counter

# Загрузка необходимых ресурсов
nltk.download('punkt')
nltk.download('stopwords')

def analyze_text(text: str) -> dict:
    try:
        # Токенизация текста
        words = word_tokenize(text.lower())

        # Удаление стоп-слов
        stop_words = set(stopwords.words('english'))
        filtered_words = [word for word in words if word.isalnum() and word not in stop_words]

        # Частотный анализ
        word_freq = Counter(filtered_words)
        
        return dict(word_freq)
    except Exception as e:
        print(f"Ошибка при анализе текста: {e}")
        return {}

# Пример использования
text = "NLTK is a leading platform for building Python programs to work with human language data."
result = analyze_text(text)
print(result)

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

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

🔧 Функция-комбинатор для управления потоком данных

from functools import reduce
from typing import Callable

def compose(*functions: Callable) -> Callable:
    def compose_two(f: Callable, g: Callable) -> Callable:
        return lambda x: f(g(x))
    return reduce(compose_two, functions, lambda x: x)

# Пример использования
def add_one(x: int) -> int:
    return x + 1

def double(x: int) -> int:
    return x * 2

def square(x: int) -> int:
    return x ** 2

process = compose(add_one, double, square)
result = process(3)  # (3 + 1) * 2 = 8; 8 ** 2 = 64
print(result)  # Вывод: 64

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

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

🤖 Создание простого чат-бота для технической поддержки

import nltk
from nltk.chat.util import Chat, reflections

# Определение пар шаблонов и ответов
pairs = [
    [r"привет|здравствуй|добрый день", ["Здравствуйте!", "Привет! Как я могу помочь вам сегодня?"]],
    [r"мне нужна помощь с (.*)", ["Конечно, я могу помочь с %1. Что именно вас интересует?"]],
    [r"спасибо|благодарю", ["Всегда рад помочь!", "Не за что!"]],
    [r"пока|до свидания", ["До свидания! Хорошего дня!"]],
]

# Создание чат-бота
def create_chatbot():
    return Chat(pairs, reflections)

# Пример использования
if __name__ == "__main__":
    nltk.download('punkt')  # Загрузка необходимых данных NLTK
    chatbot = create_chatbot()
    print("Бот: Здравствуйте! Я чат-бот технической поддержки.")
    while True:
        user_input = input("Вы: ")
        if user_input.lower() in ["пока", "до свидания"]:
            print("Бот:", chatbot.respond(user_input))
            break
        response = chatbot.respond(user_input)
        print("Бот:", response or "Извините, я не совсем понял. Можете переформулировать вопрос?")

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

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

🔄 Оптимизация 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 🤖