diff.py - master - Soitch/bot_obormot

diff.py
191 строка7 KB
Igor Sobolev | Coral Travel Russia
first_commit
05 дек 2025, 15:10
05 дек 2025, 15:10530e554
# diff.py

from typing import Dict, Any, List, TypedDict

class PriceChange(TypedDict):
    """
    Типизированный словарь для хранения информации об изменении цены
    """
    sku: str
    name: str
    old_price: float
    new_price: float
    diff_abs: float
    diff_rel: float

def compute_changes(
    old: Dict[str, Dict[str, Any]],
    new: Dict[str, Dict[str, Any]],
    min_rel_change: float = 0.05  # 5%
) -> List[PriceChange]:
    """
    Вычисляет изменения цен между двумя наборами данных
    
    Args:
        old: Старые цены в формате {sku: {"name": "...", "price": 123.45}}
        new: Новые цены в том же формате
        min_rel_change: Минимальное относительное изменение для фильтрации (по умолчанию 5%)
        
    Returns:
        Список изменений, где разница превышает min_rel_change
    """
    changes: List[PriceChange] = []
    
    for sku, item in new.items():
        if sku not in old:
            # Новая позиция — можно отдельно обрабатывать
            continue
        
        try:
            old_price = float(old[sku].get("price", 0))
            new_price = float(item.get("price", 0))
            
            if old_price <= 0:
                continue
            
            diff_abs = new_price - old_price
            diff_rel = diff_abs / old_price
            
            if abs(diff_rel) < min_rel_change:
                continue
            
            changes.append(
                PriceChange(
                    sku=sku,
                    name=item.get("name", "Без названия"),
                    old_price=old_price,
                    new_price=new_price,
                    diff_abs=diff_abs,
                    diff_rel=diff_rel,
                )
            )
            
        except (ValueError, KeyError, TypeError) as e:
            # Пропускаем товары с некорректными данными
            print(f"Ошибка при обработке SKU {sku}: {e}")
            continue
    
    return changes

def find_new_products(
    old: Dict[str, Dict[str, Any]],
    new: Dict[str, Dict[str, Any]]
) -> List[Dict[str, Any]]:
    """
    Находит новые товары, которых не было в старом наборе
    
    Returns:
        Список новых товаров
    """
    new_products = []
    
    for sku, item in new.items():
        if sku not in old:
            new_products.append({
                "sku": sku,
                "name": item.get("name", "Без названия"),
                "price": float(item.get("price", 0))
            })
    
    return new_products

def find_removed_products(
    old: Dict[str, Dict[str, Any]],
    new: Dict[str, Dict[str, Any]]
) -> List[Dict[str, Any]]:
    """
    Находит товары, которые исчезли из нового набора
    
    Returns:
        Список удаленных товаров
    """
    removed_products = []
    
    for sku, item in old.items():
        if sku not in new:
            removed_products.append({
                "sku": sku,
                "name": item.get("name", "Без названия"),
                "price": float(item.get("price", 0))
            })
    
    return removed_products

def format_price_change_message(change: PriceChange) -> str:
    """
    Форматирует информацию об изменении цены в читаемое сообщение
    
    Args:
        change: Информация об изменении цены
        
    Returns:
        Отформатированная строка
    """
    direction = "📈 выросла" if change["diff_abs"] > 0 else "📉 снизилась"
    percent = abs(change["diff_rel"]) * 100
    
    return (
        f"🔄 {change['name']} ({change['sku']})\n"
        f"💰 Цена {direction}:\n"
        f"   Было: {change['old_price']:.2f} руб.\n"
        f"   Стало: {change['new_price']:.2f} руб.\n"
        f"   Изменение: {change['diff_abs']:+.2f} руб. ({percent:+.1f}%)"
    )

def format_new_product_message(product: Dict[str, Any]) -> str:
    """
    Форматирует информацию о новом товаре
    """
    return (
        f"🆕 Новый товар!\n"
        f"📦 {product['name']} ({product['sku']})\n"
        f"💰 Цена: {product['price']:.2f} руб."
    )

def format_removed_product_message(product: Dict[str, Any]) -> str:
    """
    Форматирует информацию об удаленном товаре
    """
    return (
        f"🗑️ Товар удален\n"
        f"📦 {product['name']} ({product['sku']})\n"
        f"💸 Последняя цена: {product['price']:.2f} руб."
    )

# Тестирование
if __name__ == "__main__":
    print("Тестирование diff.py...")
    
    # Тестовые данные
    old_prices = {
        "sku_4070": {"name": "RTX 4070", "price": 60000.0},
        "sku_4080": {"name": "RTX 4080", "price": 90000.0},
        "sku_4090": {"name": "RTX 4090", "price": 140000.0},
    }
    
    new_prices = {
        "sku_4070": {"name": "RTX 4070", "price": 65000.0},  # +8.3%
        "sku_4080": {"name": "RTX 4080", "price": 92000.0},  # +2.2% (не будет в изменениях при пороге 5%)
        "sku_4090": {"name": "RTX 4090", "price": 135000.0}, # -3.6% (не будет в изменениях)
        "sku_5060": {"name": "RTX 5060", "price": 50000.0},  # Новый товар
    }
    
    # Вычисляем изменения
    changes = compute_changes(old_prices, new_prices)
    print(f"Найдено {len(changes)} значимых изменений:")
    for change in changes:
        print(f"\n{format_price_change_message(change)}")
    
    # Находим новые товары
    new_products = find_new_products(old_prices, new_prices)
    if new_products:
        print(f"\n\nНайдено {len(new_products)} новых товаров:")
        for product in new_products:
            print(f"\n{format_new_product_message(product)}")
    
    # Находим удаленные товары
    removed_products = find_removed_products(old_prices, new_prices)
    if removed_products:
        print(f"\n\nНайдено {len(removed_products)} удаленных товаров:")
        for product in removed_products:
            print(f"\n{format_removed_product_message(product)}")