🤖 Создание чат-бота для 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 🤖

🚀 Функциональные и реактивные подходы в веб-разработке с Flask и React

from flask import Flask, jsonify, request
from functools import partial
from concurrent.futures import ThreadPoolExecutor

app = Flask(__name__)

# Функция для обработки данных
def process_data(data: dict, multiplier: int) -> dict:
    return {k: v * multiplier for k, v in data.items()}

# Обработчик запросов, использующий функциональные принципы
@app.route('/process', methods=['POST'])
def process_request():
    try:
        data = request.json
        if not data:
            raise ValueError("Отсутствуют JSON-данные")

        multiplier = data.get('multiplier', 1)
        if not isinstance(multiplier, int):
            raise ValueError("Multiplier должен быть целым числом")

        values = data.get('values')
        if not isinstance(values, dict):
            raise ValueError("Values должен быть словарем")

        process = partial(process_data, values, multiplier)
        
        with ThreadPoolExecutor() as executor:
            result = executor.submit(process).result()

        return jsonify(result), 200
    except Exception as e:
        return jsonify({'error': str(e)}), 400

# Пример использования: отправка POST-запроса с JSON-данными
# curl -X POST http://localhost:5000/process -H "Content-Type: application/json" -d '{"values": {"a": 1, "b": 2}, "multiplier": 3}'

📌 Этот код демонстрирует использование функциональных принципов в Flask для обработки данных. Применяя functools.partial и ThreadPoolExecutor, мы улучшаем производительность и отзывчивость серверной части приложения. Эти подходы позволяют легко масштабировать и тестировать серверные компоненты, что особенно полезно при высокой нагрузке.

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

🚀 Оптимизация моделей машинного обучения с Optuna

import optuna
from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split, cross_val_score

# Загрузка данных и разделение на обучающую и тестовую выборки
data = load_iris()
X_train, X_test, y_train, y_test = train_test_split(
    data.data, data.target, test_size=0.2, random_state=42
)

def objective(trial):
    n_estimators = trial.suggest_int('n_estimators', 10, 100)
    max_depth = trial.suggest_int('max_depth', 2, 32, log=True)
    min_samples_split = trial.suggest_int('min_samples_split', 2, 10)

    clf = RandomForestClassifier(
        n_estimators=n_estimators,
        max_depth=max_depth,
        min_samples_split=min_samples_split,
        random_state=42
    )
    
    # Кросс-валидация с параллельными вычислениями
    return cross_val_score(clf, X_train, y_train, n_jobs=-1, cv=3).mean()

study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=100)

print(f'Лучшие параметры: {study.best_params}')
print(f'Лучшая точность кросс-валидации: {study.best_value:.2f}')

# Пример использования с найденными гиперпараметрами
best_clf = RandomForestClassifier(**study.best_params)
best_clf.fit(X_train, y_train)
accuracy = best_clf.score(X_test, y_test)
print(f'Точность на тестовом наборе: {accuracy:.2f}')

📌 Этот код показывает, как использовать Optuna для автоматизации настройки гиперпараметров модели RandomForest. Optuna помогает находить оптимальные параметры, чтобы улучшить точность модели. Использование n_jobs=-1 позволяет выполнять кросс-валидацию параллельно, что ускоряет процесс.

🛠 pip install optuna scikit-learn

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