📄 Автоматизация обработки документов с использованием PyPDF2

import PyPDF2

def merge_pdfs(pdf_list, output_path):
    pdf_writer = PyPDF2.PdfWriter()
    
    for pdf_file in pdf_list:
        try:
            with open(pdf_file, 'rb') as file:
                pdf_reader = PyPDF2.PdfReader(file)
                for page in range(len(pdf_reader.pages)):
                    pdf_writer.add_page(pdf_reader.pages)
        except Exception as e:
            print(f"Ошибка при чтении {pdf_file}: {e}")
    
    try:
        with open(output_path, 'wb') as output_pdf:
            pdf_writer.write(output_pdf)
        print(f"Объединенный PDF сохранен как {output_path}")
    except Exception as e:
        print(f"Ошибка при записи объединенного PDF: {e}")

# Пример использования
pdf_files = ["document1.pdf", "document2.pdf"]
merge_pdfs(pdf_files, "merged_document.pdf")

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

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

🕸️ Автоматизация сбора данных с веб-форм с помощью Selenium

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
import time

def setup_driver():
    chrome_options = Options()
    chrome_options.add_argument("--headless")  # Запуск браузера в режиме без интерфейса
    service = Service(executable_path='path/to/chromedriver')
    driver = webdriver.Chrome(service=service, options=chrome_options)
    return driver

def fill_form_and_submit(url, form_data):
    driver = setup_driver()
    try:
        driver.get(url)
        for field_id, value in form_data.items():
            element = driver.find_element(By.ID, field_id)
            element.clear()  # Очистка существующего текста в поле
            element.send_keys(value)  # Ввод нового текста
        
        submit_button = driver.find_element(By.ID, "submit")
        submit_button.click()
        
        time.sleep(3)  # Ожидание загрузки страницы
        return driver.page_source
    except Exception as e:
        print(f"Ошибка при отправке формы: {e}")
    finally:
        driver.quit()

# Пример использования
url = "http://example.com/form"
form_data = {
    "name": "John Doe",
    "email": "john.doe@example.com",
    "message": "Hello, this is a test message!"
}
response = fill_form_and_submit(url, form_data)
print(response[:500])  # Вывод первых 500 символов HTML страницы

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

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

👉 Ключевой бесплатный курс для разминки мозга 🤯 и понимания как работает компьютер 💻

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

📊 Управление задачами с Celery

from celery import Celery

# Создание приложения Celery
app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task
def send_email(recipient: str, subject: str, body: str) -> str:
    # Имитация отправки email
    print(f"Отправка email на {recipient} с темой '{subject}'")
    return "Email отправлен!"

# Пример использования
if __name__ == "__main__":
    result_email = send_email.delay("example@example.com", "Hello!", "Это тестовое сообщение.")
    
    print(result_email.get(timeout=10))

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

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

🔍 Автоматизация тестирования с Pytest

import pytest

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

# Тестовая функция
def test_add():
    assert add(2, 3) == 5
    assert add(-1, 1) == 0
    assert add(0, 0) == 0

# Параметризация тестов
@pytest.mark.parametrize("a, b, expected", [
    (1, 1, 2),
    (-1, -1, -2),
    (100, 200, 300),
])
def test_add_parametrized(a, b, expected):
    assert add(a, b) == expected

# Запуск тестов командой: pytest имя_файла.py
📌 Данный код демонстрирует основы использования Pytest для автоматизации тестирования функций в Python. Он включает тесты для функции сложения и пример параметризации тестов для проверки различных входных данных. Это полезно для повышения качества вашего кода и обеспечения его стабильности при изменениях.

🛠 Для начала работы установите Pytest с помощью: pip install pytest.

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

🔍 Увеличение производительности с Dask

import dask.dataframe as dd
import pandas as pd

def process_large_csv(file_path: str, column: str) -> pd.Series:
    try:
        # Чтение большого CSV-файла с помощью Dask
        ddf = dd.read_csv(file_path)
        
        # Выполнение операции группировки и агрегации
        result = ddf.groupby(column).size().compute()
        
        return result
    except Exception as e:
        print(f"Ошибка обработки файла: {e}")
        return pd.Series()

# Пример использования
if __name__ == "__main__":
    file_path = 'large_data.csv'
    column = 'category'
    result = process_large_csv(file_path, column)
    print(result)

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

🛠 pip install dask pandas

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

Вышла новая статья на 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 🤖