🔄 Автоматизация обновления данных в базах данных с использованием SQLAlchemy и Python

from sqlalchemy import create_engine, Column, Integer, String, Float, update
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import requests

Base = declarative_base()

class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Float)

def fetch_product_data(api_url: str):
    try:
        response = requests.get(api_url)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Ошибка при запросе API: {e}")
        return []

def update_database(products_data):
    engine = create_engine('sqlite:///products.db')
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    
    for product in products_data:
        stmt = update(Product).where(Product.id == product['id']).values(price=product['price'])
        session.execute(stmt)
    
    session.commit()
    session.close()

# Пример использования
product_api_url = "https://api.example.com/products"
products = fetch_product_data(product_api_url)
if products:
    update_database(products)

📌 Этот код демонстрирует, как использовать SQLAlchemy и Python для автоматизации обновления данных в базах данных.

Скрипт подключается к базе, получает данные из внешнего API, и обновляет цены продуктов в базе данных.

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

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

Вышла статья на habr по следам проблемы циклов 🧐

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

📄 Генерация отчетов в формате PDF с Python и ReportLab

from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas
from reportlab.lib.units import inch

def create_pdf(filename: str):
    try:
        c = canvas.Canvas(filename, pagesize=A4)
        width, height = A4
        
        # Заголовок
        c.setFont("Helvetica-Bold", 16)
        c.drawString(1 * inch, height - 1 * inch, "Отчет о продажах")

        # Таблица данных
        data = [("Продукт", "Количество", "Цена"),
                ("Продукт A", "10", "$100"),
                ("Продукт B", "20", "$200"),
                ("Продукт C", "15", "$150")]

        c.setFont("Helvetica", 12)
        y_position = height - 1.5 * inch
        for row in data:
            x_position = 1 * inch
            for item in row:
                c.drawString(x_position, y_position, item)
                x_position += 2 * inch
            y_position -= 0.5 * inch

        # Завершение
        c.showPage()
        c.save()
        print(f"PDF создан: {filename}")
    except Exception as e:
        print(f"Ошибка при создании PDF: {e}")

# Пример использования
create_pdf("sales_report.pdf")

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

🛠 pip install reportlab

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

📊 Автоматизация планирования задач с использованием Python и Apache Airflow

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

def my_task():
    print("Выполняется задача!")

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 10, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG('my_simple_dag', default_args=default_args, schedule_interval=timedelta(days=1)) as dag:
    start = DummyOperator(task_id='start')
    execute_task = PythonOperator(task_id='execute_task', python_callable=my_task)
    end = DummyOperator(task_id='end')

    start >> execute_task >> end

# Пример использования
# После настройки Apache Airflow, DAG будет доступен в веб-интерфейсе Airflow

📌 Этот код демонстрирует создание простого DAG (Directed Acyclic Graph) в Apache Airflow. DAG состоит из трёх задач: стартовой, выполнения
Python-функции и завершения. Airflow автоматически планирует выполнение этих задач по расписанию, что идеально подходит для автоматизации процессов обработки данных и бизнес-аналитики.

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

Не забываем про Самый быстрый курс по Python 🧐

Сейчас как раз идет какая-то титаническая распродажа на Stepik.

Покупаем себе, друзьям и даже за деньги компании можно купить🔥🔥🔥

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

🔍 Поиск файла с помощью рекурсивного обхода

import os

def find_file(filename: str, search_path: str) -> str:
    try:
        for root, dirs, files in os.walk(search_path):
            if filename in files:
                return os.path.join(root, filename)
        raise FileNotFoundError(f"'{filename}' не найден в '{search_path}'")
    except Exception as e:
        return str(e)

# Пример использования
file_path = find_file('example.txt', '/path/to/search')
print(file_path)

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

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

Добро пожаловать в no-code решения

Не будут эти решения работать в реальных боевых сценариях никогда. Это уже 4-я итерация с 00-х годов по разным темам, сейчас ИИ.

Но как же заманчиво это для клиента. Сам не разбираюсь, нанимать разработчиков не буду.

Надо понимать наивность такого подхода — я тоже сам могу построить себе дом, но я чётко осознаю: либо я потрачу много лет своего времени, разберусь и стану строителем — нафига мне это надо?

Либо я заплачу не временем, а деньгами тем, кто уже разбирается (строителям), и продать мне дом типа n8n по схеме «собери сам» не получится 🤷

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

🔄 Сравнение многопоточности и многопроцессорности в Python: threading vs. multiprocessing

import threading
import time

def io_task():
    """Симуляция задачи, связанной с вводом-выводом."""
    print(f"Начало I/O задачи в потоке {threading.current_thread().name}")
    time.sleep(2)
    print(f"Завершение I/O задачи в потоке {threading.current_thread().name}")

# Пример многопоточности для I/O bound задач
threads = 
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

📌 В этом коде демонстрируется использование многопоточности в Python для выполнения задач, связанных с вводом-выводом.
Модуль threading позволяет параллельно выполнять несколько потоков, что ускоряет операции ввода-вывода. Это полезно для
оптимизации приложений, где много времени занимает ожидание завершения операций ввода-вывода.

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

🌀 Оптимизация памяти с генераторами и yield

def read_large_file(file_path: str):
    """Читает файл построчно, используя генератор."""
    with open(file_path, 'r', encoding='utf-8') as file:
        for line in file:
            yield line.strip()

def process_lines(file_path: str):
    """Обрабатывает строки большого файла."""
    try:
        for line in read_large_file(file_path):
            print(f"Обработка строки: {line[:50]}...")
    except FileNotFoundError:
        print(f"Файл {file_path} не найден.")
    except Exception as e:
        print(f"Произошла ошибка: {e}")

# Пример использования
process_lines('large_file.txt')

📌 Этот код демонстрирует, как генераторы позволяют эффективно обрабатывать большие файлы, не загружая их полностью в
память. Функция read_large_file использует yield для построчного чтения файла, что особенно полезно при работе с
объёмными данными. Такой подход оптимизирует использование памяти и улучшает производительность программы.

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

🔄 Эффективное использование многопоточности и асинхронности: threading vs. asyncio

import asyncio
import threading
import requests
from time import time

def fetch_data_with_threads(urls):
    """Загрузка данных с помощью многопоточности."""
    def fetch(url):
        response = requests.get(url)
        print(f"Получены данные с {url}: {len(response.content)} байт")

    threads = 
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

async def fetch_data_with_asyncio(urls):
    """Загрузка данных с помощью асинхронности."""
    async def fetch(url):
        loop = asyncio.get_event_loop()
        response = await loop.run_in_executor(None, requests.get, url)
        print(f"Получены данные с {url}: {len(response.content)} байт")

    tasks = 
    await asyncio.gather(*tasks)

# Пример использования
urls = ['https://example.com'] * 5

# Многопоточность
start = time()
fetch_data_with_threads(urls)
print(f"Многопоточность: {time() - start:.2f} секунд")

# Асинхронность
start = time()
asyncio.run(fetch_data_with_asyncio(urls))
print(f"Асинхронность: {time() - start:.2f} секунд")

📌 Этот код показывает, как можно загружать данные с нескольких URL одновременно, используя многопоточность и асинхронное
программирование в Python. Модуль threading позволяет запускать несколько потоков для выполнения сетевых запросов, в то
время как asyncio обеспечивает асинхронное выполнение, что может быть более эффективно для задач, связанных с
вводом-выводом. Оба подхода полезны для повышения производительности приложений, работающих с сетью.

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

Ускорение Python кода с использованием Numba

import numpy as np
from numba import jit

# Пример на Numba
@jit(nopython=True)
def sum_of_squares(arr):
    total = 0
    for i in arr:
        total += i * i
    return total

# Пример использования
arr = np.arange(1e6)

# Используем Numba
print("Numba: ", sum_of_squares(arr))

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

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

Люди подстроятся под ботов

Любые технологии на самом деле немного ухудшают пользовательский опыт. Например, когда все перешли на цифровую связь (сотовые, IP телефония, видеоконференции), то автоматически получили задержку связи до 100 мс. Это дико много по сравнению с почти скоростью света, которая была доступна на проводных телефонах, соединённых через АТС.

Но все привыкли. Мы теперь делаем паузу после завершения своего монолога и немного ждём по завершению чужого. Попробуйте сейчас поговорить по проводному телефону — вы офигеете, насколько это удобнее ☎️

То же самое будет с тупыми чат-ботами, которые сейчас все навнедряли и которые всех бесят. Вы всё привыкнете. Вы научитесь говорить так и использовать такие запросы, которые с большей вероятностью будут понятны боту. Люди адаптивны, поэтому подстроятся под ботов 🙇

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

🔍 Кластеризация данных с K-средними

from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
from sklearn.datasets import make_blobs
import numpy as np

def prepare_data(n_samples: int = 300, n_features: int = 2, centers: int = 3) -> np.ndarray:
    # Создание случайных данных с центрами
    X, _ = make_blobs(n_samples=n_samples, n_features=n_features, centers=centers, random_state=42)
    return X

def cluster_data(X: np.ndarray, n_clusters: int = 3) -> np.ndarray:
    # Масштабирование данных
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)

    # Кластеризация методом K-средних
    kmeans = KMeans(n_clusters=n_clusters, random_state=42)
    kmeans.fit(X_scaled)
    return kmeans.labels_

# Пример использования
if __name__ == "__main__":
    data = prepare_data()
    labels = cluster_data(data)
    print("Кластеры:", labels)

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

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

🔥 Распродажа на Stepik закончится 4 декабря в 23:59 🏃🏃‍♀️

👉 Самое время пройти Python: самый быстрый курс 🧐

👉 Или окунутся в мир Data Science на курсе Python Data Science: самый быстрый курс 🤓

👉 Быстро выучить JS на курсу JavaScript: самый быстрый курс 😎

👉 И наконец узнать все сразу в программе курсов Junior Python Developer и Data Scientist +интервью тест 🧑‍🎓

КУПИТЬ КУРСЫ МОЖНО ЗА СЧЕТ РАБОТАДАТЕЛЯ 📌📌📌

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

🚀 Поиск кратчайшего пути с использованием алгоритма A*

from heapq import heappop, heappush

def heuristic(a, b):
    return abs(a[0] - b[0]) + abs(a[1] - b[1])

def a_star_search(graph, start, goal):
    open_set = []
    heappush(open_set, (0, start))
    came_from = {}
    g_score = {start: 0}
    f_score = {start: heuristic(start, goal)}

    while open_set:
        _, current = heappop(open_set)

        if current == goal:
            return reconstruct_path(came_from, current)

        for neighbor in graph.get(current, []):
            tentative_g_score = g_score + graph
            if tentative_g_score < g_score.get(neighbor, float('inf')):
                came_from = current
                g_score = tentative_g_score
                f_score = tentative_g_score + heuristic(neighbor, goal)
                if neighbor not in open_set:
                    heappush(open_set, (f_score, neighbor))

    return None

def reconstruct_path(came_from, current):
    total_path = 
    while current in came_from:
        current = came_from
        total_path.append(current)
    return total_path[::-1]

# Пример использования
graph = {
    (0, 0): {(1, 0): 1, (0, 1): 1},
    (1, 0): {(0, 0): 1, (1, 1): 1, (2, 0): 1},
    (0, 1): {(0, 0): 1, (1, 1): 1},
    (1, 1): {(1, 0): 1, (0, 1): 1, (2, 1): 1},
    (2, 0): {(1, 0): 1, (2, 1): 1},
    (2, 1): {(2, 0): 1, (1, 1): 1}
}

start = (0, 0)
goal = (2, 1)
path = a_star_search(graph, start, goal)
print("Кратчайший путь:", path)

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

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

📊 Прогнозирование временных рядов с использованием рекуррентных нейронных сетей (RNN)

import numpy as np
from sklearn.preprocessing import MinMaxScaler
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout

def prepare_data(series, time_step=1):
    X, y = [], []
    for i in range(len(series) - time_step):
        X.append(series)
        y.append(series)
    return np.array(X), np.array(y)

# Генерация временного ряда
data = np.sin(np.linspace(0, 100, 500)) + np.random.normal(0, 0.1, 500)
scaler = MinMaxScaler(feature_range=(0, 1))
data = scaler.fit_transform(data.reshape(-1, 1)).flatten()

# Подготовка данных
time_step = 10
X, y = prepare_data(data, time_step)
X = X.reshape(X.shape[0], X.shape[1], 1)

# Создание модели RNN
model = Sequential()
model.add(LSTM(units=50, return_sequences=True, input_shape=(time_step, 1)))
model.add(Dropout(0.2))
model.add(LSTM(units=50, return_sequences=False))
model.add(Dropout(0.2))
model.add(Dense(units=1))
model.compile(optimizer='adam', loss='mean_squared_error')

# Обучение модели
model.fit(X, y, epochs=100, batch_size=32, verbose=1)

# Пример использования
test_data = data[-time_step:]
test_data = test_data.reshape(1, -1, 1)
prediction = model.predict(test_data)
print("Прогноз:", scaler.inverse_transform(prediction).flatten()[0])

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

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

Тестирование мета-роев

Простой пример работы мета-роя.

1️⃣ Получен вопрос от пользователя.
2️⃣ RAG-агенты из rag-роя параллельно начали поиск информации в своих знаниях.
3️⃣ Один RAG-агент нашёл у себя инфу, другой — нет.
4️⃣ Запустился мета-рой с входными данными от rag-роя.
5️⃣ User-agent в мета-рое нашёл полезную инфу и вернул её пользователю.

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