🧪 Декоратор с повтором при исключениях (универсальный ретрай)

import time
import logging
from functools import wraps

logging.basicConfig(level=logging.INFO)

def retry_on_exception(exceptions, retries=3, delay=1, backoff=2):
    """Декоратор для повтора функции при указанных исключениях."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            current_delay = delay
            for attempt in range(1, retries + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    logging.warning(f"Попытка {attempt} не удалась: {e}")
                    if attempt == retries:
                        logging.error("Все попытки исчерпаны.")
                        raise
                    time.sleep(current_delay)
                    current_delay *= backoff
        return wrapper
    return decorator

# Пример использования
@retry_on_exception((ValueError,), retries=4, delay=1)
def flaky():
    import random
    x = random.randint(1, 4)
    if x != 3:
        raise ValueError(f"Неудачное число: {x}")
    print("Успех с числом 3!")

if __name__ == "__main__":
    flaky()

📌 Зачем это нужно:
Повтор функции с нарастающей задержкой — универсальное средство в продакшене: для запросов к БД, API, файловой системе. Оборачивает любую функцию, делает её более устойчивой к временным сбоям.

🛠 Только стандартная библиотека Python ✅

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

🔐 Быстрый клиент OAuth2 Client Credentials (например, для API доступа)

import requests
import time
from threading import Lock

class OAuth2Client:
    def __init__(self, token_url, client_id, client_secret, scope=None):
        self.token_url = token_url
        self.client_id = client_id
        self.client_secret = client_secret
        self.scope = scope
        self.token_data = {}
        self.lock = Lock()

    def _fetch_token(self):
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
        }
        if self.scope:
            payload["scope"] = self.scope

        r = requests.post(self.token_url, data=payload, timeout=10)
        r.raise_for_status()
        token = r.json()
        token["expires_at"] = time.time() + token.get("expires_in", 3600) - 60
        return token

    def get_token(self):
        with self.lock:
            if (
                not self.token_data
                or time.time() >= self.token_data.get("expires_at", 0)
            ):
                self.token_data = self._fetch_token()
            return self.token_data["access_token"]

# Пример использования
if __name__ == "__main__":
    client = OAuth2Client(
        token_url="https://example.com/oauth2/token",
        client_id="your_client_id",
        client_secret="your_client_secret",
        scope="read write",
    )
    token = client.get_token()
    print("Токен:", token)

📌 Зачем это нужно:
Многие API (Stripe, GigaChat, Microsoft, GitHub Apps, etc.) используют OAuth2 с Client Credentials Flow. Этот код кэширует токен и обновляет его только при истечении, экономя запросы и упрощая авторизацию.

🛠

pip install requests

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

🧾 Мини-логгер в файл с ротацией и форматированием

import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path

def setup_logger(log_path: str, max_bytes: int = 1_000_000, backup_count: int = 3) -> logging.Logger:
    """Создаёт логгер с ротацией файла и удобным форматированием."""
    log_path = Path(log_path)
    log_path.parent.mkdir(parents=True, exist_ok=True)

    handler = RotatingFileHandler(
        filename=log_path,
        maxBytes=max_bytes,
        backupCount=backup_count,
        encoding="utf-8"
    )
    formatter = logging.Formatter(
        fmt="[{asctime}] [{levelname}] {message}",
        datefmt="%Y-%m-%d %H:%M:%S",
        style="{"
    )
    handler.setFormatter(formatter)

    logger = logging.getLogger("app_logger")
    logger.setLevel(logging.INFO)
    logger.addHandler(handler)
    logger.propagate = False
    return logger

# Пример использования
if __name__ == "__main__":
    logger = setup_logger("logs/app.log")
    logger.info("Приложение запущено")
    logger.warning("Что-то пошло не так, но не критично")
    logger.error("Ошибка, требующая внимания")

📌 Зачем это нужно:
Лог-файлы с ротацией предотвращают переполнение диска и удобны для отладки в проде. Подходит для серверов, фоновых задач и CLI-утилит. Формат совместим с системами мониторинга.

🛠 Только стандартная библиотека Python ✅

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

📤 Отправка писем через SMTP с безопасным соединением и вложением

import smtplib
from email.message import EmailMessage
from pathlib import Path

def send_email_smtp(
    subject: str,
    body: str,
    to_email: str,
    from_email: str,
    smtp_server: str,
    smtp_port: int,
    username: str,
    password: str,
    attachment_path: Path = None
):
    msg = EmailMessage()
    msg["Subject"] = subject
    msg["From"] = from_email
    msg["To"] = to_email
    msg.set_content(body)

    if attachment_path and Path(attachment_path).is_file():
        with open(attachment_path, "rb") as f:
            data = f.read()
            msg.add_attachment(
                data,
                maintype="application",
                subtype="octet-stream",
                filename=attachment_path.name,
            )

    with smtplib.SMTP_SSL(smtp_server, smtp_port) as smtp:
        smtp.login(username, password)
        smtp.send_message(msg)

# Пример использования
if __name__ == "__main__":
    send_email_smtp(
        subject="📎 Пример письма",
        body="Привет! Это письмо с вложением.",
        to_email="recipient@example.com",
        from_email="sender@example.com",
        smtp_server="smtp.example.com",
        smtp_port=465,
        username="sender@example.com",
        password="your_password",
        attachment_path=Path("report.pdf"),
    )

📌 Зачем это нужно:
Частая задача: отправить лог, отчёт или архив из Python-скрипта. Поддерживает вложения и безопасное соединение. Используется в системах мониторинга, отчётности, алертах.

🛠 Только стандартная библиотека Python (email, smtplib) ✅

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

🌈 Цветной дождь в терминале — как в "Матрице"
ASCII-анимация, которая залипает. Работает даже через SSH и на голом сервере 🖥️

import random
import shutil
import time
import os
import sys

chars = "abcdefghijklmnopqrstuvwxyz0123456789@#$%&*"
width, height = shutil.get_terminal_size((80, 24))
drops = [random.randint(0, height) for _ in range(width)]

def color(code): return f"\033[38;5;{code}m"

print("\033[2J")  # очистка экрана

try:
    while True:
        print("\033[1;40m", end="")  # фон
        for i in range(width):
            drop_y = drops[i]
            if drop_y < height:
                print(f"\033[{drop_y};{i+1}H{color(46 + i % 50)}{random.choice(chars)}", end="")
                drops[i] += 1
            else:
                if random.random() > 0.975:
                    drops[i] = 0
        sys.stdout.flush()
        time.sleep(0.05)
except KeyboardInterrupt:
    print("\033[0m\033[2J")  # сброс цвета и очистка

📌 Что делает:
— Рисует “падающие” буквы в стиле Матрицы прямо в терминале
— Использует ANSI-цвета (поддерживаются в большинстве терминалов)
— Выглядит круто и почти ничего не весит

🛠 Без зависимостей — работает из коробки (Python 3.6+)
Для выхода — Ctrl+C

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

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

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

🔥 Пылающий текст в терминале — ASCII-огонь, как в Doom
Динамическое пламя под вашим текстом. Терминал — в огне! 🔥🔥🔥

import os
import sys
import time
import random

WIDTH = 80
HEIGHT = 30
FIRE_CHARS = " .:-=+*#%@"
TEXT = "HELLO, PYTHON!"  # 🔥 поменяй под себя

fire = [[0 for _ in range(WIDTH)] for _ in range(HEIGHT)]

def ignite():
    for x in range(WIDTH):
        fire[HEIGHT - 1][x] = random.randint(4, 9)

def propagate():
    for y in range(HEIGHT - 1, 0, -1):
        for x in range(WIDTH):
            src = fire[y][x]
            dst = max(0, src - random.randint(0, 2))
            dx = random.randint(-1, 1)
            if 0 <= x + dx < WIDTH:
                fire[y - 1][x + dx] = dst

def draw():
    os.system("clear")
    mid = (WIDTH - len(TEXT)) // 2
    print(" " * mid + TEXT + "\n")
    for y in range(HEIGHT - 1):
        line = "".join(FIRE_CHARS[fire[y][x]] for x in range(WIDTH))
        print(line)

try:
    while True:
        ignite()
        propagate()
        draw()
        time.sleep(0.05)
except KeyboardInterrupt:
    print("\033[0m\033[2J")

📌 Что делает:
— Генерирует эффект горящего ASCII-пламени прямо под вашим текстом
— Вдохновлён огнём из классических демо-сцен и игр (🔥 как в Doom)
— Легко изменить ширину, текст и стиль огня

🛠 Никаких зависимостей — только стандартная библиотека
⚙️ Работает в терминале с ANSI (Linux/macOS/WSL/Windows Terminal)
📎 Залипательно, лампово и безумно кастомизируемо!

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

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

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

🧠 Генератор абстрактных нейросетевых паттернов в PNG
Красивые, псевдоорганические текстуры — будто сгенерированы нейросетью или взяты из sci-fi интерфейса 🧬

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

W, H = 512, 512

def generate_noise(seed=None):
    if seed:
        random.seed(seed)
    base = np.random.rand(H, W) * 255
    for _ in range(4):  # несколько проходов размытия
        base = Image.fromarray(base.astype("uint8")).filter(ImageFilter.GaussianBlur(4))
        base = np.array(base)
    base = 255 * (base - base.min()) / (base.ptp())  # нормализация
    return Image.fromarray(base.astype("uint8")).convert("L")

if __name__ == "__main__":
    img = generate_noise()
    img.save("neuro_texture.png")
    print("✅ Сохранено: neuro_texture.png")

📌 Что делает:
— Генерирует PNG с абстрактным шумом, напоминающим мозг, ИИ, sci-fi интерфейс
— Полезно для фонов, генерации арта, визуального наполнения интерфейсов
— Каждый запуск даёт уникальный результат
— Можно модифицировать фильтры или цветовую палитру

🛠 pip install pillow numpy
Файл сохраняется в neuro_texture.png — можно открыть в любом просмотрщике изображений.

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

🎙️ Аутентичный экран загрузки ZX Spectrum — с шумом и полосами
Вот как *действительно* выглядела загрузка игр с кассеты: шум, хаос, полосы! Прямо как в 1984 🧨📼

import os
import random
import time
import sys

WIDTH = 40
HEIGHT = 24
NOISE_CHARS = [" ", "░", "▒", "▓", "█"]

def clear(): print("\033[2J\033[H", end="")

def draw_noise_frame():
    for _ in range(HEIGHT):
        line = "".join(random.choice(NOISE_CHARS) for _ in range(WIDTH))
        print(line)
    sys.stdout.flush()

def draw_color_bars():
    for _ in range(HEIGHT):
        color = random.randint(41, 47)
        print(f"\033[{color}m{' ' * WIDTH}\033[0m")
    sys.stdout.flush()

try:
    clear()
    for _ in range(5):
        draw_noise_frame()
        time.sleep(0.08)
        clear()
        draw_color_bars()
        time.sleep(0.08)
        clear()
    print("\n\033[1;32mLOAD \"\" ...\033[0m")
    time.sleep(1)
    print("\033[1;34mRUN\n\033[0m")
except KeyboardInterrupt:
    print("\033[0m")

📌 Что делает:
— Симулирует *настоящий* загрузочный экран ZX Spectrum:
белый шум, мерцающие полосы, артефакты и надпись LOAD ""...
— Можно вставить как заставку в ретро-игру или CLI-интерфейс

🛠 Стандартная библиотека. Работает в ANSI-терминалах (Linux, macOS, Windows Terminal)
🧠 Вдохновлено реальным поведением спектрумов при загрузке игр с аудиокассет

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

👾 Ретро-игра «Змейка» в терминале — классика на Python
Минимализм, управление стрелками, и ностальгия по 80-м прямо в консоли! 🐍

import curses
import random

def main(stdscr):
    curses.curs_set(0)
    sh, sw = stdscr.getmaxyx()
    w = curses.newwin(sh, sw, 0, 0)
    w.keypad(1)
    w.timeout(100)

    snk_x = sw//4
    snk_y = sh//2
    snake = [[snk_y, snk_x],
             [snk_y, snk_x-1],
             [snk_y, snk_x-2]]

    food = [random.randint(1, sh-2), random.randint(1, sw-2)]
    w.addch(food[0], food[1], curses.ACS_PI)

    key = curses.KEY_RIGHT
    while True:
        next_key = w.getch()
        key = key if next_key == -1 else next_key

        head = snake[0][:]
        if key == curses.KEY_DOWN:
            head[0] += 1
        elif key == curses.KEY_UP:
            head[0] -= 1
        elif key == curses.KEY_LEFT:
            head[1] -= 1
        elif key == curses.KEY_RIGHT:
            head[1] += 1
        else:
            continue

        if head in snake or head[0] in [0, sh-1] or head[1] in [0, sw-1]:
            break

        snake.insert(0, head)

        if head == food:
            food = None
            while food is None:
                nf = [random.randint(1, sh-2), random.randint(1, sw-2)]
                food = nf if nf not in snake else None
            w.addch(food[0], food[1], curses.ACS_PI)
        else:
            tail = snake.pop()
            w.addch(tail[0], tail[1], ' ')

        w.addch(head[0], head[1], curses.ACS_CKBOARD)

curses.wrapper(main)

📌 Что делает:
— Полноценная змейка с управлением стрелками и стенами в терминале
— Использует модуль curses из стандартной библиотеки
— Классика, проверенная временем и работающая на всех Unix-подобных системах и Windows 10+

🛠 pip install windows-curses для Windows
Запускай в терминале, получай ностальгию и кайф от ретро!

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

Игра из кода выше 👆👆👆

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

🔐 Проверка паролей на утечку через HaveIBeenPwned API

import hashlib
import requests

def is_password_pwned(password: str) -> bool:
    sha1 = hashlib.sha1(password.encode('utf-8')).hexdigest().upper()
    prefix, suffix = sha1[:5], sha1[5:]
    url = f"https://api.pwnedpasswords.com/range/{prefix}"
    res = requests.get(url)
    return any(line.split(':')[0] == suffix for line in res.text.splitlines())

# Пример использования
password = "123456"
if is_password_pwned(password):
    print("⚠️ Пароль найден в базе утечек!")
else:
    print("✅ Пароль не найден в утечках.")

📌 Код использует [HaveIBeenPwned API](https://haveibeenpwned.com/API/v3#SearchingPwnedPasswordsByRange) и алгоритм k-Anonymity, не передавая сам пароль или полный хеш. Идеально для проверки безопасности паролей на клиенте или при регистрации.

🛠 pip install requests

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

📥 Умная загрузка файлов с проверкой кеша

import os
import requests
from hashlib import md5

def download_if_changed(url: str, path: str):
    response = requests.get(url, stream=True)
    content = response.content
    new_hash = md5(content).hexdigest()

    if os.path.exists(path):
        with open(path, 'rb') as f:
            old_hash = md5(f.read()).hexdigest()
        if old_hash == new_hash:
            print("✅ Файл не изменился — загрузка пропущена.")
            return

    with open(path, 'wb') as f:
        f.write(content)
    print(f"⬇️ Файл обновлён: {path}")

# Пример использования
download_if_changed("https://example.com/data.json", "data.json")

📌 Скачивает файл только если он реально изменился, сравнивая MD5-хеш содержимого. Полезно при периодической синхронизации или cron-задачах.

🛠 pip install requests

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

🕵️‍♂️ Поиск TODO и FIXME в проекте

import os
import re

def find_todos(root="."):
    pattern = re.compile(r"#\s*(TODO|FIXME)\b(.*)", re.IGNORECASE)
    for dirpath, _, filenames in os.walk(root):
        for file in filenames:
            if file.endswith(".py"):
                path = os.path.join(dirpath, file)
                with open(path, "r", encoding="utf-8", errors="ignore") as f:
                    for i, line in enumerate(f, 1):
                        match = pattern.search(line)
                        if match:
                            print(f"{path}:{i}: {match.group(0).strip()}")

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

📌 Скрипт находит все # TODO и # FIXME в Python-файлах проекта. Полезен для ревью, CI или подготовки к релизу.

🛠 Без внешних зависимостей.

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

⏳ Таймер функции с логом — просто и полезно

import time
from functools import wraps

def timed(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = fn(*args, **kwargs)
        duration = time.perf_counter() - start
        print(f"⏱ {fn.__name__} выполнена за {duration:.3f} сек.")
        return result
    return wrapper

# Пример использования
@timed
def heavy_task():
    time.sleep(1.5)
    return "done"

heavy_task()

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

🛠 Ничего устанавливать не нужно.

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

🧼 Автоочистка временных файлов старше N часов

import os
import time

def cleanup_temp(folder: str, hours: int = 12):
    now = time.time()
    cutoff = now - hours * 3600
    deleted = 0

    for file in os.listdir(folder):
        path = os.path.join(folder, file)
        if os.path.isfile(path) and os.path.getmtime(path) < cutoff:
            os.remove(path)
            deleted += 1

    print(f"🧹 Удалено {deleted} файлов из {folder} старше {hours} ч.")

# Пример использования
cleanup_temp("/tmp/myapp", hours=6)

📌 Удаляет старые временные файлы из указанной папки. Удобно запускать по расписанию или при старте сервиса, чтобы избежать засорения.

🛠 Без зависимостей. Работает в Linux/macOS/Windows.

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

🚨 Уведомления в Telegram о падениях сервиса

import requests
import time

BOT_TOKEN = 'YOUR_TELEGRAM_BOT_TOKEN'
CHAT_ID = 'YOUR_CHAT_ID'
CHECK_URL = 'https://your-service.com/health'
CHECK_INTERVAL = 60  # секунд

def send_telegram_alert(message: str):
    url = f"https://a.tg.goldica.ir/b0dd72633a60ad0070e10de7b12c5322/bot{BOT_TOKEN}/sendMessage"
    payload = {'chat_id': CHAT_ID, 'text': message}
    requests.post(url, data=payload)

while True:
    try:
        response = requests.get(CHECK_URL, timeout=10)
        if response.status_code != 200:
            send_telegram_alert(f'🚨 Сервис не отвечает! Статус: {response.status_code}')
    except requests.RequestException as e:
        send_telegram_alert(f'🚨 Ошибка при проверке сервиса: {e}')
    time.sleep(CHECK_INTERVAL)

📌 Что делает:
Простой и надёжный watchdog — проверяет URL, и если сервис не отвечает или возвращает ошибку, шлёт уведомление в Telegram.

🛠 pip install requests

Можно положить в supervisor, cron или запускать как systemd-сервис.

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

🧹 Очистка временных файлов по расписанию

import os
import time
import shutil

TEMP_DIR = '/tmp/my_app_cache'
MAX_FILE_AGE = 60 * 60 * 24  # 1 день в секундах
CLEAN_INTERVAL = 60 * 10     # каждые 10 минут

def cleanup_old_files():
    now = time.time()
    for filename in os.listdir(TEMP_DIR):
        filepath = os.path.join(TEMP_DIR, filename)
        try:
            if os.path.isfile(filepath) and now - os.path.getmtime(filepath) > MAX_FILE_AGE:
                os.remove(filepath)
            elif os.path.isdir(filepath) and now - os.path.getmtime(filepath) > MAX_FILE_AGE:
                shutil.rmtree(filepath)
        except Exception as e:
            print(f"Ошибка при удалении {filepath}: {e}")

while True:
    cleanup_old_files()
    time.sleep(CLEAN_INTERVAL)

📌 Что делает:
Автоматически чистит устаревшие файлы и папки из временного каталога, чтобы кэш не разрастался. Полезно для бэкендов, аналитических скриптов и ML-сервисов.

🛠 Никаких зависимостей — работает из коробки.

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

Вышла статья про ChunkTester, будущий мегатул для правильной нарезки документов для RAG 😀

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