Pythoner

import streamlit as st
from PIL import Image
import model  # ваша ML модель

st.title("🐕 Определитель пород собак")
uploaded_file = st.file_uploader("Загрузите фото собаки")

if uploaded_file:
    image = Image.open(uploaded_file)
    st.image(image, caption="Загруженное фото")
    prediction = model.predict(image)
    st.success(f"Это {prediction} с вероятностью 95%!")

import streamlit as st
import pandas as pd
import plotly.express as px

st.title("📊 Анализ продаж")
data = pd.read_csv("sales.csv")

# Интерактивный фильтр по датам
date_range = st.date_input("Выберите период")

# Динамический график
fig = px.line(data, x="date", y="sales")
st.plotly_chart(fig)

pip install streamlit
streamlit hello  # запустит демо-приложение

from pybrain.datasets import ClassificationDataSet
from pybrain.supervised.trainers import BackpropTrainer
from pybrain.tools.shortcuts import BuildNetwork

# Создание выборки данных
data = ClassificationDataSet(2, 1)

# Добавление данных (x1, x2, label)
data.addSample((0, 0), (0,))
data.addSample((0, 1), (1,))
data.addSample((1, 0), (1,))
data.addSample((1, 1), (0,))

# Создание нейронной сети
network = BuildNetwork(2, 3, 1)

# Обучение нейронной сети
trainer = BackpropTrainer(network, data)
for _ in range(1000):
    trainer.train()

# Проверка результатов
for sample in [(0, 0), (0, 1), (1, 0), (1, 1)]:
    output = network.activate(sample)
    print(f'Input: {sample}, Output: {output}')

age = 15
type = "teenager" if age < 18 else "adult"
print(type)  # Выведет: teenager

from peewee import *

# Создаем подключение к базе данных SQLite (или подключаемся к существующей)
db = SqliteDatabase('my_database.db')

# Определяем модель данных (таблицу)
class User(Model):
    username = CharField(unique=True)
    password = CharField()
    email = CharField()
    active = BooleanField(default=True)
    joined_date = DateTimeField()

    class Meta:
        database = db  # Связываем модель с базой данных


# Создаем таблицу в базе данных (если она не существует)
db.connect()
db.create_tables([User])

# Создаем нового пользователя
user = User(username='testuser', password='password', email='[email protected]', joined_date='2024-07-26 10:00:00')
user.save() # Сохраняем пользователя в базу данных


# Получаем пользователя по имени пользователя
retrieved_user = User.get(User.username == 'testuser')
print(retrieved_user.email) # Вывод: [email protected]

#  Обновляем данные пользователя
retrieved_user.active = False
retrieved_user.save()

#  Удаляем пользователя
retrieved_user.delete_instance()

#  Запрос с фильтрацией
active_users = User.select().where(User.active == True)
for user in active_users:
    print(user.username)



db.close()

from boltons.iterutils import chunked

data = list(range(20))
chunks = list(chunked(data, 4)) #  Разделение данных на чанки размером 4
print(chunks)
# Вывод: [[0, 1, 2, 3], [4, 5, 6, 7], [8, 9, 10, 11], [12, 13, 14, 15], [16, 17, 18, 19]]


from boltons.dictutils import OMD
# Ordered Multi Dict, сохраняет порядок добавления ключей и позволяет иметь несколько значений для одного ключа

omd = OMD()
omd[1] = "a"
omd[2] = "b"
omd[1] = "c"

print(omd)
# Вывод: OMD([(1, 'a'), (2, 'b'), (1, 'c')])
print(omd.getlist(1)) # Получение всех значений для ключа 1
# Вывод: ['a', 'c']



from boltons.funcutils import wraps

def my_decorator(f):
    @wraps(f) #  Сохраняет метаданные оригинальной функции
    def wrapper(*args, **kwargs):
        print("Before function call")
        result = f(*args, **kwargs)
        print("After function call")
        return result
    return wrapper

@my_decorator
def my_function(a, b):
    """
    This is my function.
    """
    return a + b

print(my_function(1, 2))
# Вывод:
# Before function call
# After function call
# 3
print(my_function.__doc__) #  Метаданные сохраняются благодаря wraps
# Вывод: This is my function.

# !pip install modin[ray] # или modin[dask], если предпочитаете Dask
import time
import modin.pandas as pd #  Замена import pandas as pd

start_time = time.time()

#  Загрузка большого датасета (замените на свой путь к файлу)
df = pd.read_csv("large_dataset.csv")

#  Выполнение некоторых операций с DataFrame
# ... (например, фильтрация, агрегация и т.д.) ...
df = df[df['column_name'] > 100]
df = df.groupby(['column_name1','column_name2'])['column_name3'].sum()


end_time = time.time()

print(f"Время выполнения с Modin: {end_time - start_time} секунд")



# Для сравнения, выполните тот же код с обычным Pandas:
import time
import pandas as pd

start_time = time.time()

#  Загрузка большого датасета (замените на свой путь к файлу)
df = pd.read_csv("large_dataset.csv")

#  Выполнение некоторых операций с DataFrame
# ... (например, фильтрация, агрегация и т.д.) ...
df = df[df['column_name'] > 100]
df = df.groupby(['column_name1','column_name2'])['column_name3'].sum()


end_time = time.time()

print(f"Время выполнения с Pandas: {end_time - start_time} секунд")

public class HelloWorld {
    public static String greeting() {
        return "Hello from Java!";
    }
}

import jpype

# Запуск виртуальной машины Java
jpype.startJVM(jpype.getDefaultJVMPath())

# Загрузка класса
HelloWorld = jpype.JClass("HelloWorld")

# Вызов метода
result = HelloWorld.greeting()
print(result)  # Вывод: Hello from Java!

# Остановка виртуальной машины Java
jpype.shutdownJVM()

def calculate_area(length, width):
    return length * width

# Использование функции
room_area = calculate_area(5, 4)
print(f"Площадь комнаты: {room_area} кв.м.")

class BankAccount:
    def __init__(self, balance):
        self.balance = balance
    
    def deposit(self, amount):
        self.balance += amount
        return f"Новый баланс: {self.balance}"

# Использование метода
account = BankAccount(1000)
account.deposit(500)  # Вызов метода через объект

from datasets import load_dataset

# Загрузка датасета "imdb" с Hugging Face Hub
dataset = load_dataset("imdb")

print(dataset)
# Вывод: DatasetDict({
#     train: Dataset({
#         features: ['text', 'label'],
#         num_rows: 25000
#     })
#     test: Dataset({
#         features: ['text', 'label'],
#         num_rows: 25000
#     })
#     unsupervised: Dataset({
#         features: ['text', 'label'],
#         num_rows: 50000
#     })
# })

train_data = dataset['train']

print(train_data[0]) #  Доступ к первому примеру в обучающем наборе
# Вывод: {'text': 'A series of escapades demonstrating the adage that what is good for the goose is also good for the gander, some of which occasionally amuses but none of which amounts to much of a story.', 'label': 0}



small_train_dataset = dataset["train"].shuffle(seed=42).select(range(1000)) # Перемешивание и выборка


from datasets import ClassLabel
import random
import pandas as pd

def show_random_elements(dataset, num_examples=10):
    picks = []
    n = len(dataset)
    for _ in range(num_examples):
        pick = random.randint(0, n - 1)
        picks.append(dataset[pick])

    df = pd.DataFrame(picks)
    if isinstance(dataset.features['label'], ClassLabel):
        df['label'] = df['label'].apply(lambda x: dataset.features['label'].int2str(x))

    print(df)

show_random_elements(small_train_dataset)

#  Вывод: таблица с 10 случайными примерами

from transformers import pipeline

classifier = pipeline("sentiment-analysis")  #  Загрузка предобученной модели для анализа тональности текста

results = classifier([
    "I love this library!",
    "This is a terrible movie.",
    "This is a neutral statement."
])

for result in results:
    print(result)

#  Примерный вывод:
# [{'label': 'POSITIVE', 'score': 0.9998950958251953}]
# [{'label': 'NEGATIVE', 'score': 0.9991175532341003}]
# [{'label': 'NEGATIVE', 'score': 0.9865201115608215}] #  Модель может ошибаться, особенно на нейтральных высказываниях



from transformers import pipeline

generator = pipeline('text-generation', model='gpt2') # Загрузка модели для генерации текста

text = generator("Once upon a time, there was a large language model.", max_length=50, num_return_sequences=2)

for generated_text in text:
    print(generated_text['generated_text'])

# Примерный вывод (будет отличаться при каждом запуске):
# Once upon a time, there was a large language model. It was trained on a massive dataset of text and code, and it could generate text, translate languages, write different kinds of creative content, and answer your questions in an informative way.
# Once upon a time, there was a large language model. And he lived in a little house made of straw. One day, he was sitting in his house, reading a book


from transformers import pipeline

summarizer = pipeline("summarization")

text = """
The quick brown fox jumps over the lazy dog. This is a test sentence.  
It is used to demonstrate text summarization.  The fox is brown and quick.
The dog is lazy.  Summarization is a useful NLP task.
"""

summary = summarizer(text, max_length=30, min_length=10, do_sample=False)

print(summary[0]['summary_text'])
# Примерный вывод:
# The quick brown fox jumps over the lazy dog. It is used to demonstrate text summarization. The fox is brown and quick. The dog is lazy.

import plotly.express as px
import pandas as pd

# Создаём тестовые данные
df = pd.DataFrame({
    'Месяц': ['Янв', 'Фев', 'Март', 'Апр', 'Май'],
    'Продажи': [100, 150, 200, 180, 250],
    'Прибыль': [20, 30, 40, 35, 50]
})

# Создаём интерактивный график
fig = px.line(df, x='Месяц', y=['Продажи', 'Прибыль'],
              title='Динамика продаж и прибыли',
              template='plotly_dark')

# Добавляем hover-эффекты
fig.update_traces(mode='lines+markers', hovertemplate='%{y:,.0f}₽')

# Сохраняем как HTML или показываем в браузере
fig.write_html('sales_dashboard.html')

from bokeh.plotting import figure, show
from bokeh.layouts import column
from bokeh.models import ColumnDataSource, HoverTool
import numpy as np

# Создаём интерактивный scatter plot
x = np.random.normal(0, 1, 1000)
y = np.random.normal(0, 1, 1000)

source = ColumnDataSource(data=dict(
    x=x,
    y=y,
    size=np.random.uniform(5, 15, 1000),
    color=['#%06x' % np.random.randint(0, 0xFFFFFF) for _ in range(1000)]
))

p = figure(width=800, height=600, title='Интерактивный Scatter Plot')
p.scatter('x', 'y', size='size', color='color', alpha=0.6, source=source)

# Добавляем интерактивные подсказки
hover = HoverTool(tooltips=[
    ('X', '@x{0.000}'),
    ('Y', '@y{0.000}'),
    ('Размер', '@size{0.00}')
])
p.add_tools(hover)

show(p)

import plotly.graph_objects as go

fig = go.Figure(
    data=[go.Scatter(x=[1, 2, 3], y=[1, 3, 2])],
    layout=dict(
        updatemenus=[dict(
            type="buttons",
            buttons=[dict(label="Play",
                         method="animate",
                         args=[None])]
        )]
    )
)

from bokeh.plotting import curdoc
from functools import partial
from tornado.ioloop import IOLoop

def update():
    source.data['y'] = np.random.rand(100)
    
curdoc().add_periodic_callback(update, 100)  # Обновление каждые 100мс

from dataclasses import dataclass
from typing import List

@dataclass
class DataPoint:
    x: float
    y: float
    category: str

data_points: List[DataPoint] = []  # Эффективнее, чем DataFrame для больших данных

import torch
import torch.nn as nn

class SimpleNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.layer1 = nn.Linear(784, 128)
        self.relu = nn.ReLU()
        self.layer2 = nn.Linear(128, 10)
    
    def forward(self, x):
        x = self.layer1(x)
        x = self.relu(x)
        return self.layer2(x)

Создаем модель одной строчкой! 🎯
model = SimpleNet().to('cuda' if torch.cuda.is_available() else 'cpu')

import tensorflow as tf

model = tf.keras.Sequential([
    tf.keras.layers.Dense(128, activation='relu', input_shape=(784,)),
    tf.keras.layers.Dense(10)
])

# Компилируем модель
model.compile(optimizer='adam',
             loss='sparse_categorical_crossentropy',
             metrics=['accuracy'])

# PyTorch стиль
optimizer = torch.optim.Adam(model.parameters())
criterion = nn.CrossEntropyLoss()

for epoch in range(10):
    for batch_x, batch_y in dataloader:
        optimizer.zero_grad()
        outputs = model(batch_x)
        loss = criterion(outputs, batch_y)
        loss.backward()
        optimizer.step()

# Разбиваем данные
from sklearn.model_selection import train_test_split

X_train, X_val, y_train, y_val = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# Следим за метриками
val_loss = []
for epoch in range(epochs):
    model.train()
    # ... обучение ...
    
    model.eval()
    with torch.no_grad():
        val_predictions = model(X_val)
        v_loss = criterion(val_predictions, y_val)
        val_loss.append(v_loss.item())

from pygments import highlight
from pygments.lexers import PythonLexer
from pygments.formatters import HtmlFormatter

code = 'print("Hello, world!")'
highlighted_code = highlight(code, PythonLexer(), HtmlFormatter())
print(highlighted_code)

# <div class="highlight"><pre><span></span><span class="nb">print</span><span class="p">(</span><span class="s2">&quot;Hello, world!&quot;</span><span class="p">)</span>
# </pre></div>

import typer

app = typer.Typer()

@app.command()
def hello(name: str):
    """
    Простое приветствие.
    """
    print(f"Привет, {name}!")

@app.command()
def goodbye(name: str, formal: bool = False):
    """
    Прощание.  Можно сделать формальным.
    """
    if formal:
        print(f"До свидания, {name}!")
    else:
        print(f"Пока, {name}!")

if __name__ == "__main__":
    app()

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

# Инициализация драйвера Chrome (убедитесь, что ChromeDriver установлен и находится в PATH)
driver = webdriver.Chrome()

try:
    # Открытие страницы
    driver.get("https://www.example.com")

    # Явное ожидание элемента с ID "my-element" (до 10 секунд)
    element = WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.ID, "my-element"))
    )

    # Вывод текста элемента
    print(element.text)

finally:
    # Закрытие браузера
    driver.quit()

import asyncio
import aioredis

async def redis_example():
    try:
        # Подключение к Redis
        redis = await aioredis.create_redis_pool(
            'redis://localhost',
            db=0,
            encoding='utf-8',  # Декодирование в UTF-8
            minsize=5,  # Минимальное количество соединений в пуле
            maxsize=10   # Максимальное количество соединений в пуле
        )


        # Работа со строками
        await redis.set('name', 'John Doe')
        name = await redis.get('name')
        print(f"Name: {name}") # Вывод: Name: John Doe



        # Работа со списками
        await redis.rpush('my_list', 'one', 'two', 'three')
        list_length = await redis.llen('my_list')
        list_items = await redis.lrange('my_list', 0, -1)
        print(f"List length: {list_length}") # Вывод: List length: 3
        print(f"List items: {list_items}") # Вывод: List items: ['one', 'two', 'three']


        # Работа с хэшами
        await redis.hset('user:1', 'name', 'Alice')
        await redis.hset('user:1', 'age', 30)
        user_data = await redis.hgetall('user:1')
        print(f"User data: {user_data}") # Вывод: User data: {'name': 'Alice', 'age': '30'}


        # Работа с множествами
        await redis.sadd('my_set', 'apple', 'banana', 'cherry')
        set_members = await redis.smembers('my_set')
        print(f"Set members: {set_members}") # Вывод: Set members: {'apple', 'banana', 'cherry'}



        #  Использование pipeline
        async with redis.pipeline(transaction=True) as pipe:  # transaction=True для атомарности операций
            await pipe.set('key1', 'value1')
            await pipe.get('key1')
            await pipe.hset('user:2', 'name', 'Bob')
            results = await pipe.execute()
        print(f"Pipeline results: {results}")



        redis.close()
        await redis.wait_closed()


    except aioredis.RedisError as e:
        print(f"Ошибка Redis: {e}")


if __name__ == "__main__":
    asyncio.run(redis_example())

from tqdm import tqdm
import time

# Простой пример с range
for i in tqdm(range(100)):
    time.sleep(0.02)  # Имитация длительной операции


# С ручным обновлением
with tqdm(total=100) as pbar:
    for i in range(10):
        time.sleep(0.1)
        pbar.update(10) # Обновление индикатора на 10 единиц


# С описанием
for i in tqdm(range(100), desc="Обработка данных"):
    time.sleep(0.02)

# Вложенные циклы
from tqdm import trange
for i in trange(4, desc="1st loop"):
    for j in trange(5, desc="2nd loop", leave=False):
        for k in trange(50, desc="3rd loop", leave=False):
            time.sleep(0.001)

# С pandas
import pandas as pd
df = pd.DataFrame({'a': range(10000)})

# Применение tqdm к pandas DataFrame (apply с progress_apply)
tqdm.pandas() #  Необходимо для progress_apply
df['b'] = df['a'].progress_apply(lambda x: x*x) #  Индикатор выполнения для apply



# Пример с файлами
import os
with open("large_file.txt", "w") as f:
    for i in trange(10000, desc="Создание файла"):
        f.write("This is a line of text.\n")


with open("large_file.txt", "r") as f:
    for line in tqdm(f, total=os.path.getsize("large_file.txt"), unit="B", unit_scale=True, unit_divisor=1024, desc="Чтение файла"):
        # Обработка каждой строки файла
        pass

import asyncio
import httpx

async def fetch_page(client: httpx.AsyncClient, url: str):
    try:
        response = await client.get(url, timeout=10.0)
        response.raise_for_status()
        return {"url": url, "status_code": response.status_code, "content": response.text}
    except httpx.HTTPError as exc:
        return {"url": url, "error": f"HTTP Error: {exc}"}
    except httpx.TimeoutException:
        return {"url": url, "error": "Request timed out"}
    except Exception as e:
        return {"url": url, "error": f"An unexpected error occurred: {e}"}


async def main():
    urls = [
        "https://www.example.com",
        "https://www.google.com",
        "https://www.python.org",
        "https://this-is-an-invalid-url.com",  # Пример некорректного URL
        "https://jsonplaceholder.typicode.com/todos/1"  # Пример API
    ]

    async with httpx.AsyncClient(http2=True) as client:  # Использование HTTP/2
        tasks = [fetch_page(client, url) for url in urls]
        results = await asyncio.gather(*tasks)

    for result in results:
        print(result)




async def make_requests_with_limits(urls, max_concurrent=5):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_with_semaphore(url):
        async with semaphore:
            return await fetch_page(httpx.AsyncClient(), url)


    async with httpx.AsyncClient() as client:
        tasks = [fetch_with_semaphore(url) for url in urls]
        results = await asyncio.gather(*tasks)

    for result in results:
        print(result)




if __name__ == "__main__":
    asyncio.run(main())

    # Пример с ограничением числа одновременных запросов
    asyncio.run(make_requests_with_limits(["https://www.example.com"] * 10))

from pathlib import Path

# Создание объекта Path
file_path = Path("my_file.txt")
directory_path = Path("my_directory")

# Проверка существования файла/директории
if file_path.exists():
    print(f"Файл {file_path} существует")
else:
    print(f"Файл {file_path} не существует")

if directory_path.exists():
    print(f"Директория {directory_path} существует")

if directory_path.is_dir():
    print(f"Это директория")

# Создание директории
directory_path.mkdir(parents=True, exist_ok=True) # parents=True создает все родительские директории при необходимости, exist_ok=True предотвращает ошибку, если директория уже существует

# Создание файла и запись в него
file_path.write_text("Hello, pathlib!")

# Чтение из файла
file_content = file_path.read_text()
print(f"Содержимое файла: {file_content}")

# Переименование файла
new_file_path = Path("my_new_file.txt")
file_path.rename(new_file_path)

# Удаление файла
new_file_path.unlink()

# Перебор файлов в директории
for file in directory_path.iterdir():
    print(file)

# Получение абсолютного пути
absolute_path = directory_path.resolve()
print(f"Абсолютный путь: {absolute_path}")


# Создание вложенных директорий и файла
nested_dir = Path("my_directory/nested/deeper")
nested_dir.mkdir(parents=True, exist_ok=True)
nested_file = nested_dir / "nested_file.txt" #  / используется для объединения путей
nested_file.write_text("Content in nested file")


import shutil

# Удаление директории и ее содержимого
shutil.rmtree(directory_path) #  rmtree из shutil используется для удаления непустых директорий

# Создать новое окружение с Python 3.9
conda create -n my_env python=3.9

# Активировать окружение
conda activate my_env

# Установить пакеты в активное окружение
conda install numpy pandas scipy

# Просмотреть список установленных пакетов
conda list

# Деактивировать окружение
conda deactivate

# Удалить окружение
conda remove -n my_env --all

# Создать окружение из файла environment.yml
conda env create -f environment.yml

# Экспортировать окружение в файл environment.yml
conda env export > environment.yml

# Обновить все пакеты в активном окружении
conda update --all

# Поиск пакета
conda search numpy

def test_sum():
    assert sum([1, 2, 3]) == 6
    assert sum([-1, 1]) == 0

from hypothesis import given
import hypothesis.strategies as st

@given(st.lists(st.integers()))
def test_sum_properties(numbers):
    assert sum(numbers) == sum(reversed(numbers))
    assert sum(numbers + [0]) == sum(numbers)

import time

def compute_intensive_task(n):
    result = 0
    for i in range(n):
        result += i * i
    return result

start_time = time.time()
result = compute_intensive_task(10000000)  #  CPU-bound задача
end_time = time.time()

print(f"Результат: {result}")
print(f"Время выполнения: {end_time - start_time} секунд")


#  Для запуска с pypy3: pypy3 example.py

import numpy as np

# Создание массивов:
a = np.array([1, 2, 3])  # Одномерный массив
b = np.array([[1, 2, 3], [4, 5, 6]])  # Двумерный массив
c = np.zeros((2, 3))  # Массив из нулей
d = np.ones((3, 2))  # Массив из единиц
e = np.arange(10)  # Массив с числами от 0 до 9
f = np.linspace(0, 1, 5)  # Массив из 5 чисел, равномерно распределенных от 0 до 1
g = np.random.rand(2, 2)  # Массив случайных чисел от 0 до 1

print("a:", a)
print("b:", b)
print("c:", c)
print("d:", d)
print("e:", e)
print("f:", f)
print("g:", g)



# Математические операции:
print("a + 2:", a + 2)
print("a * 3:", a * 3)
print("a.sum():", a.sum())  # Сумма элементов массива
print("b.mean():", b.mean()) # Среднее значение элементов массива

print("a.dot(a): ", a.dot(a)) # Скалярное произведение векторов



# Индексация и срезы:
print("a[0]:", a[0])
print("b[1, 2]:", b[1, 2])
print("e[1:4]:", e[1:4])



# Изменение формы массива:
print("b.reshape(3, 2):", b.reshape(3, 2))
print("a.T:", a.T) # Транспонирование массива


# Линейная алгебра:
A = np.array([[1, 2], [3, 4]])
B = np.array([[5, 6], [7, 8]])
print("A.dot(B):", A.dot(B)) # Умножение матриц
print("np.linalg.inv(A):", np.linalg.inv(A)) # Обратная матрица

import shutil
import os

# Копирование файла:
shutil.copy("source_file.txt", "destination_folder/") # Копирование в директорию
shutil.copy("source_file.txt", "destination_file.txt") # Копирование с новым именем

# Копирование директории (рекурсивно):
shutil.copytree("source_folder", "destination_folder")

# Перемещение файла или директории:
shutil.move("source_file.txt", "destination_folder/")
shutil.move("source_folder", "destination_folder")

# Удаление файла:
os.remove("file_to_delete.txt")

# Удаление директории (рекурсивно):
shutil.rmtree("directory_to_delete")


# Создание архива:

shutil.make_archive("archive_name", "zip", "folder_to_archive") # ZIP архив
shutil.make_archive("archive_name", "gztar", "folder_to_archive") # tar.gz архив
shutil.make_archive("archive_name", "bztar", "folder_to_archive") # tar.bz2 архив
shutil.make_archive("archive_name", "xztar", "folder_to_archive") # tar.xz архив



# Работа с файловыми путями (некоторые примеры):

print(shutil.which("python")) #  Путь к исполняемому файлу python


#  Проверка дискового пространства

total, used, free = shutil.disk_usage("/")  #  Получение информации о дисковом пространстве (корневой раздел)
print(f"Total: {total // (2**30)} GiB")
print(f"Used: {used // (2**30)} GiB")
print(f"Free: {free // (2**30)} GiB")

import click

@click.group()
def cli():
    """Утилита для работы с файлами"""
    pass

@cli.command()
@click.argument('path')
@click.option('--lines', '-l', is_flag=True, help='Показать количество строк')
def analyze(path, lines):
    """Анализирует файл и выводит статистику"""
    try:
        with open(path) as f:
            content = f.readlines()
            if lines:
                click.echo(f'Количество строк: {len(content)}')
    except FileNotFoundError:
        click.secho('Файл не найден! 😱', fg='red')

if __name__ == '__main__':
    cli()

@cli.command()
@click.argument('path')
def process(path):
    """Обрабатывает файлы с прогресс-баром"""
    files = os.listdir(path)
    with click.progressbar(files, label='Обработка файлов') as bar:
        for f in bar:
            # что-то делаем с файлом
            time.sleep(0.1)

click.secho('Внимание! 🚨', fg='yellow', bold=True)
click.secho('Ошибка! ❌', fg='red')
click.secho('Успех! ✅', fg='green')

if click.confirm('Уверены, что хотите удалить все файлы? 🤔'):
    click.echo('Поехали! 🚀')

import concurrent.futures
import time
import requests

URLS = [
    "https://www.example.com",
    "https://www.google.com",
    "https://www.python.org",
    "https://www.wikipedia.org",
]

def load_url(url, timeout):
    try:
        response = requests.get(url, timeout=timeout)
        return f"{url}: {len(response.content)} bytes"
    except requests.exceptions.RequestException as e:
        return f"{url}: Error: {e}"


start_time = time.time()

# Последовательное выполнение:
for url in URLS:
    print(load_url(url, timeout=60))


end_time = time.time()
print(f"Последовательное выполнение заняло: {end_time - start_time} секунд")





start_time = time.time()

# Асинхронное выполнение с ThreadPoolExecutor:
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    # map() возвращает результаты в порядке URLS
    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            print(f"{url}: generated an exception: {exc}")
        else:
            print(data)

end_time = time.time()
print(f"Асинхронное выполнение заняло: {end_time - start_time} секунд")

import difflib

text1 = """
This is the first line.
This is the second line.
This is the third line.
"""

text2 = """
This is the first line.
This is the modified second line.
This is the fourth line.
"""

diff = difflib.ndiff(text1.splitlines(), text2.splitlines())


print("\n".join(diff))

# Вывод:
#   This is the first line.
# - This is the second line.
# + This is the modified second line.
# ?            +++++++++
# 
# - This is the third line.
# ?               ---
# 
# + This is the fourth line.
# ?             ++++
#           ^



# Более наглядное сравнение:

diff = difflib.unified_diff(text1.splitlines(), text2.splitlines(), fromfile='file1.txt', tofile='file2.txt')
print("\n".join(diff))



# Вывод:
# --- file1.txt
# +++ file2.txt
# @@ -1,4 +1,4 @@
# 
#  This is the first line.
# -This is the second line.
# +This is the modified second line.
# -This is the third line.
# +This is the fourth line.



#  Сравнение HTML:
from difflib import HtmlDiff

html_diff = HtmlDiff()
table = html_diff.make_table(text1.splitlines(), text2.splitlines(), fromdesc='file1.txt', todesc='file2.txt')

with open('diff.html', 'w') as f:
    f.write(table) #  Сохранение результата в HTML файл

from fastapi import FastAPI, Query, Path, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI(title="My Awesome API", version="1.0.0", description="Example FastAPI application")


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None


@app.get("/")  # GET-запрос на корневой путь
async def root():
    return {"message": "Hello World"}


@app.get("/items/{item_id}")  # GET-запрос с параметром пути
async def read_item(item_id: int = Path(..., title="The ID of the item to get", ge=1)): # Валидация параметра пути
    return {"item_id": item_id}


@app.get("/items/")
async def read_items(q: Optional[str] = Query(None, min_length=3, max_length=50)): # Валидация query-параметра
    results = {"items": [{"item_id": "foo"}, {"item_id": "bar"}]}
    if q:
        results.update({"q": q})
    return results



@app.post("/items/", response_model=Item)  # POST-запрос с валидацией тела запроса
async def create_item(item: Item):
    return item


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, **item.dict()}


@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Пример обработки ошибок
    if user_id == 404:
        raise HTTPException(status_code=404, detail="User not found")
    return {"user_id": user_id}

# Запуск приложения
uvicorn main:app --reload

from dataclasses import dataclass, field

@dataclass(order=True)  # order=True добавляет методы сравнения
class Person:
    name: str
    age: int
    city: str = "Unknown"  # Значение по умолчанию
    email: str = field(default="[email protected]", repr=False) # repr=False скрывает поле в __repr__
    scores: list[int] = field(default_factory=list)  # mutable default value


person1 = Person("Alice", 30, "New York", scores=[100, 90])
person2 = Person("Bob", 25, "Los Angeles")
person3 = Person("Charlie", 35, scores=[80, 85])
person4 = Person("Alice", 30, "New York", scores=[100, 90])


print(person1)  # Вывод: Person(name='Alice', age=30, city='New York', scores=[100, 90])
print(person2)  # Вывод: Person(name='Bob', age=25, city='Los Angeles', scores=[])
print(person3) # Вывод: Person(name='Charlie', age=35, city='Unknown', scores=[80, 85])

print(person1 == person4) # True
print(person1 > person2) # True  (сравнение по name, затем age)


# Изменение атрибутов:
person1.city = "San Francisco"
person1.scores.append(95)

print(person1)  # Вывод: Person(name='Alice', age=30, city='San Francisco', scores=[100, 90, 95])

# Post-init processing
@dataclass
class InventoryItem:
    """Class for keeping track of an item in inventory."""
    name: str
    unit_price: float
    quantity_on_hand: int = 0

    def __post_init__(self):
        self.total_cost = self.unit_price * self.quantity_on_hand



item = InventoryItem("Pen", 1.5, 50)
print(item.total_cost) # 75.0

import ctypes
import time

# Загрузка библиотеки C (libcallback_example.so - пример,  нужно скомпилировать код C)
lib = ctypes.CDLL("./libcallback_example.so")

# Определение типа функции обратного вызова
CMPFUNC = ctypes.CFUNCTYPE(None, ctypes.c_int)


# Функция Python, которая будет использоваться в качестве обратного вызова
def my_callback(value):
    print(f"Python callback called with value: {value}")



# Регистрация функции обратного вызова в библиотеке C
#  Предполагается, что в библиотеке C есть функция register_callback, принимающая указатель на функцию
register_callback = lib.register_callback
register_callback.argtypes = [CMPFUNC]
register_callback.restype = None
register_callback(CMPFUNC(my_callback))


# Вызов функции C, которая будет периодически вызывать функцию обратного вызова
# Предполагается, что в библиотеке C есть функция run_loop, которая запускает цикл
run_loop = lib.run_loop
run_loop.argtypes = []
run_loop.restype = None
run_loop()

#include <stdio.h>
#include <unistd.h> // for sleep


typedef void (*callback_function)(int);

static callback_function registered_callback = NULL;

void register_callback(callback_function callback) {
    registered_callback = callback;
}

void run_loop() {
    int i = 0;
    while (i < 5) {
        sleep(1);
        if (registered_callback != NULL) {
            registered_callback(i);
        }
        i++;
    }
}

gcc -shared -o libcallback_example.so -fPIC callback_example.c

from bs4 import BeautifulSoup

html_content = """
<!DOCTYPE html>
<html lang="en">
<head>
    <title>Example Page</title>
    <meta name="description" content="Example description">
</head>
<body>
    <h1>This is a heading</h1>
    <p>This is some text in a paragraph.</p>
    <a href="https://www.example.com">This is a link</a>
    <ul>
        <li>Item 1</li>
        <li>Item 2</li>
    </ul>
    <div class="my-class">This is a div</div>

    <p id="my-id">This is a paragraph with id </p>
</body>
</html>
"""

soup = BeautifulSoup(html_content, "html.parser")  # "html.parser" - используемый парсер


# Навигация и поиск элементов:
title = soup.title.string  # Получение текста заголовка
print("Title:", title)

description = soup.find("meta", attrs={"name": "description"})["content"]
print("Description:", description)


links = soup.find_all("a")
for link in links:
    print("Link:", link["href"])

paragraphs = soup.find_all("p")
for p in paragraphs:
    print("Paragraph:", p.text)

list_items = soup.find("ul").find_all("li")
for item in list_items:
    print("List item:", item.text)

div_with_class = soup.find("div", class_="my-class")
print("Div with class:", div_with_class.text)

paragraph_with_id = soup.find("p", id="my-id")
print("Paragraph with id:", paragraph_with_id.text)



# Модификация HTML:

new_paragraph = soup.new_tag("p")
new_paragraph.string = "This is a new paragraph."
soup.body.append(new_paragraph)

print(soup.prettify()) #  Вывод измененного HTML с отступами

print(bool(-1))  # Результат: True

os.rename(old_name, new_name)

import os

# Переименуем файл 'старый_файл.txt' в 'новый_файл.txt'
old_name = 'старый_файл.txt'
new_name = 'новый_файл.txt'

# Проверяем, существует ли старый файл
if os.path.exists(old_name):
    os.rename(old_name, new_name)
    print(f'Файл переименован в {new_name}')
else:
    print(f'Файл {old_name} не найден')

import humanize
import datetime

# Пример: время, прошедшее с 24 сентября 2023 года
dt = datetime.datetime(2023, 9, 24)
print(humanize.naturaltime(dt))  # Вывод: "1 месяц назад" (на момент текущей даты)

import humanize

number = 1500
print(humanize.intcomma(number))  # Вывод: "1,500"
print(humanize.intword(number))    # Вывод: "1.5 k"

import humanize

duration = 3661  # время в секундах
print(humanize.precisedelta(datetime.timedelta(seconds=duration))) 
# Вывод: "1 час, 1 минутa, 1 секунда"

a = 3
b = 10

print('a больше b') if a > b else print('a меньше b')

result = 'Четное' if a % 2 == 0 else 'Нечетное'
print(result)

# Результат:
# a меньше b
# Нечетное 

# Старый подход с os.path
import os.path
file_path = os.path.join('data', 'users', 'config.json')
parent_dir = os.path.dirname(file_path)
file_name = os.path.basename(file_path)

# Новый подход с pathlib
from pathlib import Path
file_path = Path('data') / 'users' / 'config.json'
parent_dir = file_path.parent
file_name = file_path.name

path = Path('config.json')
if path.exists():
    print('Файл существует!')

Path('nested/directories/structure').mkdir(parents=True, exist_ok=True)

# Найти все .py файлы в текущей директории
python_files = list(Path('.').glob('*.py'))

path = Path('document.pdf')
print(path.suffix)  # .pdf
print(path.stem)    # document

config_path = (Path.home() / 'projects' / 'app' / 'config.json')
if config_path.exists():
    data = json.loads(config_path.read_text())

with Path('log.txt').open('w') as f:
    f.write('Logging started')

def create_power_function(power):
    def power_func(x):
        return x ** power
    return power_func

# Создаём функции на лету
square = create_power_function(2)
cube = create_power_function(3)

import types

def create_dynamic_function(func_name, args, body):
    namespace = {}
    func_code = f"def {func_name}({', '.join(args)}):\n{body}"
    exec(func_code, globals(), namespace)
    return namespace[func_name]

# Магия в действии
greet = create_dynamic_function(
    "greet",
    ["name"],
    "    return f'Привет, {name}!'"
)

def main():

# Точка входа в программу 
if __name__ == '__main__':
    main()

# Без slots
class User:
    def __init__(self, name, age):
        self.name = name
        self.age = age

# Со slots
class OptimizedUser:
    __slots__ = ['name', 'age']
    
    def __init__(self, name, age):
        self.name = name
        self.age = age

class DataPoint:
    __slots__ = ['timestamp', 'value', 'sensor_id']
    
    def __init__(self, timestamp, value, sensor_id):
        self.timestamp = timestamp
        self.value = value
        self.sensor_id = sensor_id

# Представьте, что таких объектов у вас миллионы
data_points = [DataPoint(time.time(), random.random(), i) for i in range(1_000_000)]

def divide(a, b):
    if b == 0:
        raise ValueError("Деление на ноль невозможно.")
    return a / b

try:
    result = divide(10, 0)
except ValueError as e:
    print(f"Ошибка: {e}")

import whylogs as why
import pandas as pd

# Создание примера данных
data = {
    "age": [25, 30, 35, 40, 45],
    "income": [50000, 60000, 70000, 80000, 90000]
}
df = pd.DataFrame(data)

# Инициализация логгера
logger = why.logger()

# Логируем данные
logger.log_dataframe(df)

# Генерируем отчет
report = logger.report()

# Сохраняем отчет
report.save("report.whylog")

# Выводим на экран
print(report)

class Bird:
    def fly(self):
        return "I can fly!"

class Duck(Bird):
    def quack(self):
        return "Quack!"

class Airplane:
    def fly(self):
        return "I can also fly!"

def make_it_fly(flyable_thing):
    print(flyable_thing.fly())

# Создаем объекты
duck = Duck()
airplane = Airplane()

# Используем их, не смотря на разные типы
make_it_fly(duck)      # Выведет: I can fly!
make_it_fly(airplane)  # Выведет: I can also fly!

import ast

code = """
def calculate_sum(a, b):
    result = a + b
    print(f"Sum: {result}")
    return result
"""

# Создаем AST
tree = ast.parse(code)

# Анализируем структуру
for node in ast.walk(tree):
    if isinstance(node, ast.FunctionDef):
        print(f"Найдена функция: {node.name}")
    elif isinstance(node, ast.Name):
        print(f"Найдена переменная: {node.id}")

class DebugTransformer(ast.NodeTransformer):
    def visit_FunctionDef(self, node):
        # Добавляем отладочный print в начало каждой функции
        debug_print = ast.Expr(
            value=ast.Call(
                func=ast.Name(id='print', ctx=ast.Load()),
                args=[ast.Constant(value=f"Вызов функции {node.name}")],
                keywords=[]
            )
        )
        node.body.insert(0, debug_print)
        return node

# Применяем трансформацию
transformed = DebugTransformer().visit(tree)

class StringOptimizer(ast.NodeTransformer):
    def visit_BinOp(self, node):
        # Оптимизация конкатенации строк
        if isinstance(node.op, ast.Add):
            if isinstance(node.left, ast.Constant) and isinstance(node.right, ast.Constant):
                if isinstance(node.left.value, str) and isinstance(node.right.value, str):
                    return ast.Constant(value=node.left.value + node.right.value)
        return node

class ComplexityAnalyzer(ast.NodeVisitor):
    def __init__(self):
        self.complexity = 0
        
    def visit_If(self, node):
        self.complexity += 1
        self.generic_visit(node)
        
    def visit_For(self, node):
        self.complexity += 2
        self.generic_visit(node)
        
    def visit_While(self, node):
        self.complexity += 2
        self.generic_visit(node)

# Использование
analyzer = ComplexityAnalyzer()
analyzer.visit(tree)
print(f"Сложность кода: {analyzer.complexity}")

import jmespath

jmespath.search('foo.bar', {'foo': {'bar': 'baz'}})
# output: 'baz'

jmespath.search('foo.*.name', {'foo': {'bar': {'name': 'one'}, 'baz':
{'name': 'two'}}})
# output: ['one', 'two']

# Сборка образа
docker build -t my-image .

# Запуск контейнера
docker run -d --name my-container my-image

# Просмотр запущенных контейнеров
docker ps

# Остановка контейнера
docker stop my-container

version: '3'
services:
  web:
    build: .
    ports:
      - "5000:5000"
  redis:
    image: "redis:alpine"

# Создание volume
docker volume create my-vol

# Использование volume при запуске контейнера
docker run -v my-vol:/app/data my-image

from unittest.mock import patch
import asyncio

async def fetch_data(url):
    # Реальный запрос к API
    ...

@patch('your_module.fetch_data')
async def test_process_data(mock_fetch):
    mock_fetch.return_value = {'key': 'value'}
    result = await process_data('http://api.example.com')
    assert result == 'processed value'

import asyncio

async def test_order_of_execution():
    event = asyncio.Event()
    results = []

    async def task1():
        await event.wait()
        results.append(1)

    async def task2():
        results.append(2)
        event.set()

    await asyncio.gather(task1(), task2())
    assert results == [2, 1]

import pytest
import asyncio

@pytest.mark.asyncio
async def test_long_running_task():
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(never_ending_task(), timeout=1.0)

import pytest
import asyncio

@pytest.fixture
async def database():
    db = await create_database_connection()
    yield db
    await db.close()

@pytest.mark.asyncio
async def test_database_query(database):
    result = await database.fetch('SELECT * FROM users')
    assert len(result) > 0

pytest -n auto your_test_file.py

import asyncio
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

loop = asyncio.get_event_loop()
loop.set_default_executor(concurrent.futures.ThreadPoolExecutor(max_workers=10))

sem = asyncio.Semaphore(10)

async def controlled_task(i):
    async with sem:
        # Ваш асинхронный код здесь
        await asyncio.sleep(1)
        print(f"Задача {i} выполнена!")

class AsyncContextManager:
    async def __aenter__(self):
        print("Entering the matrix...")
        await asyncio.sleep(1)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        print("Exiting the matrix...")
        await asyncio.sleep(1)

async def main():
    async with AsyncContextManager() as manager:
        print("We're in!")

async def async_range(start, stop):
    for i in range(start, stop):
        await asyncio.sleep(0.1)
        yield i

async def main():
    async for num in async_range(0, 10):
        print(num)

MyClass = type('MyClass', (), {'x': 42, 'my_method': lambda self: print("Hello!")})

obj = MyClass()
print(obj.x)  # Выведет: 42
obj.my_method()  # Выведет: Hello!

class MyMetaclass(type):
    def __new__(cls, name, bases, attrs):
        attrs['additional_method'] = lambda self: print("I'm additional!")
        return super().__new__(cls, name, bases, attrs)

class MyClass(metaclass=MyMetaclass):
    pass

obj = MyClass()
obj.additional_method()  # Выведет: I'm additional!

from functools import cache

@cache
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

print(fibonacci(100))  # Мгновенный результат даже для больших чисел

from functools import wraps

def my_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        """Это документация обертки"""
        print("До вызова функции")
        result = func(*args, **kwargs)
        print("После вызова функции")
        return result
    return wrapper

@my_decorator
def say_hello(name):
    """Приветствует пользователя по имени"""
    print(f"Привет, {name}!")

say_hello("Мария")
print(say_hello.__name__)  # Выведет: say_hello
print(say_hello.__doc__)   # Выведет: Приветствует пользователя по имени

from functools import reduce

numbers = [1, 2, 3, 4, 5]
product = reduce(lambda x, y: x * y, numbers)
print(product)  # Выведет: 120

from functools import singledispatch

@singledispatch
def process(arg):
    print(f"Обработка объекта: {arg}")

@process.register(int)
def _(arg):
    print(f"Обработка целого числа: {arg}")

@process.register(list)
def _(arg):
    print(f"Обработка списка длиной {len(arg)}")

process("строка")  # Обработка объекта: строка
process(42)        # Обработка целого числа: 42
process([1, 2, 3]) # Обработка списка длиной 3

from functools import total_ordering

@total_ordering
class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age

    def __eq__(self, other):
        return self.age == other.age

    def __lt__(self, other):
        return self.age < other.age

p1 = Person("Алиса", 25)
p2 = Person("Боб", 30)

print(p1 < p2)  # True
print(p1 <= p2) # True
print(p1 > p2)  # False
print(p1 >= p2) # False

# Установка библиотеки
!pip install fugue

# Импорт необходимых модулей
from fugue import FugueEngine
from fugue.spark import SparkExecutionEngine

# Пример обработки данных
def process_data(df):
    return df.groupby("category").agg({"value": "sum"})

# Инициализация движка
engine = SparkExecutionEngine()

# Чтение данных
data = [
    {"category": "A", "value": 10},
    {"category": "B", "value": 20},
    {"category": "A", "value": 30},
]

# Обработка данных
result = engine.run(data, process_data)

# Вывод результата
print(result)

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from lineapy import LineaPy

# Создаем синтетические данные
np.random.seed(42)
X = np.random.rand(100, 1) * 10  # 100 случайных точек в диапазоне от 0 до 10
y = 2.5 * X + np.random.randn(100, 1) * 2  # Линейная зависимость с шумом

# Преобразуем данные в DataFrame
data = pd.DataFrame(np.hstack((X, y)), columns=['X', 'y'])

# Создаем модель с помощью LineaPy
model = LineaPy()
model.fit(data['X'], data['y'])

# Предсказания
predictions = model.predict(data['X'])

# Визуализация
plt.scatter(data['X'], data['y'], color='blue', label='Данные')
plt.plot(data['X'], predictions, color='red', label='Предсказание')
plt.xlabel('X')
plt.ylabel('y')
plt.title('Линейная Регрессия с LineaPy')
plt.legend()
plt.show()