import os import requests from requests.exceptions import SSLError import logging import json import uuid import time import re from abc import ABC, abstractmethod from typing import List, Optional from app.schemas.models import ParsedItem from app.services.auth import yandex_auth logger = logging.getLogger(__name__) YANDEX_GPT_URL = "https://llm.api.cloud.yandex.net/foundationModels/v1/completion" GIGACHAT_OAUTH_URL = "https://ngw.devices.sberbank.ru:9443/api/v2/oauth" GIGACHAT_COMPLETION_URL = "https://gigachat.devices.sberbank.ru/api/v1/chat/completions" class LLMProvider(ABC): @abstractmethod def generate(self, system_prompt: str, user_text: str) -> str: pass class YandexGPTProvider(LLMProvider): def __init__(self): self.folder_id = os.getenv("YANDEX_FOLDER_ID") def generate(self, system_prompt: str, user_text: str) -> str: iam_token = yandex_auth.get_iam_token() if not iam_token: raise Exception("Failed to get IAM token") prompt = { "modelUri": f"gpt://{self.folder_id}/yandexgpt/latest", "completionOptions": {"stream": False, "temperature": 0.1, "maxTokens": "2000"}, "messages": [ {"role": "system", "text": system_prompt}, {"role": "user", "text": user_text} ] } headers = { "Content-Type": "application/json", "Authorization": f"Bearer {iam_token}", "x-folder-id": self.folder_id } response = requests.post(YANDEX_GPT_URL, headers=headers, json=prompt, timeout=30) response.raise_for_status() content = response.json()['result']['alternatives'][0]['message']['text'] return content class GigaChatProvider(LLMProvider): def __init__(self): self.auth_key = os.getenv("GIGACHAT_AUTH_KEY") self._access_token = None self._expires_at = 0 def _get_token(self) -> Optional[str]: if self._access_token and time.time() < self._expires_at: return self._access_token logger.info("Obtaining GigaChat access token...") headers = { 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json', 'RqUID': str(uuid.uuid4()), 'Authorization': f'Basic {self.auth_key}' } payload = {'scope': 'GIGACHAT_API_PERS'} response = requests.post(GIGACHAT_OAUTH_URL, headers=headers, data=payload, timeout=10) response.raise_for_status() data = response.json() self._access_token = data['access_token'] self._expires_at = data['expires_at'] / 1000 # Переводим мс в сек return self._access_token def generate(self, system_prompt: str, user_text: str) -> str: token = self._get_token() if not token: raise Exception("Failed to get GigaChat token") headers = { 'Content-Type': 'application/json', 'Accept': 'application/json', 'Authorization': f'Bearer {token}' } payload = { "model": "GigaChat", "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_text} ], "temperature": 0.1 } try: response = requests.post(GIGACHAT_COMPLETION_URL, headers=headers, json=payload, timeout=30) response.raise_for_status() content = response.json()['choices'][0]['message']['content'] return content except SSLError as e: logger.error("SSL Error with GigaChat. Check certificates") raise e class LLMManager: def __init__(self): engine = os.getenv("LLM_ENGINE", "yandex").lower() self.engine = engine if engine == "gigachat": self.provider = GigaChatProvider() else: self.provider = YandexGPTProvider() logger.info(f"LLM Engine initialized: {self.engine}") def parse_receipt(self, raw_text: str) -> dict: system_prompt = """ Ты — профессиональный бухгалтер. Твоя задача — извлечь из "сырого" текста (OCR или Excel) данные о товарах и реквизиты документа. ВХОДНЫЕ ДАННЫЕ: Текст чека, накладной или УПД. В Excel-файлах колонки разделены символом "|". ФОРМАТ ОТВЕТА (JSON): { "doc_number": "Номер документа (или ФД для чеков)", "doc_date": "Дата документа (DD.MM.YYYY)", "items": [ { "raw_name": "Название товара", "amount": 1.0, "price": 100.0, // Цена за единицу (с учетом скидок) "sum": 100.0 // Стоимость позиции (ВСЕГО с НДС) } ] } ПРАВИЛА ПОИСКА РЕКВИЗИТОВ: 1. Номер документа: Ищи "Счет-фактура №", "УПД №", "Накладная №", "ФД:", "Документ №". - Пример: "УПД № 556430/123" -> "556430/123" - Пример: "ФД: 12345" -> "12345" 2. Дата: Ищи рядом с номером. Преобразуй в формат DD.MM.YYYY. ОБЩИЕ ПРАВИЛА ОБРАБОТКИ ТОВАРОВ: 1. **ЗАПРЕТ ГРУППИРОВКИ:** Если в чеке 5 раз подряд идет "Масло сливочное" (даже с разными кодами), ты должен вернуть **5 отдельных объектов**. НИКОГДА не объединяй их и не суммируй количество, если это не указано явно в одной строке (типа "5 шт x 100"). 2. **ОЧИСТКА:** Игнорируй строки "ИТОГ", "СУММА", "НДС", "Всего к оплате", "Грузоотправитель", "Продавец". Числа приводи к float. СТРАТЕГИЯ ДЛЯ EXCEL (УПД): 1. Разделитель колонок — "|". 2. Название — самая длинная текстовая ячейка. 3. Сумма — колонка "Стоимость с налогом - всего" (обычно крайняя правая сумма в строке). Если есть "без налога" и "с налогом", бери С НАЛОГОМ. СТРАТЕГИЯ ДЛЯ ЧЕКОВ (OCR): 1. **МАРКЕР НАЧАЛА:** Часто строка товара начинается с цифрового кода (артикула). Пример: `6328 Масло...`. Если видишь число в начале строки, за которым идет текст — это НАЧАЛО нового товара. 2. **МНОГОСТРОЧНОСТЬ:** Название товара может быть разбито на 2-4 строки. - Если видишь строку с цифрами (цена, кол-во), а перед ней текст — это конец описания товара. Склей текст с предыдущими строками. - Пример: `6298 Масло` `ТРАД.сл.` `1 шт 174.24` -> Название: "6298 Масло ТРАД.сл." 3. **ЦЕНА:** Если указана "Цена со скидкой", бери её. ПРИМЕР 1 (УПД): Вход: Код | Товар | Кол-во | Цена | Сумма без НДС | Сумма с НДС 1 | Сок ананасовый | 4 | 533.64 | 2134.56 | 2348.00 Выход: { "doc_number": "", "doc_date": "", "items": [ {"raw_name": "Сок ананасовый", "amount": 4.0, "price": 533.64, "sum": 2348.00} ] } ПРИМЕР 2 (Сложный чек): Вход: 5603 СЫР ПЛАВ. 45% 200Г 169.99 1шт 169.99 6328 Масло ТРАД. Сл.82.5% 175г 204.99 30.75 174.24 1шт 174.24 6298 Масло ТРАД. Сл.82.5% 175г 1шт 174.24 Выход: { "doc_number": "", "doc_date": "", "items": [ {"raw_name": "5603 СЫР ПЛАВ. 45% 200Г", "amount": 1.0, "price": 169.99, "sum": 169.99}, {"raw_name": "6328 Масло ТРАД. Сл.82.5% 175г", "amount": 1.0, "price": 174.24, "sum": 174.24}, {"raw_name": "6298 Масло ТРАД. Сл.82.5% 175г", "amount": 1.0, "price": 174.24, "sum": 174.24} ] } """ try: logger.info(f"Parsing receipt using engine: {self.engine}") response = self.provider.generate(system_prompt, raw_text) # Очистка от Markdown clean_json = response.replace("```json", "").replace("```", "").strip() # Парсинг JSON data = json.loads(clean_json) # Обработка обратной совместимости: если вернулся список, оборачиваем в словарь if isinstance(data, list): data = {"items": data, "doc_number": "", "doc_date": ""} # Извлекаем товары items_data = data.get("items", []) # Пост-обработка: исправление чисел в товарах for item in items_data: for key in ['amount', 'price', 'sum']: if isinstance(item[key], str): # Удаляем пробелы внутри чисел item[key] = float(re.sub(r'\s+', '', item[key]).replace(',', '.')) elif isinstance(item[key], (int, float)): item[key] = float(item[key]) # Формируем результат result = { "items": [ParsedItem(**item) for item in items_data], "doc_number": data.get("doc_number", ""), "doc_date": data.get("doc_date", "") } return result except Exception as e: logger.error(f"LLM Parsing error: {e}") return {"items": [], "doc_number": "", "doc_date": ""} llm_parser = LLMManager()