import os from typing import Any, Optional import httpx from langchain_core.tools import tool API_URL = os.getenv("API_URL", "https://levinaleksey-wb.hf.space").rstrip("/") HTTP_TIMEOUT = float(os.getenv("HTTP_TIMEOUT", "30.0")) ALLOWED_FINANCE_STATUSES = {"loss", "low_margin", "has_penalties", "profit_falling", "ok"} ALLOWED_FINANCE_SORT = {"profit", "gmv", "margin_net_pct", "roi_pct", "unit_profit_net"} ALLOWED_UNIT_ECONOMICS_SORT = {"unit_profit_net", "unit_revenue", "unit_cost"} async def _api_get(path: str, params: dict[str, Any] | None = None) -> Any: """Единый helper для GET-запросов к внешнему API.""" async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client: # Убираем параметры со значением None, чтобы не передавать их в URL if params: params = {k: v for k, v in params.items() if v is not None} response = await client.get(f"{API_URL}{path}", params=params) response.raise_for_status() return response.json() async def _api_post(path: str, body: dict[str, Any]) -> Any: """Единый helper для POST-запросов к внешнему API.""" async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client: response = await client.post(f"{API_URL}{path}", json=body) response.raise_for_status() return response.json() @tool async def get_finance_summary() -> Any: """Общая финансовая сводка за 30 дней. Возвращает: общую статистику (GMV, прибыль WB, чистую прибыль, рентабельность/маржу, сумму комиссий, логистики, хранения и кол-во убыточных товаров). """ return await _api_get("/finance/summary") @tool async def get_finance_products( limit: int = 50, status: Optional[str] = None, sort_by: str = "profit" ) -> Any: """Список товаров с основной финансовой информацией (маржа, прибыль, ROI). Args: limit: Сколько товаров вернуть (по умолчанию 50). status: Опциональный фильтр статуса. Допустимые: 'loss', 'low_margin', 'has_penalties', 'profit_falling', 'ok'. sort_by: Поле сортировки (по умолчанию 'profit'). Допустимые: 'profit', 'gmv', 'margin_net_pct', 'roi_pct', 'unit_profit_net'. """ if status is not None and status not in ALLOWED_FINANCE_STATUSES: raise ValueError( f"Некорректный status='{status}'. Допустимо: {sorted(ALLOWED_FINANCE_STATUSES)}" ) if sort_by not in ALLOWED_FINANCE_SORT: raise ValueError( f"Некорректный sort_by='{sort_by}'. Допустимо: {sorted(ALLOWED_FINANCE_SORT)}" ) return await _api_get("/finance/products", {"limit": limit, "status": status, "sort_by": sort_by}) @tool async def get_finance_product(nm_id: int) -> Any: """Полная финансовая карточка конкретного товара. Возвращает выручку, все виды расходов (комиссия, логистика, штрафы), юнит-экономику, динамику и ROI. Args: nm_id: Артикул Wildberries. """ return await _api_get(f"/finance/product/{nm_id}") @tool async def get_loss_products(limit: int = 50) -> Any: """Убыточные товары. Возвращает товары с отрицательной чистой прибылью (profit_net < 0). Args: limit: Сколько товаров вернуть. """ return await _api_get("/finance/loss", {"limit": limit}) @tool async def get_low_margin_products(threshold: float = 5.0, limit: int = 50) -> Any: """Товары с низкой маржой. Возвращает товары, у которых маржа (margin_net_pct) ниже заданного порога. Args: threshold: Порог маржи в процентах (по умолчанию 5.0%). limit: Сколько товаров вернуть. """ return await _api_get("/finance/low-margin", {"threshold": threshold, "limit": limit}) @tool async def get_top_profit_products(limit: int = 50) -> Any: """Самые прибыльные товары. Возвращает ТОП товаров по абсолютной чистой прибыли (profit_net > 0). Args: limit: Сколько товаров вернуть. """ return await _api_get("/finance/top-profit", {"limit": limit}) @tool async def get_top_roi_products(limit: int = 50) -> Any: """Товары с лучшим возвратом на инвестиции (ROI). Args: limit: Сколько товаров вернуть. """ return await _api_get("/finance/top-roi", {"limit": limit}) @tool async def get_unit_economics(limit: int = 50, sort_by: str = "unit_profit_net") -> Any: """Юнит-экономика товаров (в пересчете на 1 штуку). Разбивка доходов и расходов на 1 единицу проданного товара. Args: limit: Сколько товаров вернуть. sort_by: Поле сортировки (по умолчанию 'unit_profit_net'). Допустимые: 'unit_profit_net', 'unit_revenue', 'unit_cost'. """ if sort_by not in ALLOWED_UNIT_ECONOMICS_SORT: raise ValueError( f"Некорректный sort_by='{sort_by}'. Допустимо: {sorted(ALLOWED_UNIT_ECONOMICS_SORT)}" ) return await _api_get("/finance/unit-economics", {"limit": limit, "sort_by": sort_by}) @tool async def get_expenses_breakdown() -> Any: """Структура расходов продавца на WB. Разбивка в абсолютных значениях и процентах: комиссия, эквайринг, логистика, хранение, штрафы, платная приемка. """ return await _api_get("/finance/expenses-breakdown") @tool async def get_finance_trend(days: int = 30) -> Any: """Динамика финансов по дням (тренд). Возвращает: продажи, GMV, прибыль, комиссию и расходы по дням для построения графиков. Args: days: Период в днях (по умолчанию 30). """ return await _api_get("/finance/trend", {"days": days}) @tool async def get_category_report() -> Any: """Финансы агрегированные по категориям товаров. Аналитика по продажам, выручке, прибыли, средней марже и ROI в разрезе каждой категории (предмета) магазина. """ return await _api_get("/finance/category-report") @tool async def add_finance_insight(nm_id: int, tag: str, recommendation: str, score: float) -> Any: """Сохранить AI-рекомендацию по финансам для товара в базу. Args: nm_id: Артикул Wildberries. tag: Тег рекомендации (например 'reduce_cost', 'increase_price'). recommendation: Конкретное действие по улучшению юнит-экономики с цифрами. score: Уверенность от 0.5 до 1.0. """ return await _api_post("/insights/add", { "nm_id": nm_id, "tag": tag, "recommendation": recommendation, "score": score }) # Экспортируем все 12 инструментов одним списком finance_tools = [ get_finance_summary, get_finance_products, get_finance_product, get_loss_products, get_low_margin_products, get_top_profit_products, get_top_roi_products, get_unit_economics, get_expenses_breakdown, get_finance_trend, get_category_report, add_finance_insight, ]