Создал свой reddit, там будут посты на английском по теме Agentic AI. Будет больше специфики мировой и многие темы будут выходить раньше, чем в канале @aigentto 😎

Подключайтесь 👉 https://www.reddit.com/r/AIGENTTO/ 🤖

📊 Автоматизация отчетности с Pandas и Matplotlib

import pandas as pd
import matplotlib.pyplot as plt
from smtplib import SMTP
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.image import MIMEImage

def generate_report(data_file: str, email: str):
    try:
        # Чтение и обработка данных из Excel-файла
        df = pd.read_excel(data_file, engine='openpyxl')
        summary = df.describe()

        # Создание визуализации
        plt.figure(figsize=(10, 6))
        df.plot(kind='line')
        plt.title('Обзор данных о продажах')
        plt.xlabel('Дата')
        plt.ylabel('Продажи')
        plt.grid(True)
        plt.tight_layout()

        # Сохранение графика
        plot_file = 'sales_report.png'
        plt.savefig(plot_file)

        # Отправка отчета по электронной почте
        send_email(email, plot_file, summary.to_string())
    except Exception as e:
        print(f"Произошла ошибка: {e}")

def send_email(to_email: str, image_path: str, summary: str):
    msg = MIMEMultipart()
    msg['Subject'] = 'Автоматический отчет по продажам'
    msg['From'] = 'your_email@example.com'
    msg['To'] = to_email

    text = MIMEText(f"Добрый день,\n\nПожалуйста, найдите отчет по продажам ниже:\n\n{summary}", 'plain')
    msg.attach(text)

    with open(image_path, 'rb') as img:
        img_data = img.read()
        image = MIMEImage(img_data, name='Отчет о продажах')
        msg.attach(image)

    with SMTP('smtp.example.com', 587) as smtp:
        smtp.starttls()
        smtp.login('your_email@example.com', 'your_password')
        smtp.sendmail(msg['From'], msg['To'], msg.as_string())

# Пример использования
generate_report('sales_data.xlsx', 'recipient@example.com')

📌 Этот код выполняет автоматическую генерацию отчетов на основе данных из Excel и отправляет их по электронной почте. Используя Pandas для анализа данных и Matplotlib для их визуализации, он создает график и отправляет его вместе с текстовым резюме. Такой подход полезен для регулярной отчетности в бизнесе и аналитике.

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

🤖 Создание чат-бота для Telegram с использованием TeleBot и OpenAI GPT

import telebot
import openai
import os

# Установка токенов из окружения
TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN')
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')

# Инициализация клиентов
bot = telebot.TeleBot(TELEGRAM_TOKEN)
openai.api_key = OPENAI_API_KEY

def generate_response(prompt: str) -> str:
    try:
        response = openai.Completion.create(
            engine="text-davinci-003",
            prompt=prompt,
            max_tokens=150
        )
        return response.choices[0].text.strip() if response.choices else "Ошибка: пустой ответ"
    except Exception as e:
        return f"Ошибка при генерации ответа: {e}"

@bot.message_handler(func=lambda message: True)
def handle_message(message):
    chat_id = message.chat.id
    user_message = message.text
    bot_response = generate_response(user_message)
    bot.send_message(chat_id, bot_response)

# Запуск бота
bot.polling(none_stop=True)

# Пример использования:
# Отправьте сообщение вашему боту в Telegram, и он ответит, используя OpenAI GPT.

📌 Этот код создает простого чат-бота для Telegram, использующего OpenAI GPT для генерации ответов. Бот принимает текст от пользователя, отправляет его в OpenAI API и возвращает ответ. Такой бот может быть полезен для автоматизации общения и развлечения.

🛠 Убедитесь, что установлены зависимости: pip install pyTelegramBotAPI openai

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

Прогнозирование погоды с использованием OpenWeatherMap API

import requests
from datetime import datetime

def get_weather(api_key: str, city: str) -> dict:
    url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric"
    try:
        response = requests.get(url)
        response.raise_for_status()
        data = response.json()
        weather = {
            "city": data["name"],
            "temperature": data["main"]["temp"],
            "description": data["weather"][0]["description"],
            "datetime": datetime.fromtimestamp(data["dt"]).strftime('%Y-%m-%d %H:%M:%S')
        }
        return weather
    except requests.exceptions.HTTPError as http_err:
        print(f"HTTP ошибка: {http_err}")
    except Exception as err:
        print(f"Произошла ошибка: {err}")

# Пример использования
api_key = "ваш_api_ключ"  # Замените на ваш ключ API
city = "Moscow"
weather_info = get_weather(api_key, city)
if weather_info:
    print(f"Погода в {weather_info['city']} на {weather_info['datetime']}:")
    print(f"{weather_info['temperature']}°C, {weather_info['description']}")

📌 Этот код позволяет получить текущую погоду для указанного города, используя OpenWeatherMap API. Он делает HTTP-запрос, обрабатывает JSON-ответ и выводит информацию о температуре и погодных условиях. Полезно для создания погодных приложений или ботов.

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

⚡️ Параллельная обработка данных с concurrent.futures

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

def fetch_url(url: str) -> str:
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return response.text
    except requests.RequestException as e:
        return f"Ошибка при получении {url}: {e}"

def fetch_all(urls: list) -> dict:
    results = {}
    with ThreadPoolExecutor(max_workers=5) as executor:
        future_to_url = {executor.submit(fetch_url, url): url for url in urls}
        for future in as_completed(future_to_url):
            url = future_to_url[future]
            try:
                results[url] = future.result()
            except Exception as exc:
                results[url] = f"Ошибка: {exc}"
    return results

# Пример использования
urls = [
    "https://www.example.com",
    "https://www.python.org",
    "https://www.github.com",
]
results = fetch_all(urls)
for url, content in results.items():
    print(f"URL: {url}, Длина контента: {len(content)}")

📌 Этот код демонстрирует, как использовать ThreadPoolExecutor из модуля concurrent.futures для параллельной загрузки веб-страниц. Функция fetch_all принимает список URL-адресов и возвращает словарь, где ключи — это адреса, а значения — загруженные данные или сообщения об ошибках. Это может быть полезно для ускорения сетевых операций, когда нужно быстро обработать множество запросов.

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

📚 Основы обработки текста с NLTK

import nltk
from nltk.tokenize import word_tokenize
from nltk.corpus import stopwords
from collections import Counter

# Загрузка необходимых ресурсов
nltk.download('punkt')
nltk.download('stopwords')

def analyze_text(text: str) -> dict:
    try:
        # Токенизация текста
        words = word_tokenize(text.lower())

        # Удаление стоп-слов
        stop_words = set(stopwords.words('english'))
        filtered_words = [word for word in words if word.isalnum() and word not in stop_words]

        # Частотный анализ
        word_freq = Counter(filtered_words)
        
        return dict(word_freq)
    except Exception as e:
        print(f"Ошибка при анализе текста: {e}")
        return {}

# Пример использования
text = "NLTK is a leading platform for building Python programs to work with human language data."
result = analyze_text(text)
print(result)

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

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

🔧 Функция-комбинатор для управления потоком данных

from functools import reduce
from typing import Callable

def compose(*functions: Callable) -> Callable:
    def compose_two(f: Callable, g: Callable) -> Callable:
        return lambda x: f(g(x))
    return reduce(compose_two, functions, lambda x: x)

# Пример использования
def add_one(x: int) -> int:
    return x + 1

def double(x: int) -> int:
    return x * 2

def square(x: int) -> int:
    return x ** 2

process = compose(add_one, double, square)
result = process(3)  # (3 + 1) * 2 = 8; 8 ** 2 = 64
print(result)  # Вывод: 64

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

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

🤖 Создание простого чат-бота для технической поддержки

import nltk
from nltk.chat.util import Chat, reflections

# Определение пар шаблонов и ответов
pairs = [
    [r"привет|здравствуй|добрый день", ["Здравствуйте!", "Привет! Как я могу помочь вам сегодня?"]],
    [r"мне нужна помощь с (.*)", ["Конечно, я могу помочь с %1. Что именно вас интересует?"]],
    [r"спасибо|благодарю", ["Всегда рад помочь!", "Не за что!"]],
    [r"пока|до свидания", ["До свидания! Хорошего дня!"]],
]

# Создание чат-бота
def create_chatbot():
    return Chat(pairs, reflections)

# Пример использования
if __name__ == "__main__":
    nltk.download('punkt')  # Загрузка необходимых данных NLTK
    chatbot = create_chatbot()
    print("Бот: Здравствуйте! Я чат-бот технической поддержки.")
    while True:
        user_input = input("Вы: ")
        if user_input.lower() in ["пока", "до свидания"]:
            print("Бот:", chatbot.respond(user_input))
            break
        response = chatbot.respond(user_input)
        print("Бот:", response or "Извините, я не совсем понял. Можете переформулировать вопрос?")

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

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

🔄 Оптимизация SQL запросов с SQLAlchemy

from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError

def get_optimized_data(engine, query: str):
    try:
        with engine.connect() as connection:
            # Выполнение SQL-запроса и получение всех результатов
            result = connection.execute(text(query))
            return result.fetchall()
    except SQLAlchemyError as e:
        print(f"Ошибка выполнения запроса: {e}")
        return None

# Настройка подключения к базе данных
engine = create_engine('sqlite:///example.db')

# Пример оптимизированного запроса
optimized_query = """
SELECT * FROM users
WHERE age > 30
ORDER BY last_login DESC
LIMIT 10
"""

# Получение данных
data = get_optimized_data(engine, optimized_query)
for row in data:
    print(row)

📌 Этот код показывает, как использовать SQLAlchemy для выполнения оптимизированных SQL-запросов. Функция get_optimized_data выполняет запросы с обработкой ошибок, что повышает устойчивость приложения. Использование индексов и ограничений (`LIMIT`) помогает улучшить производительность при работе с большими объемами данных.

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

📊 Автоматизация анализа данных в реальном времени с Apache Kafka

from kafka import KafkaProducer, KafkaConsumer
from json import dumps, loads
import logging

logging.basicConfig(level=logging.INFO)

def create_producer(broker: str):
    return KafkaProducer(
        bootstrap_servers=[broker],
        value_serializer=lambda x: dumps(x).encode('utf-8')
    )

def create_consumer(broker: str, topic: str, group_id: str):
    return KafkaConsumer(
        topic,
        bootstrap_servers=[broker],
        auto_offset_reset='earliest',
        group_id=group_id,
        value_deserializer=lambda x: loads(x.decode('utf-8'))
    )

def send_data(producer, topic: str, data: dict):
    try:
        producer.send(topic, value=data)
        producer.flush()
        logging.info(f"Отправлены данные: {data}")
    except Exception as e:
        logging.error(f"Ошибка отправки данных: {e}")

def consume_data(consumer):
    for message in consumer:
        try:
            logging.info(f"Получены данные: {message.value}")
        except Exception as e:
            logging.error(f"Ошибка обработки сообщения: {e}")

# Пример использования
broker_address = 'localhost:9092'
producer = create_producer(broker_address)
consumer = create_consumer(broker_address, 'test-topic', 'test-group')

# Отправка данных
send_data(producer, 'test-topic', {'key': 'value'})

# Обработка данных
consume_data(consumer)

📌 Этот код демонстрирует простую реализацию продюсера и потребителя в Apache Kafka на Python. С помощью Kafka вы можете отправлять и получать сообщения в реальном времени, что идеально подходит для анализа потоковых данных. Код показывает, как использовать Kafka для создания систем мониторинга и оповещения.

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

Будет еще много типов агентов на BlackBoard

На BlackBoard можно создать любое количество и любые типы агентов.

В планах добавить:
🕷️ spider-agent — для поиска в интернете
🧭 lead-agent — для координации микро-роев агентов
🔌 API-agent — для доступа к внешним API
🗃️ sql-agent — для доступа к БД
💬 stream-user-agent — для стриминга ответа пользователю
🧠 memory-agent — для памяти всех переписок
🧩 control-agent — для контроля выполнения мета-задач
🛡️ security-agent — для контроля безопасности переписки агентов

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

📅 Автоматизированные уведомления о событиях в календаре

import os
import datetime
import requests
from google.oauth2 import service_account
from googleapiclient.discovery import build

def get_upcoming_events(service, calendar_id='primary'):
    now = datetime.datetime.utcnow().isoformat() + 'Z'
    events_result = service.events().list(calendarId=calendar_id, timeMin=now,
                                          maxResults=10, singleEvents=True,
                                          orderBy='startTime').execute()
    return events_result.get('items', [])

def send_telegram_message(chat_id, message, bot_token):
    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
    payload = {'chat_id': chat_id, 'text': message}
    response = requests.post(url, data=payload)
    response.raise_for_status()

def main():
    SERVICE_ACCOUNT_FILE = 'path/to/credentials.json'
    SCOPES = ['https://www.googleapis.com/auth/calendar.readonly']
    
    credentials = service_account.Credentials.from_service_account_file(
        SERVICE_ACCOUNT_FILE, scopes=SCOPES)
    service = build('calendar', 'v3', credentials=credentials)

    events = get_upcoming_events(service)
    if not events:
        print('No upcoming events found.')
        return

    bot_token = os.getenv('TELEGRAM_BOT_TOKEN')
    chat_id = os.getenv('TELEGRAM_CHAT_ID')

    for event in events:
        start = event['start'].get('dateTime', event['start'].get('date'))
        message = f"Upcoming Event: {event['summary']} at {start}"
        send_telegram_message(chat_id, message, bot_token)

if __name__ == '__main__':
    main()

📌 Этот скрипт автоматически извлекает предстоящие события из Google Calendar и отправляет уведомления в Telegram. Он использует Google Calendar API для доступа к данным календаря и Telegram Bot API для доставки сообщений. Полезен для управления временем и повышения продуктивности.

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

🚨 Последняя распродажа следующих курсов, после этой распродажи доступ к этим курсам будет закрыт (останется только у тех, кто купил):

👉 Python: самый быстрый курс 🔥🔥🔥
👉 Python Data Science: самый быстрый курс 🔥
👉 JavaScript: самый быстрый курс 🔥🔥

⚡️⚡️⚡️ Только до 31 января ⚡️⚡️⚡️

Размышляем над тупиками нейросетей тут https://habr.com/ru/articles/989054/ и тут https://www.reddit.com/r/AIGENTTO/comments/1qnddav/agi_is_not_coming_yet/

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

📊 Реальный анализ данных с MongoDB и PyMongo

import pymongo
from pymongo.errors import ConnectionError, OperationFailure, PyMongoError
from datetime import datetime

def connect_to_mongo(uri: str):
    try:
        client = pymongo.MongoClient(uri)
        client.admin.command('ping')  # Проверяем подключение
        print("Успешное подключение к MongoDB")
        return client
    except (ConnectionError, OperationFailure) as e:
        print(f"Ошибка подключения: {e}")
        return None

def store_social_data(db, collection_name: str, data: dict):
    try:
        collection = db[collection_name]
        data['timestamp'] = datetime.now()  # Добавляем временную метку
        collection.insert_one(data)  # Сохраняем данные в коллекцию
        print(f"Данные сохранены в коллекцию {collection_name}")
    except PyMongoError as e:
        print(f"Ошибка сохранения данных: {e}")

# Пример использования
if __name__ == "__main__":
    mongo_uri = "mongodb://localhost:27017/"
    client = connect_to_mongo(mongo_uri)

    if client:
        db = client['social_data']
        sample_data = {
            "user": "john_doe",
            "platform": "Twitter",
            "content": "Python 3.9 вышел!",
            "likes": 150
        }
        store_social_data(db, "posts", sample_data)

📌 Этот код показывает, как подключиться к MongoDB и сохранить данные из социальных сетей в реальном времени. Используя PyMongo, создаем подключение к базе данных и сохраняем пример данных с временной меткой. Это полезно для маркетинга и аналитики, отслеживая активность в социальных сетях.

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

🎬 Создание системы рекомендаций с Surprise и scikit-learn

from surprise import Dataset, Reader, SVD
from surprise.model_selection import cross_validate
import pandas as pd

# Загрузка данных
data_dict = {'item': [1, 2, 3, 1, 2],
             'user': [1, 1, 1, 2, 2],
             'rating': [4, 5, 3, 2, 4]}
df = pd.DataFrame(data_dict)
reader = Reader(rating_scale=(1, 5))
data = Dataset.load_from_df(df[['user', 'item', 'rating']], reader)

# Создание и обучение модели SVD
model = SVD()
cross_validate(model, data, measures=['RMSE', 'MAE'], cv=2, verbose=True)

# Пример предсказания
trainset = data.build_full_trainset()
model.fit(trainset)
user_id = 1
item_id = 3
prediction = model.predict(user_id, item_id)
print(f"Предсказанный рейтинг для пользователя {user_id} и товара {item_id}: {prediction.est:.2f}")

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

🛠 pip install surprise pandas

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

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

🚀 Последний шанс купить следующие курсы:

👉 Python: самый быстрый курс 🔥🔥🔥
👉 Python Data Science: самый быстрый курс 🔥
👉 JavaScript: самый быстрый курс 🔥🔥

⚡️⚡️⚡️ Только до 31 января ⚡️⚡️⚡️

🚀 Внедрение CI/CD для Python-проектов с GitHub Actions

name: CI/CD Pipeline

on:
  push:
    branches:
      - main
  pull_request:
    branches:
      - main

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v2

      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: 3.8

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt

      - name: Run tests
        run: pytest

  deploy:
    runs-on: ubuntu-latest
    needs: test
    steps:
      - name: Checkout code
        uses: actions/checkout@v2

      - name: Deploy to Production
        run: echo "Deploying application..."

📌 Этот YAML-файл описывает базовый CI/CD конвейер для Python-проекта с использованием GitHub Actions. Он включает две задачи: тестирование и развертывание. При каждом коммите в ветку main запускаются тесты с помощью pytest, и в случае их успешного прохождения выполняется развертывание. Это помогает автоматизировать процессы разработки и уменьшает вероятность ошибок при развертывании.

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

⚡️⚡️⚡️ Доступ к продаже следующих курсов закрывается сегодня ⚡️⚡️⚡️

👉 Python: самый быстрый курс 🔥🔥🔥
👉 Python Data Science: самый быстрый курс 🔥
👉 JavaScript: самый быстрый курс 🔥🔥

🗺️ Обработка и визуализация геоданных с GeoPandas

import geopandas as gpd
from shapely.geometry import Point

def load_geodata(file_path: str) -> gpd.GeoDataFrame:
    try:
        return gpd.read_file(file_path)
    except FileNotFoundError:
        print("Файл не найден.")
    except Exception as e:
        print(f"Ошибка при чтении файла: {e}")
    return gpd.GeoDataFrame()

def create_point_geodataframe(latitudes: list, longitudes: list) -> gpd.GeoDataFrame:
    if len(latitudes) != len(longitudes):
        print("Списки широт и долгот должны быть одинаковой длины.")
        return gpd.GeoDataFrame()
    geometry = [Point(xy) for xy in zip(longitudes, latitudes)]
    return gpd.GeoDataFrame(geometry=geometry, crs="EPSG:4326")

def filter_data_by_region(gdf: gpd.GeoDataFrame, region_name: str) -> gpd.GeoDataFrame:
    try:
        return gdf[gdf['region'] == region_name]
    except KeyError:
        print(f"Колонка 'region' отсутствует в данных.")
    return gpd.GeoDataFrame()

# Пример использования
if __name__ == "__main__":
    # Загрузка геоданных
    geodata = load_geodata("example.geojson")
    
    # Создание GeoDataFrame из координат
    latitudes = [37.7749, 34.0522]
    longitudes = [-122.4194, -118.2437]
    points = create_point_geodataframe(latitudes, longitudes)
    
    # Фильтрация по региону
    filtered_data = filter_data_by_region(geodata, "California")
    print(filtered_data)

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

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

🚀 Оптимизация кода с мемоизацией

from functools import lru_cache

@lru_cache(maxsize=None)
def fibonacci(n: int) -> int:
    """Рассчитывает n-е число Фибоначчи с мемоизацией."""
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# Пример использования
print(fibonacci(10))  # Вывод: 55

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

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