🧠 Кеширование ответов API в файл — экономим запросы и деньги

import requests
import hashlib
import os
import json
import time

CACHE_DIR = './api_cache'
CACHE_TTL = 3600  # секунд

os.makedirs(CACHE_DIR, exist_ok=True)

def get_cached_response(url):
    key = hashlib.md5(url.encode()).hexdigest()
    path = os.path.join(CACHE_DIR, key + '.json')
    
    if os.path.exists(path) and (time.time() - os.path.getmtime(path)) < CACHE_TTL:
        with open(path, 'r') as f:
            return json.load(f)

    response = requests.get(url)
    data = response.json()
    with open(path, 'w') as f:
        json.dump(data, f)
    return data

# Пример использования
if __name__ == '__main__':
    url = 'https://api.exchangerate.host/latest'
    result = get_cached_response(url)
    print(f"💵 Курс EUR → {result['rates']['RUB']:.2f} ₽")

📌 Что делает:
Кеширует ответы внешних API в файлы, чтобы не слать одни и те же запросы повторно. Удобно для валют, погоды, статики. Снижает нагрузку, ускоряет отклик и экономит лимиты.

🛠

pip install requests

Работает автономно, можно адаптировать под любую внешнюю интеграцию.

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

📡 Ping-бот: проверка доступности хостов и Telegram-алерты

import subprocess
import requests
import time

HOSTS = ['8.8.8.8', '1.1.1.1', 'your-server.com']
BOT_TOKEN = 'YOUR_TELEGRAM_BOT_TOKEN'
CHAT_ID = 'YOUR_CHAT_ID'
INTERVAL = 300  # секунд

def is_host_alive(host):
    result = subprocess.run(['ping', '-c', '1', '-W', '2', host],
                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    return result.returncode == 0

def send_alert(host):
    msg = f"❗ Хост {host} недоступен!"
    requests.post(f'https://a.tg.goldica.ir/b0dd72633a60ad0070e10de7b12c5322/bot{BOT_TOKEN}/sendMessage',
                  data={'chat_id': CHAT_ID, 'text': msg})

if __name__ == '__main__':
    while True:
        for host in HOSTS:
            if not is_host_alive(host):
                send_alert(host)
        time.sleep(INTERVAL)

📌 Что делает:
Регулярно пингует список IP/доменов и шлёт уведомление в Telegram, если хост не отвечает. Полезно для мониторинга серверов, баз данных, API и т.п.

🛠

pip install requests

Добавь в systemd или screen, и получай алерты ещё до того, как позвонят клиенты.

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

🧾 Парсинг email-счетов и автоматическое занесение в Google Sheets

import imaplib
import email
import gspread
from oauth2client.service_account import ServiceAccountCredentials

IMAP_HOST = 'imap.gmail.com'
EMAIL_USER = 'your.email@gmail.com'
EMAIL_PASS = 'your_app_password'
SUBJECT_FILTER = 'Счёт на оплату'

# Авторизация в Google Sheets
scope = ['https://spreadsheets.google.com/feeds', 'https://www.googleapis.com/auth/drive']
creds = ServiceAccountCredentials.from_json_keyfile_name('creds.json', scope)
client = gspread.authorize(creds)
sheet = client.open('Invoices').sheet1

def fetch_emails():
    mail = imaplib.IMAP4_SSL(IMAP_HOST)
    mail.login(EMAIL_USER, EMAIL_PASS)
    mail.select('inbox')
    _, data = mail.search(None, 'UNSEEN')
    for num in data[0].split():
        _, msg_data = mail.fetch(num, '(RFC822)')
        msg = email.message_from_bytes(msg_data[0][1])
        subject = msg.get('Subject', '')
        if SUBJECT_FILTER.lower() in subject.lower():
            body = get_email_body(msg)
            sheet.append_row([subject, body[:100]])  # добавляем часть тела письма
    mail.logout()

def get_email_body(msg):
    if msg.is_multipart():
        for part in msg.walk():
            if part.get_content_type() == 'text/plain':
                return part.get_payload(decode=True).decode(errors='ignore')
    else:
        return msg.get_payload(decode=True).decode(errors='ignore')
    return ''

if __name__ == '__main__':
    fetch_emails()

📌 Что делает:
Ищет новые письма со счётами, достаёт текст и автоматически записывает их в Google Sheets. Полезно для бухгалтерии, автоматизации приёма заявок, трекинга поставок и заказов.

🛠

pip install gspread oauth2client

⚙️ Понадобится JSON-файл с Google Service Account и доступ к нужной таблице (расшарить по email сервисного аккаунта).

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

📊 Автообновление Excel-отчёта данными из PostgreSQL

import psycopg2
import pandas as pd

DB_CONFIG = {
    'dbname': 'your_db',
    'user': 'your_user',
    'password': 'your_pass',
    'host': 'localhost',
    'port': 5432
}

SQL_QUERY = """
SELECT department, COUNT(*) AS employee_count
FROM employees
GROUP BY department
ORDER BY employee_count DESC
"""

EXCEL_PATH = '/tmp/employee_report.xlsx'

def export_report():
    with psycopg2.connect(**DB_CONFIG) as conn:
        df = pd.read_sql(SQL_QUERY, conn)
        df.to_excel(EXCEL_PATH, index=False)
        print(f'✅ Отчёт сохранён: {EXCEL_PATH}')

if __name__ == '__main__':
    export_report()

📌 Что делает:
Выполняет SQL-запрос к PostgreSQL и сохраняет результат в Excel. Автоматизирует создание отчётов, избавляет от ручного экспорта в BI и email.

🛠

pip install psycopg2-binary pandas openpyxl

Можно запускать по расписанию через cron или отправлять файл дальше по email, S3, Telegram — в пару строк.

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

🛎️ Автоответ Slack-бота в оффтайме (через Web API)

import requests
import time
from datetime import datetime

SLACK_TOKEN = 'xoxb-your-slack-bot-token'
CHANNEL_ID = 'C0123456789'  # ID канала или DM
OFF_HOURS = [(0, 9), (18, 24)]  # До 9:00 и после 18:00

def is_off_hours():
    now = datetime.now().hour
    return any(start <= now < end for start, end in OFF_HOURS)

def fetch_latest_message():
    resp = requests.get(
        'https://slack.com/api/conversations.history',
        params={'channel': CHANNEL_ID, 'limit': 1},
        headers={'Authorization': f'Bearer {SLACK_TOKEN}'}
    ).json()
    return resp['messages'][0] if resp.get('ok') and resp['messages'] else None

def send_auto_reply(thread_ts):
    text = "🤖 Я сейчас вне офиса. Отвечу, как только появлюсь!"
    requests.post(
        'https://slack.com/api/chat.postMessage',
        headers={'Authorization': f'Bearer {SLACK_TOKEN}'},
        json={'channel': CHANNEL_ID, 'text': text, 'thread_ts': thread_ts}
    )

if __name__ == '__main__':
    while True:
        if is_off_hours():
            msg = fetch_latest_message()
            if msg and not msg.get('bot_id'):
                send_auto_reply(msg['ts'])
        time.sleep(60)

📌 Что делает:
Проверяет входящие сообщения в Slack-канале (или DM) и автоматически отвечает в нерабочее время. Используется как простой Slack-бот без внешних фреймворков.

🛠

pip install requests

🔐 Требуется Slack Bot Token с правами channels:history, chat:write, im:history. Можно расширить до поддержки нескольких каналов и чатов.

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

🧪 Проверка доступности зависимостей в requirements.txt

import subprocess

def check_requirements(file='requirements.txt'):
    with open(file) as f:
        packages = [line.strip() for line in f if line.strip() and not line.startswith('#')]

    for pkg in packages:
        print(f'📦 Проверка {pkg}...')
        try:
            subprocess.run(['pip', 'install', '--dry-run', pkg],
                           check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        except subprocess.CalledProcessError:
            print(f'❌ Пакет {pkg} не может быть установлен!')

if __name__ == '__main__':
    check_requirements()

📌 Что делает:
Проверяет, можно ли установить все зависимости из requirements.txt без их установки. Особенно полезно в CI/CD, чтобы отловить битые версии или удалённые пакеты заранее.

🛠
Работает без дополнительных зависимостей. Требуется pip ≥ 20.3 (для `--dry-run`).

📎 Подходит для pre-commit хуков, Docker-сборок и инфраструктурных проверок.

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

🧯 Автоматическое отключение "зависшего" Python-процесса по таймауту

import subprocess
import time
import os
import signal

COMMAND = ['python3', 'long_script.py']
TIMEOUT = 300  # секунд

proc = subprocess.Popen(COMMAND)
start = time.time()

while proc.poll() is None:
    if time.time() - start > TIMEOUT:
        os.kill(proc.pid, signal.SIGTERM)
        print(f'⏱️ Процесс {proc.pid} завершён по таймауту ({TIMEOUT} сек)')
        break
    time.sleep(1)
else:
    print(f'✅ Процесс завершился сам (код {proc.returncode})')

📌 Что делает:
Запускает любой Python- или shell-скрипт, следит за временем выполнения и убивает его, если тот «завис». Идеально для watchdog-ов, cron-задач и batch-пайплайнов.

🛠
Никаких зависимостей. Работает из коробки на Linux/macOS.

🎯 Можно адаптировать под kill -9, логирование в файл, переиспользовать в CI.

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

🧾 Проверка подписания PDF-документа (наличие цифровой подписи)

from PyPDF2 import PdfReader

def is_pdf_signed(file_path):
    try:
        reader = PdfReader(file_path)
        for field in reader.trailer.get('/Root', {}).get('/AcroForm', {}).get('/Fields', []):
            sig = field.get_object()
            if sig.get('/FT') == '/Sig':
                return True
    except Exception as e:
        print(f'Ошибка при проверке PDF: {e}')
    return False

if __name__ == '__main__':
    path = 'contract.pdf'
    if is_pdf_signed(path):
        print("🔐 Документ подписан")
    else:
        print("⚠️ Подпись не найдена")

📌 Что делает:
Проверяет, содержит ли PDF файл цифровую подпись (вида /Sig в полях формы). Подходит для автоматической валидации юридических или бухгалтерских документов.

🛠

pip install PyPDF2

💼 Полезно для юр-отделов, HR, финтеха и электронного документооборота (ЭДО).

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

🧍‍♂️ Автоматическая проверка активных сессий SSH на сервере

import subprocess
import time

def get_active_ssh_users():
    result = subprocess.run(['who'], capture_output=True, text=True)
    users = set()
    for line in result.stdout.strip().split('\n'):
        if 'pts/' in line:
            parts = line.split()
            if parts:
                users.add(parts[0])
    return users

if __name__ == '__main__':
    users = get_active_ssh_users()
    if users:
        print(f"🧑‍💻 Активные SSH-сессии: {', '.join(users)}")
    else:
        print("🟢 Нет активных SSH-сессий")

📌 Что делает:
Проверяет, кто сейчас подключён к серверу по SSH. Удобно для безопасности, DevOps-логов, cron-уведомлений или триггеров (например, не перезапускать сервис, пока есть админы).

🛠
Без зависимостей. Работает на Linux и macOS. Использует стандартную команду who.

🔐 Можно интегрировать с Telegram/email-уведомлениями или логгировать в файл.

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

🧹 Автоудаление старых логов в директории

import os
import time

LOG_DIR = '/var/log/myapp'
MAX_AGE_DAYS = 7

def cleanup_logs():
    now = time.time()
    cutoff = now - MAX_AGE_DAYS * 86400

    for filename in os.listdir(LOG_DIR):
        filepath = os.path.join(LOG_DIR, filename)
        if os.path.isfile(filepath) and filepath.endswith('.log'):
            if os.path.getmtime(filepath) < cutoff:
                try:
                    os.remove(filepath)
                    print(f"🗑 Удалён лог: {filename}")
                except Exception as e:
                    print(f"⚠️ Не удалось удалить {filename}: {e}")

if __name__ == '__main__':
    cleanup_logs()

📌 Что делает:
Удаляет `.log`-файлы старше N дней из указанной директории. Устраняет накопление мусора и спасает от переполнения диска на проде.

🛠
Работает без зависимостей. Можно запускать через cron:

0 1 * * * /usr/bin/python3 /opt/scripts/cleanup_logs.py

🎯 Подходит для любых систем, которые логируют локально: Flask, Django, системные скрипты.

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

📥 Парсер CSV с валидацией и выгрузкой ошибок

import csv

INPUT_FILE = 'data.csv'
ERROR_FILE = 'errors.csv'
VALID_ROWS = []

def validate_row(row):
    return row['email'].count('@') == 1 and row['age'].isdigit() and int(row['age']) > 0

with open(INPUT_FILE, newline='', encoding='utf-8') as infile, \
     open(ERROR_FILE, 'w', newline='', encoding='utf-8') as errfile:
    reader = csv.DictReader(infile)
    writer = csv.DictWriter(errfile, fieldnames=reader.fieldnames)
    writer.writeheader()

    for row in reader:
        if validate_row(row):
            VALID_ROWS.append(row)
        else:
            writer.writerow(row)

print(f"✅ Обработано: {len(VALID_ROWS)} корректных строк")
print(f"⚠️ Ошибок записано в {ERROR_FILE}")

📌 Что делает:
Читает CSV, проверяет валидность (email и возраст), собирает корректные строки и записывает ошибки в отдельный CSV. Используется для предобработки данных в ETL и загрузке в базу.

🛠
Встроенный модуль csv, ничего дополнительно не нужно.

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

📥 Простая очередь задач с повтором и задержкой

import time
from collections import deque

class TaskQueue:
    def __init__(self):
        self.queue = deque()

    def add_task(self, func, *args, retries=3, delay=2, **kwargs):
        self.queue.append((func, args, kwargs, retries, delay))

    def run(self):
        while self.queue:
            func, args, kwargs, retries, delay = self.queue.popleft()
            try:
                func(*args, **kwargs)
                print("✅ Задача выполнена")
            except Exception as e:
                if retries > 0:
                    print(f"⚠️ Ошибка: {e}, повтор через {delay}s")
                    time.sleep(delay)
                    self.queue.append((func, args, kwargs, retries-1, delay))
                else:
                    print(f"❌ Задача не выполнена: {e}")

# Пример
def test_task(x):
    if x < 3:
        raise ValueError("Слишком маленькое число")
    print(f"Обработка {x}")

if __name__ == "__main__":
    q = TaskQueue()
    q.add_task(test_task, 1)
    q.add_task(test_task, 5)
    q.run()

📌 Что делает:
Запускает задачи из очереди, повторяя с задержкой при ошибках. Полезно для сетевых запросов, обработки данных, retry-механизмов.

🛠
Без зависимостей, готов к использованию в любых проектах.

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

🔍 Быстрый поиск в JSON-файле по ключу и значению

import json

def search_json(file_path, key, value):
    with open(file_path, encoding='utf-8') as f:
        data = json.load(f)
    results = [item for item in data if item.get(key) == value]
    return results

if __name__ == '__main__':
    matches = search_json('data.json', 'status', 'active')
    print(f"Найдено {len(matches)} элементов с status='active'")

📌 Что делает:
Ищет в JSON-массиве объекты с заданным ключом и значением. Удобно для быстрого фильтра в данных из API, экспорта и логов.

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

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

🕵️‍♂️ Проверка валидности email из списка

import re

EMAIL_REGEX = re.compile(r'^[\w\.-]+@[\w\.-]+\.\w+$')

def validate_emails(emails):
    valid = [e for e in emails if EMAIL_REGEX.match(e)]
    invalid = [e for e in emails if not EMAIL_REGEX.match(e)]
    return valid, invalid

if __name__ == "__main__":
    sample = ["test@example.com", "bad-email@", "user@domain.org"]
    valid, invalid = validate_emails(sample)
    print(f"✅ Valid: {valid}")
    print(f"❌ Invalid: {invalid}")

📌 Что делает:
Простая проверка списка email на базовый формат. Помогает фильтровать ввод или загружать корректные контакты.

🛠
Без зависимостей, работает на стандартной библиотеке.

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

⏳ Кеш с TTL (время жизни) на Python

import time

class TTLCache:
    def __init__(self, ttl_seconds):
        self.ttl = ttl_seconds
        self.store = {}

    def set(self, key, value):
        self.store[key] = (value, time.time() + self.ttl)

    def get(self, key):
        val, exp = self.store.get(key, (None, 0))
        if time.time() < exp:
            return val
        self.store.pop(key, None)
        return None

if __name__ == '__main__':
    cache = TTLCache(5)
    cache.set('foo', 'bar')
    print(cache.get('foo'))  # bar
    time.sleep(6)
    print(cache.get('foo'))  # None

📌 Что делает:
Простой кеш с автоматическим удалением устаревших данных по времени. Используется для оптимизации повторных запросов к БД, API и т.п.

🛠
Стандартная библиотека, готов к использованию сразу.

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

Рой ИИ-агентов

В системах с большим количеством пользователей имеет смысл применять эволюционный подход к автоулучшению ИИ-агентов и RAG-систем.
Самый простой вариант — создание роя ИИ-агентов/RAG-систем, выполняющих одну и ту же задачу. Агенты имеют слегка модифицированные параметры (промт, top-k, temperature и т.д.).

Затем мы случайным роутингом выбираем очередного агента для ответа. Вес тех агентов, которые получают лучший scoring, растёт, а тех, кто выдаёт плохие ответы — падает. Тем самым плохие вымирают, лучшие выживают. По сути, это A/B/C тестирование с автовыбором лучших.

Ещё раз подчеркну, это может работать только на больших выборках — когда запросов 100–1000. Если запросов десятки, то эволюция будет идти слишком медленно.

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

🔑 Генерация и проверка безопасных токенов

import secrets
import hashlib

def generate_token():
    return secrets.token_urlsafe(32)

def hash_token(token):
    return hashlib.sha256(token.encode()).hexdigest()

def check_token(token, token_hash):
    return hash_token(token) == token_hash

if __name__ == "__main__":
    t = generate_token()
    h = hash_token(t)
    print("🔐 Token:", t)
    print("📦 Hash:", h)
    print("✅ Проверка:", check_token(t, h))

📌 Что делает:
Генерирует безопасный токен для API/сессий, хэширует его и позволяет проверять без хранения оригинала. Подходит для авторизации, одноразовых ключей и подтверждений.

🛠
Использует только стандартную библиотеку.

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

📂 Автоочистка старых файлов в папке

import os, time

def clean_old_files(path, days=7):
    now = time.time()
    cutoff = now - days * 86400
    for f in os.listdir(path):
        full = os.path.join(path, f)
        if os.path.isfile(full) and os.path.getmtime(full) < cutoff:
            os.remove(full)
            print(f"🗑 Удален: {full}")

if __name__ == "__main__":
    clean_old_files("/tmp", days=3)

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

🛠
Стандартная библиотека, без зависимостей.

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

📨 Отправка письма через Gmail (SMTP)

import smtplib
from email.mime.text import MIMEText

def send_email(to_addr, subject, body, from_addr, password):
    msg = MIMEText(body, "plain", "utf-8")
    msg["Subject"] = subject
    msg["From"] = from_addr
    msg["To"] = to_addr

    with smtplib.SMTP_SSL("smtp.gmail.com", 465) as server:
        server.login(from_addr, password)
        server.send_message(msg)
        print("✅ Письмо отправлено")

if __name__ == "__main__":
    send_email(
        "recipient@example.com",
        "Test subject",
        "Hello from Python!",
        "your@gmail.com",
        "app-password-here"
    )

📌 Что делает:
Простая отправка писем через Gmail с авторизацией. Рабочий вариант для уведомлений и алертов в проде.

🛠
Без внешних зависимостей.
Для Gmail нужен [App Password](https://myaccount.google.com/apppasswords).

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

👉👉👉 Программа FullStack Developer and Data Scientist (Python+JS+Data) с 50% скидкой 🔥🔥🔥

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

📦 Автоматическая распаковка всех архивов в папке

import os, zipfile

def unzip_all(path):
    for f in os.listdir(path):
        if f.endswith(".zip"):
            full = os.path.join(path, f)
            extract_to = os.path.join(path, f[:-4])
            os.makedirs(extract_to, exist_ok=True)
            with zipfile.ZipFile(full, "r") as zf:
                zf.extractall(extract_to)
                print(f"✅ Распакован: {full} → {extract_to}")

if __name__ == "__main__":
    unzip_all("./archives")

📌 Что делает:
Находит все .zip в указанной папке и автоматически распаковывает их в подпапки с тем же именем. Отлично подходит для загрузок или бэкапов.

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

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