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

2612-есть ок OCR, нужно допиливать бота под новый flow для операторов

Browse Source
This commit is contained in:
SERTY
2026-01-27 00:17:10 +03:00
parent 7d2ffb54b5
commit 1843cb9c20
22 changed files with 1011 additions and 577 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/main.go
View File

@@ -113,7 +113,7 @@ func main() {
// 8. Telegram Bot (Передаем syncService)
if cfg.Telegram.Token != "" {
bot, err := tgBot.NewBot(cfg.Telegram, ocrService, syncService, billingService, accountRepo, rmsFactory, cryptoManager)
bot, err := tgBot.NewBot(cfg.Telegram, ocrService, syncService, billingService, accountRepo, rmsFactory, cryptoManager, cfg.App.MaintenanceMode, cfg.App.DevIDs)
if err != nil {
logger.Log.Fatal("Ошибка создания Telegram бота", zap.Error(err))
}

184
go_backend_dump.py Normal file
View File

File diff suppressed because one or more lines are too long

2
internal/infrastructure/ocr_client/dto.go
View File

@@ -3,6 +3,8 @@ package ocr_client
// RecognitionResult - ответ от Python сервиса
type RecognitionResult struct {
Items []RecognizedItem `json:"items"`
DocNumber string `json:"doc_number"`
DocDate string `json:"doc_date"`
}
type RecognizedItem struct {

26
internal/services/ocr/service.go
View File

@@ -78,8 +78,8 @@ func (s *Service) checkWriteAccess(userID, serverID uuid.UUID) error {
return nil
}
// ProcessReceiptImage - Доступно всем (включая Операторов)
func (s *Service) ProcessReceiptImage(ctx context.Context, userID uuid.UUID, imgData []byte) (*drafts.DraftInvoice, error) {
// ProcessDocument - Доступно всем (включая Операторов)
func (s *Service) ProcessDocument(ctx context.Context, userID uuid.UUID, imgData []byte, filename string) (*drafts.DraftInvoice, error) {
server, err := s.accountRepo.GetActiveServer(userID)
if err != nil || server == nil {
return nil, fmt.Errorf("no active server for user")
@@ -90,7 +90,7 @@ func (s *Service) ProcessReceiptImage(ctx context.Context, userID uuid.UUID, img
photoID := uuid.New()
draftID := uuid.New()
fileName := fmt.Sprintf("receipt_%s.jpg", photoID.String())
fileName := fmt.Sprintf("receipt_%s_%s", photoID.String(), filename)
filePath := filepath.Join(s.storagePath, serverID.String(), fileName)
// 2. Создаем директорию если не существует
@@ -102,7 +102,7 @@ func (s *Service) ProcessReceiptImage(ctx context.Context, userID uuid.UUID, img
if err := os.WriteFile(filePath, imgData, 0644); err != nil {
return nil, fmt.Errorf("failed to save image: %w", err)
}
fileURL := "/uploads/" + fileName
fileURL := fmt.Sprintf("/uploads/%s/%s", serverID.String(), fileName)
// 4. Создаем запись ReceiptPhoto
photo := &photos.ReceiptPhoto{
@@ -111,7 +111,7 @@ func (s *Service) ProcessReceiptImage(ctx context.Context, userID uuid.UUID, img
UploadedBy: userID,
FilePath: filePath,
FileURL: fileURL,
FileName: fileName,
FileName: filename,
FileSize: int64(len(imgData)),
DraftID: &draftID, // Сразу связываем с будущим черновиком
CreatedAt: time.Now(),
@@ -140,14 +140,26 @@ func (s *Service) ProcessReceiptImage(ctx context.Context, userID uuid.UUID, img
}
// 6. Отправляем в Python OCR
rawResult, err := s.pyClient.ProcessImage(ctx, imgData, "receipt.jpg")
rawResult, err := s.pyClient.ProcessImage(ctx, imgData, filename)
if err != nil {
draft.Status = drafts.StatusError
_ = s.draftRepo.Update(draft)
return nil, fmt.Errorf("python ocr error: %w", err)
}
// 6. Матчинг и сохранение позиций
// Парсим дату документа
var dateIncoming *time.Time
if rawResult.DocDate != "" {
if t, err := time.Parse("02.01.2006", rawResult.DocDate); err == nil {
dateIncoming = &t
}
}
// Заполняем номер и дату документа
draft.IncomingDocumentNumber = rawResult.DocNumber
draft.DateIncoming = dateIncoming
// 7. Матчинг и сохранение позиций
var draftItems []drafts.DraftInvoiceItem
for _, rawItem := range rawResult.Items {
item := drafts.DraftInvoiceItem{

116
internal/transport/telegram/bot.go
View File

@@ -7,6 +7,7 @@ import (
"fmt"
"io"
"net/http"
"path/filepath"
"strings"
"time"
@@ -36,6 +37,8 @@ type Bot struct {
fsm *StateManager
adminIDs map[int64]struct{}
devIDs map[int64]struct{}
maintenanceMode bool
webAppURL string
menuServers *tele.ReplyMarkup
@@ -51,6 +54,8 @@ func NewBot(
accountRepo account.Repository,
rmsFactory *rms.Factory,
cryptoManager *crypto.CryptoManager,
maintenanceMode bool,
devIDs []int64,
) (*Bot, error) {
pref := tele.Settings{
@@ -71,6 +76,11 @@ func NewBot(
admins[id] = struct{}{}
}
devs := make(map[int64]struct{})
for _, id := range devIDs {
devs[id] = struct{}{}
}
bot := &Bot{
b: b,
ocrService: ocrService,
@@ -81,6 +91,8 @@ func NewBot(
cryptoManager: cryptoManager,
fsm: NewStateManager(),
adminIDs: admins,
devIDs: devs,
maintenanceMode: maintenanceMode,
webAppURL: cfg.WebAppURL,
}
@@ -93,6 +105,11 @@ func NewBot(
return bot, nil
}
func (bot *Bot) isDev(userID int64) bool {
_, ok := bot.devIDs[userID]
return !bot.maintenanceMode || ok
}
func (bot *Bot) initMenus() {
bot.menuServers = &tele.ReplyMarkup{}
@@ -134,6 +151,7 @@ func (bot *Bot) initHandlers() {
bot.b.Handle(tele.OnText, bot.handleText)
bot.b.Handle(tele.OnPhoto, bot.handlePhoto)
bot.b.Handle(tele.OnDocument, bot.handleDocument)
}
func (bot *Bot) Start() {
@@ -160,6 +178,11 @@ func (bot *Bot) handleStartCommand(c tele.Context) error {
return bot.handleInviteLink(c, strings.TrimPrefix(payload, "invite_"))
}
userID := c.Sender().ID
if bot.maintenanceMode && !bot.isDev(userID) {
return c.Send("🛠 **Сервис на техническом обслуживании** Мы проводим плановые работы... 📸 **Вы можете отправить фото чека или накладной** прямо в этот чат — наша команда обработает его и добавит в систему вручную.", tele.ModeHTML)
}
welcomeTxt := "🚀 <b>RMSer — ваш умный ассистент для iiko</b>\n\n" +
"Больше не нужно вводить накладные вручную. Просто сфотографируйте чек, а я сделаю всё остальное.\n\n" +
"<b>Почему это удобно:</b>\n" +
@@ -443,6 +466,10 @@ func (bot *Bot) handleCallback(c tele.Context) error {
}
userDB, _ := bot.accountRepo.GetUserByTelegramID(c.Sender().ID)
userID := c.Sender().ID
if bot.maintenanceMode && !bot.isDev(userID) {
return c.Respond(&tele.CallbackResponse{Text: "Сервис на обслуживании"})
}
// --- INTEGRATION: Billing Callbacks ---
if strings.HasPrefix(data, "bill_") || strings.HasPrefix(data, "buy_id_") || strings.HasPrefix(data, "pay_") {
@@ -713,6 +740,10 @@ func (bot *Bot) handleText(c tele.Context) error {
state := bot.fsm.GetState(userID)
text := strings.TrimSpace(c.Text())
if bot.maintenanceMode && !bot.isDev(userID) {
return c.Send("Сервис на обслуживании", tele.ModeHTML)
}
if strings.ToLower(text) == "отмена" || strings.ToLower(text) == "/cancel" {
bot.fsm.Reset(userID)
return bot.renderMainMenu(c)
@@ -799,6 +830,7 @@ func (bot *Bot) handlePhoto(c tele.Context) error {
if err != nil {
return c.Send("Ошибка базы данных пользователей")
}
userID := c.Sender().ID
_, err = bot.rmsFactory.GetClientForUser(userDB.ID)
if err != nil {
return c.Send("⛔ Ошибка доступа к iiko или сервер не выбран.\nПроверьте статус сервера в меню.")
@@ -821,7 +853,7 @@ func (bot *Bot) handlePhoto(c tele.Context) error {
c.Send("⏳ <b>ИИ анализирует документ...</b>\nОбрабатываю позиции и ищу совпадения в вашей базе iiko.", tele.ModeHTML)
ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
defer cancel()
draft, err := bot.ocrService.ProcessReceiptImage(ctx, userDB.ID, imgData)
draft, err := bot.ocrService.ProcessDocument(ctx, userDB.ID, imgData, "photo.jpg")
if err != nil {
logger.Log.Error("OCR processing failed", zap.Error(err))
return c.Send("❌ Ошибка обработки: " + err.Error())
@@ -832,6 +864,7 @@ func (bot *Bot) handlePhoto(c tele.Context) error {
matchedCount++
}
}
if bot.isDev(userID) {
baseURL := strings.TrimRight(bot.webAppURL, "/")
fullURL := fmt.Sprintf("%s/invoice/%s", baseURL, draft.ID.String())
var msgText string
@@ -849,6 +882,87 @@ func (bot *Bot) handlePhoto(c tele.Context) error {
msgText += "\n\n<i>(Редактирование доступно Администратору)</i>"
}
return c.Send(msgText, menu, tele.ModeHTML)
} else {
return c.Send("✅ **Фото принято!** Мы получили ваш документ и передали его оператору. Спасибо!", tele.ModeHTML)
}
}
func (bot *Bot) handleDocument(c tele.Context) error {
userDB, err := bot.accountRepo.GetOrCreateUser(c.Sender().ID, c.Sender().Username, "", "")
if err != nil {
return c.Send("Ошибка базы данных пользователей")
}
userID := c.Sender().ID
_, err = bot.rmsFactory.GetClientForUser(userDB.ID)
if err != nil {
return c.Send("⛔ Ошибка доступа к iiko или сервер не выбран.\nПроверьте статус сервера в меню.")
}
doc := c.Message().Document
filename := doc.FileName
// Проверяем расширение файла
ext := strings.ToLower(filepath.Ext(filename))
allowedExtensions := map[string]bool{
".jpg": true,
".jpeg": true,
".png": true,
".xlsx": true,
}
if !allowedExtensions[ext] {
return c.Send("❌ Неподдерживаемый формат файла. Пожалуйста, отправьте изображение (.jpg, .jpeg, .png) или Excel файл (.xlsx).")
}
file, err := bot.b.FileByID(doc.FileID)
if err != nil {
return c.Send("Ошибка доступа к файлу.")
}
fileURL := fmt.Sprintf("https://api.telegram.org/file/bot%s/%s", bot.b.Token, file.FilePath)
resp, err := http.Get(fileURL)
if err != nil {
return c.Send("Ошибка скачивания файла.")
}
defer resp.Body.Close()
fileData, err := io.ReadAll(resp.Body)
if err != nil {
return c.Send("Ошибка чтения файла.")
}
c.Send("⏳ <b>ИИ анализирует документ...</b>\nОбрабатываю позиции и ищу совпадения в вашей базе iiko.", tele.ModeHTML)
ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
defer cancel()
draft, err := bot.ocrService.ProcessDocument(ctx, userDB.ID, fileData, filename)
if err != nil {
logger.Log.Error("OCR processing failed", zap.Error(err))
return c.Send("❌ Ошибка обработки: " + err.Error())
}
matchedCount := 0
for _, item := range draft.Items {
if item.IsMatched {
matchedCount++
}
}
if bot.isDev(userID) {
baseURL := strings.TrimRight(bot.webAppURL, "/")
fullURL := fmt.Sprintf("%s/invoice/%s", baseURL, draft.ID.String())
var msgText string
if matchedCount == len(draft.Items) {
msgText = fmt.Sprintf("✅ <b>Все позиции распознаны!</b>\n\nВсе товары найдены в вашей номенклатуре. Проверьте количество (%d) и цену в приложении перед отправкой.", len(draft.Items))
} else {
msgText = fmt.Sprintf("⚠️ <b>Распознано позиций: %d из %d</b>\n\nОстальные товары я вижу впервые. Воспользуйтесь <b>удобным интерфейсом сопоставления</b> в приложении — я запомню ваш выбор навсегда.", matchedCount, len(draft.Items))
}
menu := &tele.ReplyMarkup{}
role, _ := bot.accountRepo.GetUserRole(userDB.ID, draft.RMSServerID)
if role != account.RoleOperator {
btnOpen := menu.WebApp("📝 Открыть и сопоставить", &tele.WebApp{URL: fullURL})
menu.Inline(menu.Row(btnOpen))
} else {
msgText += "\n\n<i>(Редактирование доступно Администратору)</i>"
}
return c.Send(msgText, menu, tele.ModeHTML)
} else {
return c.Send("✅ **Документ принят!** Мы получили ваш документ и передали его оператору. Спасибо!", tele.ModeHTML)
}
}
func (bot *Bot) handleConfirmNameYes(c tele.Context) error {

5
ocr-service/Dockerfile
View File

@@ -2,13 +2,10 @@
FROM python:3.10-slim
# Установка системных зависимостей
# tesseract-ocr + rus: для распознавания текста
# libgl1, libglib2.0-0: для работы OpenCV
# libzbar0: для сканирования QR-кодов
RUN apt-get update && apt-get install -y \
curl \
tesseract-ocr \
tesseract-ocr-rus \
libgl1 \
libglib2.0-0 \
libzbar0 \
@@ -31,4 +28,4 @@ COPY . .
EXPOSE 5000
# Запускаем приложение
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "5000"]
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "5000"]

109
ocr-service/main.py → ocr-service/app/main.py
View File

@@ -8,13 +8,12 @@ import cv2
import numpy as np
# Импортируем модули
from imgproc import preprocess_image
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
from app.schemas.models import ParsedItem, RecognitionResult
from app.services.qr import detect_and_decode_qr, fetch_data_from_api, extract_fiscal_data
# Импортируем новый модуль
from yandex_ocr import yandex_engine
from llm_parser import llm_parser
from app.services.ocr import yandex_engine
from app.services.llm import llm_parser
from app.services.excel import extract_text_from_excel
logging.basicConfig(
level=logging.INFO,
@@ -22,12 +21,7 @@ logging.basicConfig(
)
logger = logging.getLogger(__name__)
app = FastAPI(title="RMSER OCR Service (Hybrid: QR + Yandex + Tesseract)")
class RecognitionResult(BaseModel):
source: str # 'qr_api', 'yandex_vision', 'tesseract_ocr'
items: List[ParsedItem]
raw_text: str = ""
app = FastAPI(title="RMSER OCR Service (Cloud-only: QR + Yandex + GigaChat)")
@app.get("/health")
def health_check():
@@ -37,20 +31,52 @@ def health_check():
async def recognize_receipt(image: UploadFile = File(...)):
"""
Стратегия:
1. QR Code + FNS API (Приоритет 1 - Идеальная точность)
2. Yandex Vision OCR (Приоритет 2 - Высокая точность, если настроен)
3. Tesseract OCR (Приоритет 3 - Локальный фолбэк)
1. Excel файл (.xlsx) -> Извлечение текста -> LLM парсинг
2. QR Code + FNS API (Приоритет 1 - Идеальная точность)
3. Yandex Vision OCR + LLM (Приоритет 2 - Высокая точность, если настроен)
Если ничего не найдено, возвращает пустой результат.
"""
logger.info(f"Received file: {image.filename}, content_type: {image.content_type}")
# Проверка на Excel файл
if image.filename and image.filename.lower().endswith('.xlsx'):
logger.info("Processing Excel file...")
try:
content = await image.read()
excel_text = extract_text_from_excel(content)
if excel_text:
logger.info(f"Excel text extracted, length: {len(excel_text)}")
logger.info("Calling LLM Manager to parse Excel text...")
excel_result = llm_parser.parse_receipt(excel_text)
return RecognitionResult(
source="excel_llm",
items=excel_result["items"],
raw_text=excel_text,
doc_number=excel_result["doc_number"],
doc_date=excel_result["doc_date"]
)
else:
logger.warning("Excel file is empty or contains no text")
return RecognitionResult(
source="none",
items=[],
raw_text=""
)
except Exception as e:
logger.error(f"Error processing Excel file: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"Error processing Excel file: {str(e)}")
# Проверка на изображение
if not image.content_type.startswith("image/"):
raise HTTPException(status_code=400, detail="File must be an image")
raise HTTPException(status_code=400, detail="File must be an image or .xlsx file")
try:
# Читаем сырые байты
content = await image.read()
# Конвертируем в numpy для QR и локального препроцессинга
# Конвертируем в numpy для QR
nparr = np.frombuffer(content, np.uint8)
original_cv_image = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
@@ -78,7 +104,7 @@ async def recognize_receipt(image: UploadFile = File(...)):
logger.info("QR code not found. Proceeding to OCR.")
# --- ЭТАП 2: OCR + Virtual QR Strategy ---
if yandex_engine.oauth_token and yandex_engine.folder_id:
if yandex_engine.is_configured():
logger.info("--- Stage 2: Yandex Vision OCR + Virtual QR ---")
yandex_text = yandex_engine.recognize(content)
@@ -98,41 +124,30 @@ async def recognize_receipt(image: UploadFile = File(...)):
raw_text=yandex_text
)
# Если виртуальный QR не сработал, пробуем Regex
yandex_items = parse_receipt_text(yandex_text)
# Если Regex пуст — вызываем LLM (GigaChat / YandexGPT)
if not yandex_items:
logger.info("Regex found nothing. Calling LLM Manager...")
iam_token = yandex_engine._get_iam_token()
yandex_items = llm_parser.parse_with_priority(yandex_text, iam_token)
# Вызываем LLM для парсинга текста
logger.info("Calling LLM Manager to parse text...")
yandex_result = llm_parser.parse_receipt(yandex_text)
return RecognitionResult(
source="yandex_vision_llm",
items=yandex_items,
raw_text=yandex_text
items=yandex_result["items"],
raw_text=yandex_text,
doc_number=yandex_result["doc_number"],
doc_date=yandex_result["doc_date"]
)
else:
logger.warning("Yandex Vision returned empty text or failed. Falling back to Tesseract.")
else:
logger.info("Yandex Vision credentials not set. Skipping Stage 2.")
# --- ЭТАП 3: Tesseract Strategy (Local Fallback) ---
logger.info("--- Stage 3: Tesseract OCR (Local) ---")
# 1. Image Processing (бинаризация, выравнивание)
processed_img = preprocess_image(content)
# 2. OCR
tesseract_text = ocr_engine.recognize(processed_img)
# 3. Parsing
ocr_items = parse_receipt_text(tesseract_text)
logger.warning("Yandex Vision returned empty text or failed. No fallback available.")
return RecognitionResult(
source="tesseract_ocr",
items=ocr_items,
raw_text=tesseract_text
source="none",
items=[],
raw_text=""
)
else:
logger.info("Yandex Vision credentials not set. No OCR available.")
return RecognitionResult(
source="none",
items=[],
raw_text=""
)
except Exception as e:

15
ocr-service/app/schemas/models.py Normal file
View File

@@ -0,0 +1,15 @@
from typing import List
from pydantic import BaseModel
class ParsedItem(BaseModel):
raw_name: str
amount: float
price: float
sum: float
class RecognitionResult(BaseModel):
source: str # 'qr_api', 'virtual_qr_api', 'yandex_vision_llm', 'none'
items: List[ParsedItem]
raw_text: str = ""
doc_number: str = "" # Номер документа
doc_date: str = "" # Дата документа

68
ocr-service/app/services/auth.py Normal file
View File

@@ -0,0 +1,68 @@
import os
import time
import logging
import requests
from typing import Optional
logger = logging.getLogger(__name__)
IAM_TOKEN_URL = "https://iam.api.cloud.yandex.net/iam/v1/tokens"
class YandexAuthManager:
def __init__(self):
self.oauth_token = os.getenv("YANDEX_OAUTH_TOKEN")
# Кэширование IAM токена
self._iam_token = None
self._token_expire_time = 0
if not self.oauth_token:
logger.warning("YANDEX_OAUTH_TOKEN not set. Yandex services will be unavailable.")
def is_configured(self) -> bool:
return bool(self.oauth_token)
def reset_token(self):
"""Сбрасывает кэшированный токен, заставляя получить новый при следующем вызове."""
self._iam_token = None
self._token_expire_time = 0
def get_iam_token(self) -> Optional[str]:
"""
Получает IAM-токен. Если есть живой кэшированный — возвращает его.
Если нет — обменивает OAuth на IAM.
"""
current_time = time.time()
# Если токен есть и он "свежий" (с запасом в 5 минут)
if self._iam_token and current_time < self._token_expire_time - 300:
return self._iam_token
if not self.oauth_token:
logger.error("OAuth token not available.")
return None
logger.info("Obtaining new IAM token from Yandex...")
try:
response = requests.post(
IAM_TOKEN_URL,
json={"yandexPassportOauthToken": self.oauth_token},
timeout=10
)
response.raise_for_status()
data = response.json()
self._iam_token = data["iamToken"]
# Токен обычно живет 12 часов, но мы будем ориентироваться на поле expiresAt если нужно,
# или просто поставим таймер. Для простоты берем 1 час жизни кэша.
self._token_expire_time = current_time + 3600
logger.info("IAM token received successfully.")
return self._iam_token
except Exception as e:
logger.error(f"Failed to get IAM token: {e}")
return None
# Глобальный инстанс
yandex_auth = YandexAuthManager()

46
ocr-service/app/services/excel.py Normal file
View File

@@ -0,0 +1,46 @@
import io
import logging
import re
from openpyxl import load_workbook
logger = logging.getLogger(__name__)
def extract_text_from_excel(content: bytes) -> str:
"""
Извлекает текстовое содержимое из Excel файла (.xlsx).
Проходит по всем строкам активного листа, собирает значения ячеек
и формирует текстовую строку для передачи в LLM парсер.
Args:
content: Байтовое содержимое Excel файла
Returns:
Строка с текстовым представлением содержимого Excel файла
"""
try:
# Загружаем workbook из байтов, data_only=True берет значения, а не формулы
wb = load_workbook(filename=io.BytesIO(content), data_only=True)
sheet = wb.active
lines = []
for row in sheet.iter_rows(values_only=True):
# Собираем непустые значения в строку через разделитель
row_text = " | ".join([
str(cell).strip()
for cell in row
if cell is not None and str(cell).strip() != ""
])
# Простая эвристика: строка должна содержать хотя бы одну букву (кириллица/латиница) И хотя бы одну цифру.
# Это отсеет пустые разделители и чистые заголовки.
if row_text and re.search(r'[a-zA-Zа-яА-Я]', row_text) and re.search(r'\d', row_text):
lines.append(row_text)
result = "\n".join(lines)
logger.info(f"Extracted {len(lines)} lines from Excel file")
return result
except Exception as e:
logger.error(f"Error extracting text from Excel: {e}", exc_info=True)
raise

237
ocr-service/app/services/llm.py Normal file
View File

@@ -0,0 +1,237 @@
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()

69
ocr-service/yandex_ocr.py → ocr-service/app/services/ocr.py
View File

@@ -1,70 +1,47 @@
from abc import ABC, abstractmethod
from typing import Optional
import os
import time
import json
import base64
import logging
import requests
from typing import Optional
from app.services.auth import yandex_auth
logger = logging.getLogger(__name__)
IAM_TOKEN_URL = "https://iam.api.cloud.yandex.net/iam/v1/tokens"
VISION_URL = "https://ocr.api.cloud.yandex.net/ocr/v1/recognizeText"
class YandexOCREngine:
class OCREngine(ABC):
@abstractmethod
def recognize(self, image_bytes: bytes) -> str:
"""Распознает текст из изображения и возвращает строку."""
pass
@abstractmethod
def is_configured(self) -> bool:
"""Проверяет, настроен ли движок (наличие ключей/настроек)."""
pass
class YandexOCREngine(OCREngine):
def __init__(self):
self.oauth_token = os.getenv("YANDEX_OAUTH_TOKEN")
self.folder_id = os.getenv("YANDEX_FOLDER_ID")
# Кэширование IAM токена
self._iam_token = None
self._token_expire_time = 0
if not self.oauth_token or not self.folder_id:
if not yandex_auth.is_configured() or not self.folder_id:
logger.warning("Yandex OCR credentials (YANDEX_OAUTH_TOKEN, YANDEX_FOLDER_ID) not set. Yandex OCR will be unavailable.")
def _get_iam_token(self) -> Optional[str]:
"""
Получает IAM-токен. Если есть живой кэшированный — возвращает его.
Если нет — обменивает OAuth на IAM.
"""
current_time = time.time()
# Если токен есть и он "свежий" (с запасом в 5 минут)
if self._iam_token and current_time < self._token_expire_time - 300:
return self._iam_token
logger.info("Obtaining new IAM token from Yandex...")
try:
response = requests.post(
IAM_TOKEN_URL,
json={"yandexPassportOauthToken": self.oauth_token},
timeout=10
)
response.raise_for_status()
data = response.json()
self._iam_token = data["iamToken"]
# Токен обычно живет 12 часов, но мы будем ориентироваться на поле expiresAt если нужно,
# или просто поставим таймер. Для простоты берем 1 час жизни кэша.
self._token_expire_time = current_time + 3600
logger.info("IAM token received successfully.")
return self._iam_token
except Exception as e:
logger.error(f"Failed to get IAM token: {e}")
return None
def is_configured(self) -> bool:
return yandex_auth.is_configured() and bool(self.folder_id)
def recognize(self, image_bytes: bytes) -> str:
"""
Отправляет изображение в Yandex Vision и возвращает полный текст.
"""
if not self.oauth_token or not self.folder_id:
if not self.is_configured():
logger.error("Yandex credentials missing.")
return ""
iam_token = self._get_iam_token()
iam_token = yandex_auth.get_iam_token()
if not iam_token:
return ""
@@ -96,8 +73,8 @@ class YandexOCREngine:
# Если 401 Unauthorized, возможно токен протух раньше времени (редко, но бывает)
if response.status_code == 401:
logger.warning("Got 401 from Yandex. Retrying with fresh token...")
self._iam_token = None # сброс кэша
iam_token = self._get_iam_token()
yandex_auth.reset_token() # сброс кэша
iam_token = yandex_auth.get_iam_token()
if iam_token:
headers["Authorization"] = f"Bearer {iam_token}"
response = requests.post(VISION_URL, headers=headers, json=payload, timeout=20)

51
ocr-service/qr_manager.py → ocr-service/app/services/qr.py
View File

@@ -1,18 +1,65 @@
import logging
import requests
import re
from typing import Optional, List
from pyzbar.pyzbar import decode
from PIL import Image
import numpy as np
# Импортируем модель из parser.py
from parser import ParsedItem
from app.schemas.models import ParsedItem
API_TOKEN = "36590.yqtiephCvvkYUKM2W"
API_URL = "https://proverkacheka.com/api/v1/check/get"
logger = logging.getLogger(__name__)
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 is_valid_fiscal_qr(qr_string: str) -> bool:
"""
Проверяет, соответствует ли строка формату фискального чека ФНС.

97
ocr-service/imgproc.py
View File

@@ -1,97 +0,0 @@
import cv2
import numpy as np
import logging
logger = logging.getLogger(__name__)
def order_points(pts):
rect = np.zeros((4, 2), dtype="float32")
s = pts.sum(axis=1)
rect[0] = pts[np.argmin(s)]
rect[2] = pts[np.argmax(s)]
diff = np.diff(pts, axis=1)
rect[1] = pts[np.argmin(diff)]
rect[3] = pts[np.argmax(diff)]
return rect
def four_point_transform(image, pts):
rect = order_points(pts)
(tl, tr, br, bl) = rect
widthA = np.sqrt(((br[0] - bl[0]) ** 2) + ((br[1] - bl[1]) ** 2))
widthB = np.sqrt(((tr[0] - tl[0]) ** 2) + ((tr[1] - tl[1]) ** 2))
maxWidth = max(int(widthA), int(widthB))
heightA = np.sqrt(((tr[0] - br[0]) ** 2) + ((tr[1] - br[1]) ** 2))
heightB = np.sqrt(((tl[0] - bl[0]) ** 2) + ((tl[1] - bl[1]) ** 2))
maxHeight = max(int(heightA), int(heightB))
dst = np.array([
[0, 0],
[maxWidth - 1, 0],
[maxWidth - 1, maxHeight - 1],
[0, maxHeight - 1]], dtype="float32")
M = cv2.getPerspectiveTransform(rect, dst)
warped = cv2.warpPerspective(image, M, (maxWidth, maxHeight))
return warped
def preprocess_image(image_bytes: bytes) -> np.ndarray:
"""
Возвращает БИНАРНОЕ (Ч/Б) изображение для Tesseract.
"""
nparr = np.frombuffer(image_bytes, np.uint8)
image = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
if image is None:
raise ValueError("Could not decode image")
# Ресайз для поиска контуров
ratio = image.shape[0] / 500.0
orig = image.copy()
image_small = cv2.resize(image, (int(image.shape[1] / ratio), 500))
gray = cv2.cvtColor(image_small, cv2.COLOR_BGR2GRAY)
gray = cv2.GaussianBlur(gray, (5, 5), 0)
edged = cv2.Canny(gray, 75, 200)
cnts, _ = cv2.findContours(edged.copy(), cv2.RETR_LIST, cv2.CHAIN_APPROX_SIMPLE)
cnts = sorted(cnts, key=cv2.contourArea, reverse=True)[:5]
screenCnt = None
found = False
for c in cnts:
peri = cv2.arcLength(c, True)
approx = cv2.approxPolyDP(c, 0.02 * peri, True)
if len(approx) == 4:
screenCnt = approx
found = True
break
# Изображение, с которым будем работать дальше
target_img = None
if found:
logger.info("Receipt contour found (Tesseract mode).")
target_img = four_point_transform(orig, screenCnt.reshape(4, 2) * ratio)
else:
logger.warning("Receipt contour NOT found. Using full image.")
target_img = orig
# --- Подготовка для Tesseract (Бинаризация) ---
# Переводим в Gray
gray_final = cv2.cvtColor(target_img, cv2.COLOR_BGR2GRAY)
# Адаптивный порог (превращаем в чисто черное и белое)
# block_size=11, C=2 - классические параметры для текста
thresh = cv2.adaptiveThreshold(
gray_final, 255,
cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
cv2.THRESH_BINARY, 11, 2
)
# Немного убираем шум
# thresh = cv2.medianBlur(thresh, 3)
return thresh

150
ocr-service/llm_parser.py
View File

@@ -1,150 +0,0 @@
import os
import requests
import logging
import json
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")
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"},
"messages": [
{
"role": "system",
"text": (
"Ты — помощник по бухгалтерии. Извлеки список товаров из текста документа. "
"Верни ответ строго в формате JSON: "
'[{"raw_name": string, "amount": float, "price": float, "sum": float}]. '
"Если количество не указано, считай 1.0. Не пиши ничего, кроме JSON."
)
},
{"role": "user", "text": raw_text}
]
}
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {iam_token}",
"x-folder-id": self.folder_id
}
try:
response = requests.post(YANDEX_GPT_URL, headers=headers, json=prompt, timeout=30)
response.raise_for_status()
content = response.json()['result']['alternatives'][0]['message']['text']
clean_json = content.replace("```json", "").replace("```", "").strip()
return [ParsedItem(**item) for item in json.loads(clean_json)]
except Exception as e:
logger.error(f"YandexGPT Parsing error: {e}")
return []
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/ocr.py
View File

@@ -1,38 +0,0 @@
import logging
import pytesseract
from PIL import Image
import numpy as np
logger = logging.getLogger(__name__)
# Если tesseract не в PATH, раскомментируй и укажи путь:
# pytesseract.pytesseract.tesseract_cmd = r'/usr/bin/tesseract'
class OCREngine:
def __init__(self):
logger.info("Initializing Tesseract OCR wrapper...")
# Tesseract не требует загрузки моделей в память,
# проверка версии просто чтобы убедиться, что он установлен
try:
version = pytesseract.get_tesseract_version()
logger.info(f"Tesseract version found: {version}")
except Exception as e:
logger.error("Tesseract not found! Make sure it is installed (apt install tesseract-ocr).")
raise e
def recognize(self, image: np.ndarray) -> str:
"""
Принимает бинарное изображение (numpy array).
"""
# Tesseract работает лучше с PIL Image
pil_img = Image.fromarray(image)
# Конфигурация:
# -l rus+eng: русский и английский
# --psm 6: Assume a single uniform block of text (хорошо для чеков)
custom_config = r'--oem 3 --psm 6'
text = pytesseract.image_to_string(pil_img, lang='rus+eng', config=custom_config)
return text
ocr_engine = OCREngine()

132
ocr-service/parser.py
View File

@@ -1,132 +0,0 @@
import re
from typing import List, Optional
from pydantic import BaseModel
from datetime import datetime
class ParsedItem(BaseModel):
raw_name: str
amount: float
price: float
sum: float
# Регулярка для поиска чисел с плавающей точкой: 123.00, 123,00, 10.5
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 = []
name_buffer = []
for line in lines:
line = line.strip()
if not line:
continue
floats = re.findall(FLOAT_RE, line)
is_price_line = False
if len(floats) >= 2:
is_price_line = True
vals = [parse_float(f) for f in floats]
price = 0.0
amount = 1.0
total = vals[-1]
if len(vals) == 2:
price = vals[0]
amount = 1.0
if total > price and price > 0:
calc_amount = total / price
if abs(round(calc_amount) - calc_amount) < 0.05:
amount = float(round(calc_amount))
elif len(vals) >= 3:
v1, v2 = vals[-3], vals[-2]
if abs(v1 * v2 - total) < 0.5:
price, amount = v1, v2
elif abs(v2 * v1 - total) < 0.5:
price, amount = v2, v1
else:
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,
amount=amount,
price=price,
sum=total
))
name_buffer = []
else:
upper_line = line.upper()
if any(stop in upper_line for stop in ["ИТОГ", "СУММА", "ПРИХОД"]):
name_buffer = []
continue
name_buffer.append(line)
return items

37
ocr-service/python_project_dump.py
View File

File diff suppressed because one or more lines are too long

2
ocr-service/requirements.txt
View File

@@ -4,8 +4,8 @@ python-multipart
pydantic
numpy
opencv-python-headless
pytesseract
requests
pyzbar
pillow
certifi
openpyxl

67
ocr-service/scripts/collect_data_raw.py Normal file
View File

@@ -0,0 +1,67 @@
import os
import requests
import json
import mimetypes
# Папка, куда вы положите фото чеков для теста
INPUT_DIR = "./test_receipts"
# Папка, куда сохраним сырой текст
OUTPUT_DIR = "./raw_outputs"
# Адрес запущенного OCR сервиса
API_URL = "http://10.25.100.250:5006/recognize"
def process_images():
if not os.path.exists(OUTPUT_DIR):
os.makedirs(OUTPUT_DIR)
if not os.path.exists(INPUT_DIR):
os.makedirs(INPUT_DIR)
print(f"Папка {INPUT_DIR} создана. Положите туда фото чеков и перезапустите скрипт.")
return
files = [f for f in os.listdir(INPUT_DIR) if f.lower().endswith(('.jpg', '.jpeg', '.png', '.xlsx'))]
if not files:
print(f"В папке {INPUT_DIR} нет изображений.")
return
print(f"Найдено {len(files)} файлов. Начинаю обработку...")
for filename in files:
file_path = os.path.join(INPUT_DIR, filename)
# Явное определение mime_type для Excel файлов
if filename.lower().endswith('.xlsx'):
mime_type = 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
else:
mime_type, _ = mimetypes.guess_type(file_path)
mime_type = mime_type or 'image/jpeg'
print(f"Processing {filename}...", end=" ")
try:
with open(file_path, 'rb') as f:
files = {'image': (filename, f, mime_type or 'image/jpeg')}
response = requests.post(API_URL, files=files, timeout=30)
if response.status_code == 200:
data = response.json()
raw_text = data.get("raw_text", "")
source = data.get("source", "unknown")
# Сохраняем RAW текст
out_name = f"{filename}_RAW.txt"
with open(os.path.join(OUTPUT_DIR, out_name), "w", encoding="utf-8") as out:
out.write(f"Source: {source}\n")
out.write("="*20 + "\n")
out.write(raw_text)
print(f"OK ({source}) -> {out_name}")
else:
print(f"FAIL: {response.status_code} - {response.text}")
except Exception as e:
print(f"ERROR: {e}")
if __name__ == "__main__":
process_images()

62
ocr-service/scripts/test_parsing_quality.py Normal file
View File

@@ -0,0 +1,62 @@
import os
import requests
import json
import mimetypes
# Папка с фото/excel
INPUT_DIR = "./test_receipts"
# Папка для результатов
OUTPUT_DIR = "./json_results"
# Адрес сервиса
API_URL = "http://10.25.100.250:5006/recognize"
def test_parsing():
if not os.path.exists(OUTPUT_DIR):
os.makedirs(OUTPUT_DIR)
if not os.path.exists(INPUT_DIR):
print(f"Папка {INPUT_DIR} не найдена.")
return
files = [f for f in os.listdir(INPUT_DIR) if f.lower().endswith(('.jpg', '.jpeg', '.png', '.xlsx'))]
print(f"Найдено {len(files)} файлов. Тестируем парсинг...")
for filename in files:
file_path = os.path.join(INPUT_DIR, filename)
# Определение MIME
if filename.lower().endswith('.xlsx'):
mime_type = 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
else:
mime_type, _ = mimetypes.guess_type(file_path)
mime_type = mime_type or 'image/jpeg'
print(f"Processing {filename} ({mime_type})...", end=" ")
try:
with open(file_path, 'rb') as f:
files = {'image': (filename, f, mime_type)}
# Тайм-аут побольше, так как Excel + LLM может быть долгим
response = requests.post(API_URL, files=files, timeout=60)
if response.status_code == 200:
data = response.json()
items = data.get("items", [])
source = data.get("source", "unknown")
doc_number = data.get("doc_number", "")
# Сохраняем JSON
out_name = f"{filename}_RESULT.json"
with open(os.path.join(OUTPUT_DIR, out_name), "w", encoding="utf-8") as out:
json.dump(data, out, ensure_ascii=False, indent=2)
print(f"OK ({source}) -> Found {len(items)} items. Doc#: {doc_number}")
else:
print(f"FAIL: {response.status_code} - {response.text}")
except Exception as e:
print(f"ERROR: {e}")
if __name__ == "__main__":
test_parsing()

1
rmser-view/.gitignore vendored
View File

@@ -23,3 +23,4 @@ dist-ssr
*.sln
*.sw?
*.txt
*.py