Kodduu Python

from textblob import TextBlob
from typing import Tuple

def analyze_text(text: str) -> Tuple[float, str]:
    """Анализирует текст: sentiment, полярность и субъективность."""
    blob = TextBlob(text)
    
    # Sentiment analysis
    sentiment = blob.sentiment
    polarity = sentiment.polarity  # от -1 (отрицательный) до 1 (положительный)
    
    if polarity > 0.1:
        mood = "Положительный"
    elif polarity < -0.1:
        mood = "Отрицательный"
    else:
        mood = "Нейтральный"
    
    return polarity, mood

if __name__ == "__main__":
    try:
        texts = [
            "Python — замечательный язык для разработки!",
            "Сегодня ужасная погода, всё идёт не так.",
            "Средний отчёт: ничего особенного.",
            "Этот продукт изменил мою жизнь к лучшему!"
        ]
        
        for t in texts:
            polarity, mood = analyze_text(t)
            print(f"Текст: {t}")
            print(f"Настроение: {mood} (полярность: {polarity:.2f})\n")
    except Exception as e:
        print(f"Ошибка: {e}")

from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
import numpy as np

def train_iris_classifier() -> float:
    """Обучает классификатор на датасете Iris."""
    iris = load_iris()
    X, y = iris.data, iris.target
    
    # Разделяем на train/test
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.3, random_state=42
    )
    
    # Модель случайного леса
    clf = RandomForestClassifier(n_estimators=100, random_state=42)
    clf.fit(X_train, y_train)
    
    # Предсказания и точность
    y_pred = clf.predict(X_test)
    accuracy = accuracy_score(y_test, y_pred)
    
    print(f"Точность модели: {accuracy:.2%}")
    print(f"Пример предсказания: {iris.target_names[clf.predict([X_test[0]])[0]]}")
    return accuracy

if __name__ == "__main__":
    try:
        train_iris_classifier()
    except Exception as e:
        print(f"Ошибка: {e}")

import networkx as nx
from typing import List, Tuple

def build_and_analyze_graph() -> None:
    """Пример работы с графами: города и дороги."""
    G = nx.Graph()
    
    # Добавляем узлы (города) и рёбра (дороги)
    G.add_edges_from([
        ("Москва", "Питер", {"distance": 700}),
        ("Москва", "Казань", {"distance": 800}),
        ("Питер", "Новгород", {"distance": 200}),
        ("Казань", "Екатеринбург", {"distance": 900}),
        ("Новгород", "Екатеринбург", {"distance": 1200}),
    ])
    
    # Кратчайший путь
    path = nx.shortest_path(G, "Москва", "Екатеринбург", weight="distance")
    distance = nx.shortest_path_length(G, "Москва", "Екатеринбург", weight="distance")
    
    print("📍 Кратчайший путь Москва → Екатеринбург:", " → ".join(path))
    print(f"Расстояние: {distance} км\n")
    
    # Центральность узлов
    centrality = nx.degree_centrality(G)
    print("🔝 Самые связанные города:")
    for city, cent in sorted(centrality.items(), key=lambda x: x[1], reverse=True):
        print(f"  • {city}: {cent:.2f}")

if __name__ == "__main__":
    try:
        build_and_analyze_graph()
    except Exception as e:
        print(f"Ошибка: {e}")

from sympy import symbols, solve, diff, simplify, integrate
from sympy.abc import x

def math_demo():
    """Примеры возможностей SymPy."""
    # Решение уравнения
    eq = x**2 - 5*x + 6
    roots = solve(eq, x)
    ic("Корни уравнения x²-5x+6=0:", roots)
    
    # Производная
    f = x**3 + 2*x**2 - 5*x + 3
    derivative = diff(f, x)
    ic("Производная:", derivative)
    
    # Упрощение выражения
    expr = (x**2 + 2*x + 1) / (x + 1)
    simplified = simplify(expr)
    ic("Упрощённое выражение:", simplified)
    
    # Интеграл
    integral = integrate(x**2 + 3*x, x)
    ic("Интеграл:", integral)

if __name__ == "__main__":
    try:
        math_demo()
    except Exception as e:
        print(f"Ошибка: {e}")

import asyncio
import time
from typing import List

async def fetch_data(url: str) -> str:
    """Имитация асинхронного запроса к API."""
    print(f"🔄 Запрос к {url}...")
    await asyncio.sleep(1)  # имитация сети
    return f"Данные с {url} получены!"

async def main() -> None:
    """Запускает несколько асинхронных задач параллельно."""
    urls = [
        "https://api.example.com/users",
        "https://api.example.com/posts",
        "https://api.example.com/comments",
        "https://api.example.com/products"
    ]
    
    start = time.time()
    # gather запускает все задачи одновременно
    results = await asyncio.gather(*[fetch_data(url) for url in urls])
    
    print(f"\n✅ Получено {len(results)} ответов за {time.time() - start:.2f} сек")
    for result in results:
        print(f"  • {result}")

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        print(f"Ошибка: {e}")

from bs4 import BeautifulSoup
from typing import List, Dict

# Пример HTML (можно заменить на requests.get(url).text)
html_example = """
<html>
<head><title>Каталог товаров</title></head>
<body>
    <h1>Наши продукты</h1>
    <div class="product">
        <h2>Ноутбук ASUS</h2>
        <p class="price">74990 ₽</p>
        <a href="/product/123">Подробнее</a>
    </div>
    <div class="product">
        <h2>Смартфон Xiaomi</h2>
        <p class="price">32990 ₽</p>
        <a href="/product/456">Подробнее</a>
    </div>
</body>
</html>
"""

def parse_products(html: str) -> List[Dict]:
    """Парсит HTML и извлекает товары по классам и тегам."""
    soup = BeautifulSoup(html, "html.parser")
    products = []
    
    for item in soup.find_all("div", class_="product"):
        try:
            product = {
                "title": item.find("h2").get_text(strip=True),
                "price": item.find("p", class_="price").get_text(strip=True),
                "link": item.find("a")["href"]
            }
            products.append(product)
        except AttributeError:
            continue  # пропускаем некорректные блоки
    return products

if __name__ == "__main__":
    try:
        products = parse_products(html_example)
        print(f"✅ Найдено товаров: {len(products)}\n")
        for p in products:
            print(f"📦 {p['title']}")
            print(f"   Цена: {p['price']}")
            print(f"   Ссылка: {p['link']}\n")
    except Exception as e:
        print(f"❌ Ошибка парсинга: {e}")

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List

def cpu_heavy_task(n: int) -> int:
    """Имитация тяжёлой CPU-задачи (вычисление суммы квадратов)."""
    return sum(i * i for i in range(n))

def run_sequential(numbers: List[int]) -> List[int]:
    """Последовательное выполнение (медленно)."""
    return [cpu_heavy_task(n) for n in numbers]

def run_parallel(numbers: List[int], max_workers: int = 4) -> List[int]:
    """Параллельное выполнение с ThreadPoolExecutor."""
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_n = {executor.submit(cpu_heavy_task, n): n for n in numbers}
        for future in as_completed(future_to_n):
            try:
                results.append(future.result())
            except Exception as e:
                print(f"Ошибка в задаче {future_to_n[future]}: {e}")
    return results

if __name__ == "__main__":
    try:
        numbers = [300_000, 400_000, 500_000, 600_000]
        
        print("⏱️  Последовательно:")
        start = time.time()
        seq_result = run_sequential(numbers)
        print(f"Время: {time.time() - start:.2f} сек | Результаты: {seq_result[:2]}...\n")
        
        print("⚡ Параллельно (4 потока):")
        start = time.time()
        par_result = run_parallel(numbers)
        print(f"Время: {time.time() - start:.2f} сек | Результаты: {par_result[:2]}...")
        
        print("\n✅ В параллельном режиме значительно быстрее!")
    except Exception as e:
        print(f"Ошибка: {e}")

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List

def cpu_heavy_task(n: int) -> int:
    """Имитация тяжёлой CPU-задачи (вычисление суммы квадратов)."""
    return sum(i * i for i in range(n))

def run_sequential(numbers: List[int]) -> List[int]:
    """Последовательное выполнение (медленно)."""
    return [cpu_heavy_task(n) for n in numbers]

def run_parallel(numbers: List[int], max_workers: int = 4) -> List[int]:
    """Параллельное выполнение с ThreadPoolExecutor."""
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_n = {executor.submit(cpu_heavy_task, n): n for n in numbers}
        for future in as_completed(future_to_n):
            try:
                results.append(future.result())
            except Exception as e:
                print(f"Ошибка в задаче {future_to_n[future]}: {e}")
    return results

if __name__ == "__main__":
    try:
        numbers = [300_000, 400_000, 500_000, 600_000]
        
        print("⏱️  Последовательно:")
        start = time.time()
        seq_result = run_sequential(numbers)
        print(f"Время: {time.time() - start:.2f} сек | Результаты: {seq_result[:2]}...\n")
        
        print("⚡ Параллельно (4 потока):")
        start = time.time()
        par_result = run_parallel(numbers)
        print(f"Время: {time.time() - start:.2f} сек | Результаты: {par_result[:2]}...")
        
        print("\n✅ В параллельном режиме значительно быстрее!")
    except Exception as e:
        print(f"Ошибка: {e}")

from typing import Dict, Type

class PluginMeta(type):
    """Метакласс, который автоматически регистрирует все плагины."""
    registry: Dict[str, Type] = {}

    def __new__(mcs, name, bases, attrs):
        cls = super().__new__(mcs, name, bases, attrs)
        if name != "PluginBase":
            mcs.registry[name] = cls
            print(f"🔌 Зарегистрирован плагин: {name}")
        return cls


class PluginBase(metaclass=PluginMeta):
    """Базовый класс для всех плагинов."""
    def run(self):
        raise NotImplementedError("Плагин должен реализовать метод run()")


class DataProcessor(PluginBase):
    """Пример плагина."""
    def run(self):
        print("📊 Обработка данных...")


class ReportGenerator(PluginBase):
    """Ещё один плагин."""
    def run(self):
        print("📄 Генерация отчёта...")


if __name__ == "__main__":
    try:
        print(f"Всего зарегистрировано плагинов: {len(PluginMeta.registry)}")
        
        for name, plugin_cls in PluginMeta.registry.items():
            plugin = plugin_cls()
            plugin.run()
        
        print("\n✅ Все плагины успешно выполнены!")
    except Exception as e:
        print(f"Ошибка: {e}")

from contextlib import contextmanager
import time
import os
from pathlib import Path

@contextmanager
def timer(name: str):
    """Контекст-менеджер для замера времени выполнения блока кода."""
    start = time.time()
    try:
        yield
    finally:
        print(f"⏱️ {name} заняло {time.time() - start:.4f} сек")

@contextmanager
def chdir(path: str):
    """Временно меняет текущую рабочую директорию."""
    old_cwd = os.getcwd()
    try:
        os.chdir(path)
        yield
    finally:
        os.chdir(old_cwd)

if __name__ == "__main__":
    try:
        with timer("Тяжёлая операция"):
            time.sleep(1.2)
        
        with chdir("/tmp") as tmp:
            print(f"Временно в директории: {os.getcwd()}")
        
        print("✅ Контекст-менеджеры сработали!")
    except Exception as e:
        print(f"Ошибка: {e}")

import schedule
import time
from typing import Callable

def job(name: str):
    """Пример задачи, которая выполняется по расписанию."""
    print(f"✅ Задача '{name}' выполнена в {time.strftime('%H:%M:%S')}")

def main():
    """Настраиваем расписание."""
    schedule.every(2).seconds.do(job, name="Проверка почты")
    schedule.every(5).seconds.do(job, name="Обновление данных")
    schedule.every().day.at("10:00").do(job, name="Ежедневный отчёт")
    
    print("⏰ Расписание запущено. Ctrl+C для остановки.\n")
    
    try:
        while True:
            schedule.run_pending()
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n👋 Расписание остановлено.")

if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"Ошибка: {e}")

import random
from typing import List, Set

WORDS: List[str] = ["python", "программирование", "телеграм", "алгоритм", "разработчик", "функция", "декоратор"]

def print_board(guessed: List[str], attempts: int) -> None:
    """Выводит текущее состояние игры."""
    print(" ".join(guessed))
    print(f"Попыток осталось: {attempts}\n")

def check_winner(guessed: List[str]) -> bool:
    """Проверяет, угадано ли слово."""
    return "_" not in guessed

def play_hangman() -> None:
    """Основной игровой цикл."""
    word = random.choice(WORDS)
    guessed = ["_"] * len(word)
    attempts = 6
    guessed_letters: Set[str] = set()

    print("🎮 Добро пожаловать в Виселицу!\n")

    while attempts > 0 and not check_winner(guessed):
        print_board(guessed, attempts)
        letter = input("Введите букву: ").lower().strip()

        if len(letter) != 1 or not letter.isalpha():
            print("❌ Введите одну букву!\n")
            continue

        if letter in guessed_letters:
            print("⚠️ Вы уже вводили эту букву!\n")
            continue

        guessed_letters.add(letter)

        if letter in word:
            for i, char in enumerate(word):
                if char == letter:
                    guessed[i] = letter
            print("✅ Есть такая буква!\n")
        else:
            attempts -= 1
            print("❌ Такой буквы нет.\n")

    if check_winner(guessed):
        print(f"🎉 Поздравляем! Слово: {word.upper()}")
    else:
        print(f"💀 Вы проиграли. Слово было: {word.upper()}")

if __name__ == "__main__":
    try:
        play_hangman()
    except KeyboardInterrupt:
        print("\n👋 Игра прервана. До новых встреч!")
    except Exception as e:
        print(f"Неожиданная ошибка: {e}")

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List

def cpu_heavy_task(n: int) -> int:
    """Имитация тяжёлой CPU-задачи (например, вычисление суммы квадратов)."""
    return sum(i * i for i in range(n))

def run_sequential(numbers: List[int]) -> List[int]:
    """Последовательное выполнение."""
    return [cpu_heavy_task(n) for n in numbers]

def run_parallel(numbers: List[int], max_workers: int = 4) -> List[int]:
    """Параллельное выполнение с ThreadPoolExecutor."""
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_n = {executor.submit(cpu_heavy_task, n): n for n in numbers}
        for future in as_completed(future_to_n):
            try:
                results.append(future.result())
            except Exception as e:
                print(f"Ошибка в задаче: {e}")
    return results

if __name__ == "__main__":
    try:
        numbers = [300_000, 400_000, 500_000, 600_000]
        
        print("⏱️  Последовательно:")
        start = time.time()
        seq_result = run_sequential(numbers)
        print(f"Время: {time.time() - start:.2f} сек | Результаты: {seq_result[:2]}...\n")
        
        print("⚡ Параллельно (4 потока):")
        start = time.time()
        par_result = run_parallel(numbers)
        print(f"Время: {time.time() - start:.2f} сек | Результаты: {par_result[:2]}...")
        
        print("\n✅ В параллельном режиме значительно быстрее!")
    except Exception as e:
        print(f"Ошибка: {e}")

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List

def cpu_heavy_task(n: int) -> int:
    """Имитация тяжёлой CPU-задачи (например, вычисление суммы квадратов)."""
    return sum(i * i for i in range(n))

def run_sequential(numbers: List[int]) -> List[int]:
    """Последовательное выполнение — медленно."""
    return [cpu_heavy_task(n) for n in numbers]

def run_parallel(numbers: List[int], max_workers: int = 4) -> List[int]:
    """Параллельное выполнение с ThreadPoolExecutor."""
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # Отправляем все задачи сразу
        future_to_n = {executor.submit(cpu_heavy_task, n): n for n in numbers}
        
        # Получаем результаты по мере готовности
        for future in as_completed(future_to_n):
            try:
                results.append(future.result())
            except Exception as e:
                print(f"Ошибка в задаче: {e}")
    return results

if __name__ == "__main__":
    try:
        numbers = [500_000, 600_000, 700_000, 800_000]  # 4 тяжёлые задачи
        
        print("⏱️  Последовательно:")
        start = time.time()
        seq_result = run_sequential(numbers)
        print(f"Время: {time.time() - start:.2f} сек | Результаты: {seq_result[:2]}...\n")
        
        print("⚡ Параллельно (4 потока):")
        start = time.time()
        par_result = run_parallel(numbers)
        print(f"Время: {time.time() - start:.2f} сек | Результаты: {par_result[:2]}...")
        
        print("\n✅ В параллельном режиме почти в 4 раза быстрее!")
    except Exception as e:
        print(f"Ошибка: {e}")