mirror of
https://github.com/serty2005/rmser.git
synced 2026-02-05 03:12:34 -06:00
237 lines
10 KiB
Python
237 lines
10 KiB
Python
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() |