💻 CODE GUIDE - Стандарты кодирования

# ❌ Плохо
def f(x, y):
    return x + y

# ✅ Хорошо
def calculate_total_price(base_price: float, tax: float) -> float:
    """
    Вычисляет итоговую цену с учётом налога.

    Args:
        base_price: Базовая цена товара
        tax: Налог (в процентах, например 20 для 20%)

    Returns:
        Итоговая цена с налогом

    Example:
        >>> calculate_total_price(100, 20)
        120.0
    """
    return base_price * (1 + tax / 100)

# ❌ Плохо - очевидный комментарий
# Увеличиваем счётчик на 1
counter += 1

# ✅ Хорошо - объясняет ПОЧЕМУ
# Ozon API требует задержку минимум 0.6 сек между запросами (100 req/min)
time.sleep(0.6)

# ✅ Хорошо - TODO с контекстом
# TODO: Добавить retry logic после интеграции с СДЭК API
# Issue: #123

# ✅ snake_case для переменных и функций
user_name = "Иван"
order_total = 1500.00

def fetch_ozon_orders():
    pass

def calculate_delivery_cost():
    pass

# ✅ PascalCase для классов
class OrderProcessor:
    pass

class OzonAPIClient:
    pass

class DatabaseHelper:
    pass

# ✅ UPPER_CASE для констант
MAX_RETRIES = 3
API_TIMEOUT = 10
DEFAULT_PAGE_SIZE = 50

# config.py
OZON_API_URL = "https://api-seller.ozon.ru"
CDEK_API_URL = "https://api.cdek.ru/v2"

class OrderService:
    def process_order(self, order_id):
        """Публичный метод"""
        self._validate_order(order_id)
        self._save_to_db(order_id)

    def _validate_order(self, order_id):
        """Приватный метод (одно подчёркивание)"""
        pass

    def _save_to_db(self, order_id):
        """Приватный метод"""
        pass

from typing import List, Dict, Optional, Union
from datetime import datetime

# ✅ Всегда использовать type hints
def fetch_orders(
    date_from: datetime,
    date_to: datetime,
    status: Optional[str] = None
) -> List[Dict[str, any]]:
    """Получить заказы за период"""
    pass

# ✅ Type hints для переменных (если неочевидно)
orders: List[Dict] = []
user_id: Optional[int] = None

# ✅ Type hints для моделей SQLAlchemy
from sqlalchemy.orm import Session

def get_user(db: Session, user_id: int) -> Optional[User]:
    return db.query(User).filter(User.id == user_id).first()

import logging
from typing import Optional

logger = logging.getLogger(__name__)

# ✅ Специфичные исключения
class OzonAPIError(Exception):
    """Базовое исключение для Ozon API"""
    pass

class OzonAuthError(OzonAPIError):
    """Ошибка авторизации"""
    pass

class OzonRateLimitError(OzonAPIError):
    """Превышен лимит запросов"""
    pass

# ✅ Обработка с логированием
def fetch_data_from_api(url: str) -> Optional[dict]:
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.Timeout:
        logger.error(f"Timeout при запросе к {url}")
        return None
    except requests.RequestException as e:
        logger.error(f"Ошибка при запросе к {url}: {e}")
        return None
    except Exception as e:
        logger.exception(f"Неожиданная ошибка: {e}")
        raise

# ✅ Не глушить исключения
try:
    risky_operation()
except Exception:
    pass  # ❌ НИКОГДА так не делать!

# ✅ Всегда логировать или re-raise
try:
    risky_operation()
except ValueError as e:
    logger.warning(f"Некорректное значение: {e}")
    # Обработать или пробросить дальше
except Exception as e:
    logger.exception("Критическая ошибка")
    raise  # Пробросить дальше

import logging

# ✅ Настройка логгера
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

# ✅ Уровни логов
logger.debug("Детальная информация для отладки")
logger.info("Заказ #12345 обработан успешно")
logger.warning("Товар SKU-001 заканчивается (осталось 5 шт)")
logger.error("Ошибка при подключении к Ozon API")
logger.critical("База данных недоступна!")

# ✅ Параметризированное логирование (эффективнее)
logger.info("Заказ %s обработан за %.2f сек", order_id, elapsed_time)

# ❌ Плохо - конкатенация
logger.info("Заказ " + str(order_id) + " обработан")

# ✅ f-strings для сложных сообщений
logger.error(f"Ошибка обработки заказа {order_id}: {error_details}")

# ✅ Одна функция = одна ответственность
def get_user_by_email(db: Session, email: str) -> Optional[User]:
    """Получить пользователя по email"""
    return db.query(User).filter(User.email == email).first()

# ✅ Короткие функции (< 50 строк)
# ✅ Мало параметров (< 5)
def create_order(
    db: Session,
    user_id: int,
    items: List[Dict],
    shipping_address: str,
    delivery_type: str = "pvz"
) -> Order:
    """Создать заказ"""
    pass

# ❌ Плохо - слишком много параметров
def create_order(db, user_id, item1, item2, item3, addr1, addr2, city, zip, country, phone, email, notes):
    pass

# ✅ Хорошо - использовать dataclass или Pydantic
from dataclasses import dataclass

@dataclass
class OrderData:
    user_id: int
    items: List[Dict]
    shipping_address: str
    delivery_type: str = "pvz"

def create_order(db: Session, order_data: OrderData) -> Order:
    """Создать заказ"""
    pass

# ✅ Порядок импортов:
# 1. Стандартная библиотека
import os
import sys
from datetime import datetime, timedelta
from typing import List, Dict, Optional

# 2. Сторонние библиотеки
import streamlit as st
import pandas as pd
from sqlalchemy.orm import Session

# 3. Локальные модули
from modules.database.models import User, Order
from modules.api.ozon import OzonAPI
from config import settings

# ✅ Абсолютные импорты вместо относительных
from modules.database.models import User  # ✅ Хорошо
from ..database.models import User  # ❌ Избегать

# ✅ Не импортировать всё
from datetime import datetime  # ✅
from datetime import *  # ❌

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from datetime import datetime

# ✅ Явное именование таблиц
class User(Base):
    __tablename__ = "users"

    # ✅ Всегда указывать типы
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(255), unique=True, nullable=False, index=True)
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)

    # ✅ Relationships с lazy loading
    orders = relationship("Order", back_populates="user", lazy="selectin")

# ✅ Composite indexes для частых запросов
class Order(Base):
    __tablename__ = "orders"

    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    status = Column(String(50), nullable=False, index=True)
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)

    # ✅ Composite index для фильтрации
    __table_args__ = (
        Index('ix_order_user_status', 'user_id', 'status'),
    )

from sqlalchemy.orm import Session

# ✅ Использовать session context manager
from contextlib import contextmanager

@contextmanager
def get_db_session():
    db = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

# Использование
with get_db_session() as db:
    user = db.query(User).filter(User.id == 1).first()

# ✅ N+1 query problem - использовать joinedload
from sqlalchemy.orm import joinedload

# ❌ Плохо - N+1 queries
users = db.query(User).all()
for user in users:
    print(user.orders)  # Отдельный запрос для каждого user!

# ✅ Хорошо - 1 query
users = db.query(User).options(joinedload(User.orders)).all()
for user in users:
    print(user.orders)  # Уже загружено

# ✅ Фильтрация с параметрами (защита от SQL injection)
email = user_input
user = db.query(User).filter(User.email == email).first()  # ✅ Безопасно

# ❌ НИКОГДА не использовать f-strings в SQL
db.execute(f"SELECT * FROM users WHERE email = '{email}'")  # ❌ SQL INJECTION!

# ✅ Всегда создавать миграции для изменений схемы
# Создание новой миграции
alembic revision --autogenerate -m "add fulfillment_scheme to orders"

# ✅ Проверять миграцию перед применением
# alembic/versions/xxx_add_fulfillment_scheme.py

def upgrade():
    # ✅ Добавлять значения по умолчанию
    op.add_column('orders', 
        sa.Column('fulfillment_scheme', sa.String(20), 
                  nullable=False, server_default='fbs')
    )

    # ✅ Создавать индексы
    op.create_index('ix_orders_fulfillment_scheme', 
                    'orders', ['fulfillment_scheme'])

def downgrade():
    # ✅ Всегда писать downgrade
    op.drop_index('ix_orders_fulfillment_scheme', 'orders')
    op.drop_column('orders', 'fulfillment_scheme')

# Применить миграцию
alembic upgrade head

# Откатить миграцию
alembic downgrade -1

# ✅ Использовать транзакции для связанных операций
def transfer_stock(db: Session, from_warehouse_id: int, to_warehouse_id: int, sku: str, quantity: int):
    try:
        # Уменьшаем остаток на складе-источнике
        from_stock = db.query(Stock).filter(
            Stock.warehouse_id == from_warehouse_id,
            Stock.sku == sku
        ).first()
        from_stock.quantity -= quantity

        # Увеличиваем остаток на складе-назначении
        to_stock = db.query(Stock).filter(
            Stock.warehouse_id == to_warehouse_id,
            Stock.sku == sku
        ).first()
        to_stock.quantity += quantity

        # Логируем перемещение
        movement = StockMovement(
            from_warehouse_id=from_warehouse_id,
            to_warehouse_id=to_warehouse_id,
            sku=sku,
            quantity=quantity
        )
        db.add(movement)

        db.commit()  # ✅ Всё или ничего
    except Exception as e:
        db.rollback()  # ✅ Откат при ошибке
        logger.error(f"Ошибка перемещения товара: {e}")
        raise

import streamlit as st

# ✅ Всегда устанавливать page_config первым
st.set_page_config(
    page_title="Заказы | MP1",
    page_icon="📦",
    layout="wide"
)

# ✅ Использовать session_state для данных между запусками
if 'orders' not in st.session_state:
    st.session_state.orders = []

# ✅ Структура: заголовок → фильтры → данные → действия
st.title("📦 Заказы")

# Фильтры
col1, col2, col3 = st.columns(3)
with col1:
    date_from = st.date_input("От")
with col2:
    date_to = st.date_input("До")
with col3:
    status = st.selectbox("Статус", ["Все", "Новые", "В сборке"])

# Данные
df = load_orders(date_from, date_to, status)
st.dataframe(df, use_container_width=True)

# Действия
if st.button("Загрузить новые заказы"):
    with st.spinner("Загрузка..."):
        fetch_new_orders()
    st.success("Заказы загружены")

# ✅ Использовать st.form для группировки полей
with st.form("order_form"):
    st.subheader("Создать заказ")

    customer_name = st.text_input("Имя покупателя")
    customer_phone = st.text_input("Телефон")
    delivery_address = st.text_area("Адрес доставки")

    submitted = st.form_submit_button("Создать")
    if submitted:
        if not customer_name or not customer_phone:
            st.error("Заполните обязательные поля")
        else:
            create_order(customer_name, customer_phone, delivery_address)
            st.success("Заказ создан")

# ❌ Плохо - кнопки вне формы перезапускают страницу при каждом клике
name = st.text_input("Имя")
if st.button("Сохранить"):  # Перезапуск при каждом изменении name
    save_name(name)

import streamlit as st

# ✅ Кешировать тяжёлые операции
@st.cache_data(ttl=3600)  # Кеш на 1 час
def load_large_dataset():
    """Загрузка больших данных"""
    return pd.read_csv("large_file.csv")

# ✅ Кешировать подключения к БД
@st.cache_resource
def get_database_connection():
    """Singleton подключение к БД"""
    return create_engine(DATABASE_URL)

# ✅ Кешировать API запросы
@st.cache_data(ttl=600)  # 10 минут
def fetch_ozon_orders(date_from: str, date_to: str):
    api = OzonAPI()
    return api.fetch_orders(date_from, date_to)

# ❌ Не кешировать изменяемые данные без ttl
@st.cache_data  # ❌ Данные могут устареть
def get_current_orders():
    return db.query(Order).all()

# ✅ Использовать columns для компактности
col1, col2, col3, col4 = st.columns([3, 2, 2, 1])
with col1:
    st.write("📦 Заказ #12345")
with col2:
    st.write("Новый")
with col3:
    st.write("1500 ₽")
with col4:
    if st.button("⚙️", key="order_12345"):
        st.write("Действия")

# ✅ Использовать expander для деталей
with st.expander("Подробности заказа"):
    st.json(order_details)

# ✅ Spinner для долгих операций
with st.spinner("Обработка заказа..."):
    time.sleep(2)
    process_order()

# ✅ Progress bar для пакетных операций
progress_bar = st.progress(0)
for i, order in enumerate(orders):
    process_order(order)
    progress_bar.progress((i + 1) / len(orders))

# ✅ Формат: <тип>: <описание>
git commit -m "feat: добавлена интеграция СДЭК API"
git commit -m "fix: исправлена загрузка заказов realFBS"
git commit -m "docs: обновлён MP1-API-GUIDE.md"
git commit -m "refactor: оптимизирован OzonAPI.fetch_orders"

# Типы коммитов:
# feat     - новая функциональность
# fix      - исправление бага
# docs     - документация
# style    - форматирование кода
# refactor - рефакторинг без изменения функциональности
# test     - добавление тестов
# chore    - рутинные задачи (обновление зависимостей и т.д.)

# ✅ Атомарные коммиты (одно изменение = один коммит)
git add modules/api/cdek.py
git commit -m "feat: добавлен CdekAPI.create_order"

git add modules/api/cdek.py
git commit -m "feat: добавлен CdekAPI.get_label"

# ❌ Плохо - всё в одном коммите
git add .
git commit -m "добавлено много всего"

# ✅ Работать в feature branches
git checkout -b feature/cdek-integration
# ... работа ...
git commit -m "feat: интеграция СДЭК"
git checkout master
git merge feature/cdek-integration

# ✅ Именование веток
feature/cdek-integration    # новая функциональность
fix/order-loading-bug       # исправление бага
docs/update-api-guide       # документация
refactor/optimize-queries   # рефакторинг

# ❌ Плохо
git checkout -b test
git checkout -b new-branch

# ✅ Всегда игнорировать
.env
*.pyc
__pycache__/
venv/
.venv/
.DS_Store
.idea/
.vscode/
*.log
*.sqlite

# Secrets
secrets/
credentials.json
*.pem
*.key

# Data
data/*.csv
data/*.xlsx

# ✅ Использовать переменные окружения
import os
from dotenv import load_dotenv

load_dotenv()

OZON_CLIENT_ID = os.getenv("OZON_CLIENT_ID")
OZON_API_KEY = os.getenv("OZON_API_KEY")
DATABASE_URL = os.getenv("DATABASE_URL")

# ❌ НИКОГДА не коммитить секреты
OZON_API_KEY = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"  # ❌

# ✅ Использовать параметризованные запросы
user_input = request.form['email']
user = db.query(User).filter(User.email == user_input).first()  # ✅

# ❌ НИКОГДА не конкатенировать SQL
query = f"SELECT * FROM users WHERE email = '{user_input}'"  # ❌ SQL INJECTION
db.execute(query)

# ✅ Streamlit автоматически экранирует HTML
st.write(user_input)  # ✅ Безопасно

# ❌ Не использовать st.markdown с unsafe_allow_html без санитизации
st.markdown(user_input, unsafe_allow_html=True)  # ❌ XSS уязвимость

# ✅ Если нужен HTML - санитизировать
import bleach
safe_html = bleach.clean(user_input)
st.markdown(safe_html, unsafe_allow_html=True)

from pydantic import BaseModel, validator, Field
from typing import Optional

# ✅ Использовать Pydantic для валидации
class OrderCreate(BaseModel):
    customer_name: str = Field(..., min_length=2, max_length=100)
    customer_phone: str = Field(..., regex=r'^\+?\d{10,15}$')
    total_amount: float = Field(..., gt=0)

    @validator('customer_phone')
    def validate_phone(cls, v):
        # Дополнительная валидация
        if not v.startswith('+7'):
            raise ValueError('Телефон должен начинаться с +7')
        return v

# Использование
try:
    order_data = OrderCreate(**request_data)
except ValidationError as e:
    logger.error(f"Ошибка валидации: {e}")
    return {"error": "Некорректные данные"}

# ❌ Плохо - загружаем всё
all_orders = db.query(Order).all()  # 10000 заказов!

# ✅ Хорошо - пагинация
page = 1
page_size = 50
orders = db.query(Order).limit(page_size).offset((page - 1) * page_size).all()

# ✅ Хорошо - загружаем только нужные поля
from sqlalchemy import select

stmt = select(Order.id, Order.status, Order.total_amount)
orders = db.execute(stmt).all()

# ✅ Использовать bulk операции
# ❌ Плохо
for order_data in orders_data:
    order = Order(**order_data)
    db.add(order)
    db.commit()  # 1000 коммитов!

# ✅ Хорошо
db.bulk_insert_mappings(Order, orders_data)
db.commit()  # 1 коммит

from functools import lru_cache
import redis

# ✅ Python LRU cache для дорогих вычислений
@lru_cache(maxsize=128)
def calculate_delivery_cost(from_city: str, to_city: str, weight: int) -> float:
    # Сложные вычисления
    return cost

# ✅ Redis для кеширования API ответов
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def get_ozon_warehouses():
    cache_key = "ozon:warehouses"
    cached = redis_client.get(cache_key)

    if cached:
        return json.loads(cached)

    # Запрос к API
    warehouses = ozon_api.get_warehouses()

    # Кеш на 1 час
    redis_client.setex(cache_key, 3600, json.dumps(warehouses))

    return warehouses

import asyncio
import aiohttp

# ✅ Параллельные API запросы
async def fetch_all_marketplaces():
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_ozon_orders(session),
            fetch_wildberries_orders(session),
            fetch_yandex_orders(session)
        ]
        results = await asyncio.gather(*tasks)
    return results

# ✅ Вместо последовательных запросов (медленно)
ozon_orders = fetch_ozon_orders()  # 2 сек
wb_orders = fetch_wb_orders()      # 2 сек
# Итого: 4 секунды

# ✅ Параллельно (быстро)
orders = await fetch_all_marketplaces()
# Итого: 2 секунды

import pytest
from modules.api.ozon import OzonAPI

# ✅ Именование: test_<что_тестируем>_<ожидаемый_результат>
def test_calculate_total_price_with_tax():
    result = calculate_total_price(100, 20)
    assert result == 120.0

def test_calculate_total_price_zero_tax():
    result = calculate_total_price(100, 0)
    assert result == 100.0

def test_calculate_total_price_negative_raises_error():
    with pytest.raises(ValueError):
        calculate_total_price(-100, 20)

# ✅ Использовать fixtures
@pytest.fixture
def db_session():
    """Тестовая сессия БД"""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.close()

def test_create_user(db_session):
    user = User(email="test@example.com")
    db_session.add(user)
    db_session.commit()

    assert user.id is not None
    assert user.email == "test@example.com"

# ✅ Мокать внешние API
from unittest.mock import Mock, patch

def test_fetch_ozon_orders():
    with patch('modules.api.ozon.requests.post') as mock_post:
        mock_post.return_value.json.return_value = {
            "result": {"postings": []}
        }

        api = OzonAPI(client_id="test", api_key="test")
        orders = api.fetch_orders("2025-11-01", "2025-11-08")

        assert orders == []
        assert mock_post.called

# ✅ Тестировать реальные сценарии
def test_order_creation_workflow(db_session):
    # 1. Создать пользователя
    user = User(email="test@example.com")
    db_session.add(user)
    db_session.commit()

    # 2. Создать заказ
    order = Order(
        user_id=user.id,
        status="new",
        total_amount=1500.00
    )
    db_session.add(order)
    db_session.commit()

    # 3. Проверить
    assert order.id is not None
    assert order.user_id == user.id
    assert db_session.query(Order).count() == 1

// ✅ Тестировать UI workflow
test('создание заказа через интерфейс', async ({ page }) => {
  await page.goto('http://localhost:8502');

  // Переход на страницу заказов
  await page.click('text=Заказы');
  await expect(page).toHaveURL(/.*orders/);

  // Создание заказа
  await page.fill('input[name="customer_name"]', 'Иван Иванов');
  await page.fill('input[name="customer_phone"]', '+79991234567');
  await page.click('button:has-text("Создать заказ")');

  // Проверка успеха
  await expect(page.locator('text=Заказ создан')).toBeVisible();
});

# Запустить перед коммитом
flake8 modules/
mypy modules/
pytest tests/