📊 Автоматизация обработки данных в Pandas

import pandas as pd

def clean_data(file_path: str) -> pd.DataFrame:
    try:
        df = pd.read_csv(file_path)
        df.dropna(inplace=True)
        df.drop_duplicates(inplace=True)
        df.columns = df.columns.str.strip().str.lower().str.replace(' ', '_')
        return df
    except FileNotFoundError:
        print(f"Файл не найден: {file_path}")
        return pd.DataFrame()
    except pd.errors.EmptyDataError:
        print("Файл пустой.")
        return pd.DataFrame()
    except Exception as e:
        print(f"Ошибка при обработке файла: {e}")
        return pd.DataFrame()

# Пример использования
dataframe = clean_data('data.csv')
print(dataframe.head())

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

🛠 pip install pandas

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

Агенты общаются как люди (без оркестрации)

Вот пример общения агентов без оркестрации — они как люди сами решают, что делать и когда делать, а когда не делать...

1️⃣ Пришел запрос от пользователя.

2️⃣ Проверка пользователя на анонимность (если залогинен, то разрешаем).

3️⃣ Агент не блокировал НЕ персональный вопрос.

4️⃣ Агент-юрист понял, что вопрос не по его теме.

5️⃣ Агенту HR разрешено всё, он попытался найти инфу в документах компании, но не смог 😀.

6️⃣ Forward-agent понял, что на вопрос никто не ответил, и приказал его отправить людям 👨.

7️⃣ Агент task tracker взял и создал задачу на человека с этим вопросом.

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

🔧 Автоматизация управления облачными ресурсами с использованием AWS SDK и Python

import boto3
from botocore.exceptions import NoCredentialsError, PartialCredentialsError

def create_ec2_instance(image_id: str, instance_type: str, key_name: str) -> str:
    """Создает EC2-инстанс с указанными параметрами."""
    ec2 = boto3.resource('ec2')
    try:
        instance = ec2.create_instances(
            ImageId=image_id,
            InstanceType=instance_type,
            KeyName=key_name,
            MinCount=1,
            MaxCount=1
        )[0]
        return f"EC2 Instance {instance.id} создан."
    except (NoCredentialsError, PartialCredentialsError):
        return "Ошибка: Учетные данные AWS не найдены."
    except Exception as e:
        return f"Ошибка при создании инстанса: {e}"

# Пример использования
print(create_ec2_instance('ami-12345678', 't2.micro', 'my-key-pair'))

📌 Этот код демонстрирует автоматизацию управления облачными ресурсами AWS с использованием AWS SDK для Python (boto3). Скрипт позволяет создавать EC2-инстансы, что может значительно ускорить работу с ресурсами AWS для разработчиков и системных администраторов.

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

🔄 Автоматизация обновления данных в базах данных с использованием 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 🤖