serty/rmser
1
0
Fork 0
mirror of https://github.com/serty2005/rmser.git synced 2026-02-04 19:02:33 -06:00
Code Issues Packages Projects Releases Wiki Activity

добавил гигачада и заворачивание в проверку чека, если данные для QR распознались

Browse Source
This commit is contained in:
SERTY
2025-12-23 07:37:35 +03:00
parent b5b9504019
commit 9441579a34
7 changed files with 223 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docker-compose.yml
View File

@@ -32,6 +32,8 @@ services:
- "5005:5000"
environment:
- LOG_LEVEL=INFO
- LLM_ENGINE=gigachat
- GIGACHAT_AUTH_KEY=MDE5YjQzNzgtNWFkOS03MmNmLWFiYjUtNjQ2NmJkMDM2ZjZlOjNkZjBlNDkzLWRlOTEtNGY4Yi04MDFjLWRiMzAxNDlmYTRmNw==
- YANDEX_OAUTH_TOKEN=y0__xDK_988GMHdEyDc2M_XFTDIv-CCCP0kok1p0yRYJCgQrj8b9Kwylo25
- YANDEX_FOLDER_ID=b1gas1sh12oui8cskgcm

4
ocr-service/Dockerfile
View File

@@ -6,6 +6,7 @@ FROM python:3.10-slim
# libgl1, libglib2.0-0: для работы OpenCV
# libzbar0: для сканирования QR-кодов
RUN apt-get update && apt-get install -y \
curl \
tesseract-ocr \
tesseract-ocr-rus \
libgl1 \
@@ -20,6 +21,9 @@ WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Установка сертификатов Минцифры для корректной работы GigaChat (requests использует certifi)
RUN curl -k "https://gu-st.ru/content/Other/doc/russian_trusted_root_ca.cer" -w "\n" >> $(python -m certifi)
# Копируем код приложения
COPY . .

136
ocr-service/llm_parser.py
View File

@@ -2,32 +2,29 @@ import os
import requests
import logging
import json
from typing import List
import uuid
import time
from typing import List, Optional
from parser import ParsedItem
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 YandexGPTParser:
def __init__(self):
self.folder_id = os.getenv("YANDEX_FOLDER_ID")
self.api_key = os.getenv("YANDEX_OAUTH_TOKEN") # Используем тот же доступ
self.api_key = os.getenv("YANDEX_OAUTH_TOKEN")
def parse_with_llm(self, raw_text: str, iam_token: str) -> List[ParsedItem]:
"""
Отправляет текст в YandexGPT для структурирования.
"""
def parse(self, raw_text: str, iam_token: str) -> List[ParsedItem]:
if not iam_token:
return []
prompt = {
"modelUri": f"gpt://{self.folder_id}/yandexgpt/latest",
"completionOptions": {
"stream": False,
"temperature": 0.1, # Низкая температура для точности
"maxTokens": "2000"
},
"completionOptions": {"stream": False, "temperature": 0.1, "maxTokens": "2000"},
"messages": [
{
"role": "system",
@@ -38,10 +35,7 @@ class YandexGPTParser:
"Если количество не указано, считай 1.0. Не пиши ничего, кроме JSON."
)
},
{
"role": "user",
"text": raw_text
}
{"role": "user", "text": raw_text}
]
}
@@ -54,21 +48,103 @@ class YandexGPTParser:
try:
response = requests.post(YANDEX_GPT_URL, headers=headers, json=prompt, timeout=30)
response.raise_for_status()
result = response.json()
# Извлекаем текст ответа
content = result['result']['alternatives'][0]['message']['text']
# Очищаем от возможных markdown-оберток ```json ... ```
content = response.json()['result']['alternatives'][0]['message']['text']
clean_json = content.replace("```json", "").replace("```", "").strip()
items_raw = json.loads(clean_json)
parsed_items = [ParsedItem(**item) for item in items_raw]
return parsed_items
return [ParsedItem(**item) for item in json.loads(clean_json)]
except Exception as e:
logger.error(f"LLM Parsing error: {e}")
logger.error(f"YandexGPT Parsing error: {e}")
return []
llm_parser = YandexGPTParser()
class GigaChatParser:
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'}
try:
# verify=False может понадобиться, если сертификаты Минцифры не в системном хранилище,
# но вы указали, что установите их в контейнер.
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
except Exception as e:
logger.error(f"GigaChat Auth error: {e}")
return None
def parse(self, raw_text: str) -> List[ParsedItem]:
token = self._get_token()
if not token:
return []
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json',
'Authorization': f'Bearer {token}'
}
payload = {
"model": "GigaChat",
"messages": [
{
"role": "system",
"content": (
"Ты — эксперт по распознаванию чеков. Извлеки товары из текста. "
"Верни ТОЛЬКО JSON массив объектов с полями: raw_name (строка), "
"amount (число), price (число), sum (число). "
"Если данных нет, верни []. Никаких пояснений."
)
},
{"role": "user", "content": raw_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']
clean_json = content.replace("```json", "").replace("```", "").strip()
return [ParsedItem(**item) for item in json.loads(clean_json)]
except Exception as e:
logger.error(f"GigaChat Parsing error: {e}")
return []
class LLMManager:
def __init__(self):
self.yandex = YandexGPTParser()
self.giga = GigaChatParser()
self.engine = os.getenv("LLM_ENGINE", "yandex").lower()
def parse_with_priority(self, raw_text: str, yandex_iam_token: Optional[str] = None) -> List[ParsedItem]:
if self.engine == "gigachat":
logger.info("Using GigaChat as primary LLM")
items = self.giga.parse(raw_text)
if not items and yandex_iam_token:
logger.info("GigaChat failed, falling back to YandexGPT")
items = self.yandex.parse(raw_text, yandex_iam_token)
return items
else:
logger.info("Using YandexGPT as primary LLM")
items = self.yandex.parse(raw_text, yandex_iam_token) if yandex_iam_token else []
if not items:
logger.info("YandexGPT failed, falling back to GigaChat")
items = self.giga.parse(raw_text)
return items
llm_parser = LLMManager()

38
ocr-service/main.py
View File

@@ -9,7 +9,7 @@ import numpy as np
# Импортируем модули
from imgproc import preprocess_image
from parser import parse_receipt_text, ParsedItem
from parser import parse_receipt_text, ParsedItem, extract_fiscal_data
from ocr import ocr_engine
from qr_manager import detect_and_decode_qr, fetch_data_from_api
# Импортируем новый модуль
@@ -77,28 +77,38 @@ async def recognize_receipt(image: UploadFile = File(...)):
else:
logger.info("QR code not found. Proceeding to OCR.")
# --- ЭТАП 2: Yandex Vision Strategy (Cloud OCR) ---
# Проверяем, настроен ли Яндекс
# --- ЭТАП 2: OCR + Virtual QR Strategy ---
if yandex_engine.oauth_token and yandex_engine.folder_id:
logger.info("--- Stage 2: Yandex Vision OCR ---")
# Яндекс принимает сырые байты картинки (Base64), ему не нужен наш препроцессинг
logger.info("--- Stage 2: Yandex Vision OCR + Virtual QR ---")
yandex_text = yandex_engine.recognize(content)
if yandex_text and len(yandex_text) > 10:
logger.info(f"Yandex OCR success. Text length: {len(yandex_text)}")
logger.info(f"Yandex RAW OUTPUT:\n{yandex_text}")
logger.info(f"OCR success. Raw text length: {len(yandex_text)}")
# Попытка собрать виртуальный QR из текста
virtual_qr = extract_fiscal_data(yandex_text)
if virtual_qr:
logger.info(f"Virtual QR constructed: {virtual_qr}")
api_items = fetch_data_from_api(virtual_qr)
if api_items:
logger.info(f"Success: Retrieved {len(api_items)} items via Virtual QR API.")
return RecognitionResult(
source="virtual_qr_api",
items=api_items,
raw_text=yandex_text
)
# Если виртуальный QR не сработал, пробуем Regex
yandex_items = parse_receipt_text(yandex_text)
logger.info(f"Parsed items preview: {yandex_items[:3]}...")
# Если Regex не нашел позиций (как в нашем случае со счетом)
# Если Regex пуст — вызываем LLM (GigaChat / YandexGPT)
if not yandex_items:
logger.info("Regex found nothing. Calling YandexGPT for semantic parsing...")
logger.info("Regex found nothing. Calling LLM Manager...")
iam_token = yandex_engine._get_iam_token()
yandex_items = llm_parser.parse_with_llm(yandex_text, iam_token)
logger.info(f"Semantic parsed items preview: {yandex_items[:3]}...")
yandex_items = llm_parser.parse_with_priority(yandex_text, iam_token)
return RecognitionResult(
source="yandex_vision",
source="yandex_vision_llm",
items=yandex_items,
raw_text=yandex_text
)

112
ocr-service/parser.py
View File

@@ -1,6 +1,7 @@
import re
from typing import List
from typing import List, Optional
from pydantic import BaseModel
from datetime import datetime
class ParsedItem(BaseModel):
raw_name: str
@@ -13,20 +14,64 @@ FLOAT_RE = r'\d+[.,]\d{2}'
def clean_text(text: str) -> str:
"""Удаляет лишние символы из названия товара."""
# Оставляем буквы, цифры, пробелы и базовые знаки
return re.sub(r'[^\w\s.,%/-]', '', text).strip()
def parse_float(val: str) -> float:
"""Преобразует строку '123,45' или '123.45' в float."""
if not val:
return 0.0
# Заменяем запятую на точку и убираем возможные пробелы
return float(val.replace(',', '.').replace(' ', ''))
def extract_fiscal_data(text: str) -> Optional[str]:
"""
Ищет в тексте:
- Дата и время (t): 19.12.25 12:16 -> 20251219T1216
- Сумма (s): ИТОГ: 770.00
- ФН (fn): 16 цифр, начинается на 73
- ФД (i): до 8 цифр (ищем после Д: или ФД:)
- ФП (fp): 10 цифр (ищем после П: или ФП:)
Возвращает строку формата: t=20251219T1216&s=770.00&fn=7384440800514469&i=11194&fp=3334166168&n=1
"""
# 1. Поиск даты и времени
# Ищем форматы DD.MM.YY HH:MM или DD.MM.YYYY HH:MM
date_match = re.search(r'(\d{2})\.(\d{2})\.(\d{2,4})\s+(\d{2}):(\d{2})', text)
t_param = ""
if date_match:
d, m, y, hh, mm = date_match.groups()
if len(y) == 2: y = "20" + y
t_param = f"{y}{m}{d}T{hh}{mm}"
# 2. Поиск суммы (Итог)
# Ищем слово ИТОГ и число после него
sum_match = re.search(r'(?:ИТОГ|СУММА|СУММА:)\s*[:]*\s*(\d+[.,]\d{2})', text, re.IGNORECASE)
s_param = ""
if sum_match:
s_param = sum_match.group(1).replace(',', '.')
# 3. Поиск ФН (16 цифр, начинается с 73)
fn_match = re.search(r'\b(73\d{14})\b', text)
fn_param = fn_match.group(1) if fn_match else ""
# 4. Поиск ФД (i) - ищем после маркеров Д: или ФД:
# Берем набор цифр до 8 знаков
fd_match = re.search(r'(?:ФД|Д)[:\s]+(\d{1,8})\b', text)
i_param = fd_match.group(1) if fd_match else ""
# 5. Поиск ФП (fp) - ищем после маркеров П: или ФП:
# Строго 10 цифр
fp_match = re.search(r'(?:ФП|П)[:\s]+(\d{10})\b', text)
fp_param = fp_match.group(1) if fp_match else ""
# Валидация: для формирования запроса к API нам критически важны все параметры
if all([t_param, s_param, fn_param, i_param, fp_param]):
return f"t={t_param}&s={s_param}&fn={fn_param}&i={i_param}&fp={fp_param}&n=1"
return None
def parse_receipt_text(text: str) -> List[ParsedItem]:
"""
Парсит текст чека построчно.
Логика: накапливаем строки названия, пока не встретим строку с математикой (цена/сумма).
Парсит текст чека построчно (Regex-метод).
"""
lines = text.split('\n')
items = []
@@ -37,74 +82,38 @@ def parse_receipt_text(text: str) -> List[ParsedItem]:
if not line:
continue
# Ищем все числа, похожие на цену (с двумя знаками после запятой/точки)
# Пример: 129.99
floats = re.findall(FLOAT_RE, line)
# Эвристика: строка считается "товарной позицией с ценой", если в ней есть минимум 2 числа
# (обычно Цена и Сумма, или Кол-во и Сумма)
# ИЛИ одно число, если это итоговая сумма, но мы ищем товары.
is_price_line = False
if len(floats) >= 2:
is_price_line = True
vals = [parse_float(f) for f in floats]
# Попытка определить структуру: Цена x Кол-во = Сумма
price = 0.0
amount = 1.0
total = 0.0
# Обычно последнее число - это сумма (итог по строке)
total = vals[-1]
if len(vals) == 2:
# Скорее всего: Цена ... Сумма (кол-во = 1)
# Или: Кол-во ... Сумма (если цена не распозналась)
# Предположим amount=1, тогда первое число - цена
price = vals[0]
amount = 1.0
# Проверка на адекватность: сумма обычно >= цены
# Если total < price, возможно порядок перепутан или это скидка
if total < price and total != 0:
# Если total сильно меньше, возможно это не сумма
pass
elif total > price and price > 0:
# Пытаемся вычислить кол-во
if total > price and price > 0:
calc_amount = total / price
# Если результат близок к целому (например 1.999 -> 2), то ок
if abs(round(calc_amount) - calc_amount) < 0.05:
amount = float(round(calc_amount))
elif len(vals) >= 3:
# Варианты: [Цена, Кол-во, Сумма] или [Кол-во, Цена, Сумма]
# Проверяем математику: A * B = C
v1, v2 = vals[-3], vals[-2]
if abs(v1 * v2 - total) < 0.5: # Допуск 0.5 руб
price = v1
amount = v2
if abs(v1 * v2 - total) < 0.5:
price, amount = v1, v2
elif abs(v2 * v1 - total) < 0.5:
price = v2
amount = v1
price, amount = v2, v1
else:
# Если математика не сходится, берем предпоследнее как цену
price = vals[-2]
amount = 1.0
price, amount = vals[-2], 1.0
# Сборка названия
full_name = " ".join(name_buffer).strip()
# Если буфер пуст, возможно название в этой же строке слева
if not full_name:
# Удаляем найденные числа из строки, остаток считаем названием
text_without_floats = re.sub(FLOAT_RE, '', line)
full_name = clean_text(text_without_floats)
# Фильтрация мусора (слишком короткие названия или нулевые суммы)
if len(full_name) > 2 and total > 0:
items.append(ParsedItem(
raw_name=full_name,
@@ -112,21 +121,12 @@ def parse_receipt_text(text: str) -> List[ParsedItem]:
price=price,
sum=total
))
# Очищаем буфер, так как позиция закрыта
name_buffer = []
else:
# Строка не похожа на цену.
# Проверяем на стоп-слова (конец чека)
upper_line = line.upper()
if "ИТОГ" in upper_line or "СУММА" in upper_line or "ПРИХОД" in upper_line:
# Считаем, что товары закончились
if any(stop in upper_line for stop in ["ИТОГ", "СУММА", "ПРИХОД"]):
name_buffer = []
# Можно здесь сделать break, если уверены, что ниже товаров нет
continue
# Добавляем в буфер названия
name_buffer.append(line)
return items

29
ocr-service/python_project_dump.py Normal file
View File

File diff suppressed because one or more lines are too long

3
ocr-service/requirements.txt
View File

@@ -7,4 +7,5 @@ opencv-python-headless
pytesseract
requests
pyzbar
pillow
pillow
certifi