Switch OCR to Cyrillic PaddleOCR

CV-Lenta-product-main/pyproject.toml

@@ -11,6 +11,8 @@ requires-python = ">=3.10"
 dependencies = [
     "numpy>=1.24",
     "opencv-python-headless>=4.10",
+    "paddleocr>=3.5.0",
+    "paddlepaddle>=3.3.0",
     "pillow>=10.0",
     "torch>=2.8",
     "torchvision>=0.23",

CV-Lenta-product-main/requirements.txt

@@ -1,5 +1,7 @@
 numpy>=1.24
 opencv-python-headless>=4.10
+paddleocr>=3.5.0
+paddlepaddle>=3.3.0
 pillow>=10.0
 torch>=2.8
 torchvision>=0.23

CV-Lenta-product-main/src/lenta_price_tags/config.py

@@ -62,9 +62,10 @@ class ImageEnhancementConfig:
 
 @dataclass(frozen=True, slots=True)
 class OCRConfig:
-    provider: str = "docscope"
-    model_name: str = "prithivMLmods/docscopeOCR-7B-050425-exp"
-    device: str | None = None
+    provider: str = "paddle"
+    model_name: str = "cyrillic_PP-OCRv5_mobile_rec"
+    docscope_model_name: str = "prithivMLmods/docscopeOCR-7B-050425-exp"
+    device: str | None = "cpu"
     batch_size: int = 16
     temperature: float = 0.0
     max_retries: int = 1
@@ -76,6 +77,13 @@ class OCRConfig:
     precache_model: bool = True
     skip_when_code_has_price_and_identity: bool = True
     fallback_to_mock_on_error: bool = True
+    paddle_text_detection_model_name: str = "PP-OCRv5_mobile_det"
+    paddle_text_det_limit_side_len: int = 960
+    paddle_text_det_thresh: float = 0.30
+    paddle_text_det_box_thresh: float = 0.50
+    paddle_text_det_unclip_ratio: float = 1.8
+    paddle_text_rec_score_thresh: float = 0.0
+    paddle_cache_dir: str | None = "runs/paddlex_cache"
 
 
 @dataclass(frozen=True, slots=True)

CV-Lenta-product-main/src/lenta_price_tags/ocr/docscope_provider.py

@@ -27,6 +27,10 @@ except ImportError:
 class DocscopeOCRProvider:
     def __init__(self, config: OCRConfig | None = None) -> None:
         self.config = config or OCRConfig()
+        if self.config.model_name == "cyrillic_PP-OCRv5_mobile_rec":
+            self.model_name = self.config.docscope_model_name
+        else:
+            self.model_name = self.config.model_name
 
     def recognize(self, image_path: str, prompt: str, metadata: dict | None = None) -> OCREvidence:
         return self.recognize_many([(image_path, prompt, metadata)])[0]
@@ -35,7 +39,7 @@ class DocscopeOCRProvider:
         if not requests:
             return []
         if self.config.precache_model:
-            _ensure_model_snapshot_cached(self.config.model_name)
+            _ensure_model_snapshot_cached(self.model_name)
         output: list[OCREvidence] = []
         batch_size = max(1, int(self.config.batch_size))
         for batch in _chunks(requests, batch_size):
@@ -58,7 +62,7 @@ class DocscopeOCRProvider:
         metadatas = [metadata or {} for _, _, metadata in requests]
         started = time.perf_counter()
         raw_texts, error = _run_docscope_batch_inference(
-            self.config.model_name,
+            self.model_name,
             [image_path for image_path, _, _ in requests],
             [prompt for _, prompt, _ in requests],
             self.config.device if hasattr(self.config, "device") else None,
@@ -84,7 +88,7 @@ class DocscopeOCRProvider:
                     raw_text_lines=lines,
                     parsed_fields=parsed,
                     field_confidence={key: 0.70 for key, value in parsed.items() if key in OCR_FIELD_NAMES and value},
-                    model_name=self.config.model_name,
+                    model_name=self.model_name,
                     latency_ms=per_item_latency,
                     error=final_error,
                 )
@@ -93,7 +97,7 @@ class DocscopeOCRProvider:
 
 
 def preload_docscope_model(model_name: str | None = None, device: str = "cuda") -> None:
-    _load_runner(model_name or OCRConfig().model_name, device)
+    _load_runner(model_name or OCRConfig().docscope_model_name, device)
 
 
 def _run_docscope_batch_inference(

CV-Lenta-product-main/src/lenta_price_tags/ocr/paddle_provider.py

@@ -0,0 +1,354 @@
+from __future__ import annotations
+
+import json
+import os
+import re
+import time
+from decimal import Decimal
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+from lenta_price_tags.config import OCRConfig
+from lenta_price_tags.matching.normalizers import normalize_barcode, normalize_discount, normalize_price, normalize_sku
+from lenta_price_tags.ocr.parser import OCR_FIELD_NAMES
+from lenta_price_tags.schemas import OCREvidence
+
+
+_PRICE_KEYS = ("price_default", "price_card", "price_discount")
+
+
+class PaddleOCRProvider:
+    def __init__(self, config: OCRConfig | None = None) -> None:
+        self.config = config or OCRConfig()
+        self.model_name = self.config.model_name or "cyrillic_PP-OCRv5_mobile_rec"
+
+    def recognize(self, image_path: str, prompt: str, metadata: dict | None = None) -> OCREvidence:
+        return self.recognize_many([(image_path, prompt, metadata)])[0]
+
+    def recognize_many(self, requests: list[tuple[str, str, dict | None]]) -> list[OCREvidence]:
+        if not requests:
+            return []
+        recognizer = _load_paddle_ocr(
+            self.model_name,
+            self.config.paddle_text_detection_model_name,
+            self.config.device or "cpu",
+            self.config.batch_size,
+            self.config.paddle_text_det_limit_side_len,
+            self.config.paddle_text_det_thresh,
+            self.config.paddle_text_det_box_thresh,
+            self.config.paddle_text_det_unclip_ratio,
+            self.config.paddle_text_rec_score_thresh,
+            self.config.paddle_cache_dir,
+        )
+        return self._recognize_batch(recognizer, requests)
+
+    def _recognize_batch(self, recognizer, requests: list[tuple[str, str, dict | None]]) -> list[OCREvidence]:
+        started = time.perf_counter()
+        image_paths = [image_path for image_path, _, _ in requests]
+        try:
+            raw_results = recognizer.predict(image_paths)
+        except Exception:
+            return [self._recognize_one(recognizer, image_path, metadata or {}) for image_path, _, metadata in requests]
+        elapsed_ms = int((time.perf_counter() - started) * 1000)
+        if len(raw_results) != len(requests):
+            return [self._recognize_one(recognizer, image_path, metadata or {}) for image_path, _, metadata in requests]
+        per_item_latency = int(elapsed_ms / max(1, len(requests)))
+        return [
+            self._evidence_from_raw_result(raw_result, metadata or {}, per_item_latency)
+            for raw_result, (_, _, metadata) in zip(raw_results, requests)
+        ]
+
+    def _recognize_one(self, recognizer, image_path: str, metadata: dict[str, Any]) -> OCREvidence:
+        started = time.perf_counter()
+        try:
+            raw_result = recognizer.predict(str(image_path))
+        except Exception as exc:
+            return self._evidence_from_error(metadata, f"paddle_error: {exc}", int((time.perf_counter() - started) * 1000))
+        latency_ms = int((time.perf_counter() - started) * 1000)
+        return self._evidence_from_raw_result(raw_result, metadata, latency_ms)
+
+    def _evidence_from_raw_result(self, raw_result: Any, metadata: dict[str, Any], latency_ms: int) -> OCREvidence:
+        lines, scores = _extract_lines_and_scores(raw_result)
+        fields = parse_paddle_text_lines(lines)
+        error = None if lines else "paddle_no_text"
+        raw_text = json.dumps({"raw_text_lines": lines, **fields}, ensure_ascii=False)
+        confidence = _field_confidence(fields, scores)
+        return OCREvidence(
+            group_id=str(metadata.get("group_id", "")),
+            crop_id=str(metadata.get("crop_id", Path(image_path).stem)),
+            image_variant=str(metadata.get("image_variant", "original")),
+            raw_text=raw_text,
+            raw_text_lines=lines,
+            parsed_fields=fields,
+            field_confidence=confidence,
+            model_name=self.model_name,
+            latency_ms=latency_ms,
+            error=error,
+        )
+
+    def _evidence_from_error(self, metadata: dict[str, Any], error: str, latency_ms: int) -> OCREvidence:
+        return OCREvidence(
+            group_id=str(metadata.get("group_id", "")),
+            crop_id=str(metadata.get("crop_id", "")),
+            image_variant=str(metadata.get("image_variant", "original")),
+            raw_text="",
+            raw_text_lines=[],
+            parsed_fields={},
+            field_confidence={},
+            model_name=self.model_name,
+            latency_ms=latency_ms,
+            error=error,
+        )
+
+
+def parse_paddle_text_lines(lines: list[str]) -> dict[str, str]:
+    clean_lines = [_clean_line(line) for line in lines]
+    clean_lines = [line for line in clean_lines if line]
+    prices = _extract_prices(clean_lines)
+    fields: dict[str, str] = {key: "" for key in OCR_FIELD_NAMES}
+    fields["barcode"] = _extract_barcode(clean_lines)
+    fields["discount_amount"] = _extract_discount(clean_lines)
+    fields["id_sku"] = _extract_sku(clean_lines, fields["barcode"])
+    fields["print_datetime"] = _extract_datetime(clean_lines)
+    if prices:
+        unique_prices = sorted(set(prices), key=lambda value: Decimal(value), reverse=True)
+        fields["price_default"] = unique_prices[0]
+        fields["price_card"] = unique_prices[-1] if len(unique_prices) > 1 else unique_prices[0]
+    fields["product_name"] = _extract_product_name(clean_lines)
+    return {key: value for key, value in fields.items() if value}
+
+
+@lru_cache(maxsize=2)
+def _load_paddle_ocr(
+    text_recognition_model_name: str,
+    text_detection_model_name: str,
+    device: str,
+    text_recognition_batch_size: int,
+    text_det_limit_side_len: int,
+    text_det_thresh: float,
+    text_det_box_thresh: float,
+    text_det_unclip_ratio: float,
+    text_rec_score_thresh: float,
+    cache_dir: str | None,
+):
+    _configure_paddle_cache(cache_dir)
+    from paddleocr import PaddleOCR
+
+    return PaddleOCR(
+        text_detection_model_name=text_detection_model_name,
+        text_recognition_model_name=text_recognition_model_name,
+        text_recognition_batch_size=max(1, int(text_recognition_batch_size)),
+        use_doc_orientation_classify=False,
+        use_doc_unwarping=False,
+        use_textline_orientation=False,
+        text_det_limit_side_len=text_det_limit_side_len,
+        text_det_thresh=text_det_thresh,
+        text_det_box_thresh=text_det_box_thresh,
+        text_det_unclip_ratio=text_det_unclip_ratio,
+        text_rec_score_thresh=text_rec_score_thresh,
+        device=device,
+    )
+
+
+def _configure_paddle_cache(cache_dir: str | None) -> None:
+    if cache_dir:
+        path = Path(cache_dir)
+        if not path.is_absolute():
+            path = Path.cwd() / path
+        path.mkdir(parents=True, exist_ok=True)
+        os.environ.setdefault("PADDLE_PDX_CACHE_HOME", str(path))
+    os.environ.setdefault("PADDLE_PDX_DISABLE_MODEL_SOURCE_CHECK", "True")
+
+
+def _extract_lines_and_scores(raw_result: Any) -> tuple[list[str], list[float]]:
+    lines: list[str] = []
+    scores: list[float] = []
+    for item in _iter_result_items(raw_result):
+        payload = _payload(item)
+        rec_texts = payload.get("rec_texts") or payload.get("texts") or []
+        rec_scores = payload.get("rec_scores") or payload.get("scores") or []
+        for text in rec_texts:
+            clean = _clean_line(text)
+            if clean:
+                lines.append(clean)
+        for score in rec_scores:
+            try:
+                scores.append(float(score))
+            except (TypeError, ValueError):
+                continue
+        if not rec_texts:
+            for text, score in _legacy_lines(item):
+                clean = _clean_line(text)
+                if clean:
+                    lines.append(clean)
+                    scores.append(score)
+    return lines, scores
+
+
+def _iter_result_items(raw_result: Any) -> list[Any]:
+    if raw_result is None:
+        return []
+    if isinstance(raw_result, dict) or any(hasattr(raw_result, attr) for attr in ("to_dict", "dict", "res")):
+        return [raw_result]
+    if isinstance(raw_result, (list, tuple)):
+        if raw_result and _is_legacy_line(raw_result[0]):
+            return [raw_result]
+        return list(raw_result)
+    return [raw_result]
+
+
+def _is_legacy_line(entry: Any) -> bool:
+    if not isinstance(entry, (list, tuple)) or len(entry) < 2:
+        return False
+    candidate = entry[1]
+    return isinstance(candidate, (list, tuple)) and bool(candidate)
+
+
+def _payload(item: Any) -> dict[str, Any]:
+    if isinstance(item, dict):
+        return item.get("res", item)
+    for attr in ("to_dict", "dict"):
+        method = getattr(item, attr, None)
+        if callable(method):
+            try:
+                value = method()
+                if isinstance(value, dict):
+                    return value.get("res", value)
+            except Exception:
+                pass
+    value = getattr(item, "json", None)
+    if isinstance(value, dict):
+        return value.get("res", value)
+    value = getattr(item, "res", None)
+    if isinstance(value, dict):
+        return value
+    return {}
+
+
+def _legacy_lines(item: Any) -> list[tuple[str, float]]:
+    output: list[tuple[str, float]] = []
+    if not isinstance(item, list):
+        return output
+    for entry in item:
+        if not isinstance(entry, (list, tuple)) or len(entry) < 2:
+            continue
+        candidate = entry[1]
+        if isinstance(candidate, (list, tuple)) and candidate:
+            text = str(candidate[0])
+            score = float(candidate[1]) if len(candidate) > 1 else 0.7
+            output.append((text, score))
+    return output
+
+
+def _extract_prices(lines: list[str]) -> list[str]:
+    prices: list[str] = []
+    for line in lines:
+        if "%" in line or _looks_like_barcode(line) or _looks_like_datetime(line):
+            continue
+        explicit = re.findall(r"(?<!\d)(\d{1,5})\s*[,.\-]\s*(\d{2})(?!\d)", line)
+        for rub, kop in explicit:
+            price = normalize_price(f"{rub}.{kop}")
+            if _valid_price(price):
+                prices.append(price)
+        if explicit:
+            continue
+        if _price_like_short_line(line):
+            match = re.search(r"(?<!\d)(\d{1,5})\s+(\d{2})(?!\d)", line)
+            if match:
+                price = normalize_price(f"{match.group(1)}.{match.group(2)}")
+                if _valid_price(price):
+                    prices.append(price)
+    return prices
+
+
+def _extract_barcode(lines: list[str]) -> str:
+    candidates: list[str] = []
+    for line in lines:
+        digits = normalize_barcode(line)
+        if len(digits) in {8, 12, 13, 14}:
+            candidates.append(digits)
+    for preferred_length in (13, 14, 8, 12):
+        for digits in candidates:
+            if len(digits) == preferred_length:
+                return digits
+    return ""
+
+
+def _extract_discount(lines: list[str]) -> str:
+    for line in lines:
+        discount = normalize_discount(line)
+        if discount:
+            return discount
+    return ""
+
+
+def _extract_sku(lines: list[str], barcode: str) -> str:
+    for line in lines:
+        digits = normalize_barcode(line)
+        if not digits or digits == barcode:
+            continue
+        if 9 <= len(digits) <= 12:
+            return normalize_sku(digits)
+    return ""
+
+
+def _extract_datetime(lines: list[str]) -> str:
+    for line in lines:
+        match = re.search(r"\b\d{2}[.\/-]\d{2}[.\/-]\d{4}(?:\s+\d{1,2}:\d{2})?\b", line)
+        if match:
+            return match.group(0)
+    return ""
+
+
+def _extract_product_name(lines: list[str]) -> str:
+    candidates: list[str] = []
+    for line in lines:
+        if _skip_name_line(line):
+            continue
+        if len(re.findall(r"[A-Za-zА-Яа-яЁё]", line)) < 3:
+            continue
+        candidates.append(line)
+    return " ".join(candidates[:3])[:220]
+
+
+def _field_confidence(fields: dict[str, str], scores: list[float]) -> dict[str, float]:
+    base = max(0.35, min(0.99, sum(scores) / len(scores))) if scores else 0.65
+    return {key: base for key, value in fields.items() if key in OCR_FIELD_NAMES and value}
+
+
+def _clean_line(value: Any) -> str:
+    return " ".join(str(value).replace("\u00a0", " ").split())
+
+
+def _valid_price(value: str) -> bool:
+    try:
+        price = Decimal(value)
+    except Exception:
+        return False
+    return Decimal("0.01") <= price <= Decimal("999999.99")
+
+
+def _looks_like_barcode(line: str) -> bool:
+    return len(normalize_barcode(line)) >= 8 and len(re.sub(r"\D", "", line)) / max(1, len(line)) > 0.65
+
+
+def _looks_like_datetime(line: str) -> bool:
+    return bool(re.search(r"\b\d{2}[.\/-]\d{2}[.\/-]\d{4}\b", line))
+
+
+def _price_like_short_line(line: str) -> bool:
+    if len(line) > 24:
+        return False
+    if re.search(r"[A-Za-zА-Яа-яЁё]{4,}", line):
+        return False
+    return bool(re.search(r"(?<!\d)\d{1,5}\s+\d{2}(?!\d)", line))
+
+
+def _skip_name_line(line: str) -> bool:
+    lower = line.casefold()
+    if any(word in lower for word in ("цена", "скид", "руб", "коп", "карта", "штрих", "barcode")):
+        return True
+    if "%" in line or _looks_like_barcode(line) or _looks_like_datetime(line):
+        return True
+    return bool(_extract_prices([line]))

CV-Lenta-product-main/src/lenta_price_tags/pipeline.py

@@ -18,6 +18,7 @@ from lenta_price_tags.cv.quality import CropQualitySelector
 from lenta_price_tags.matching.matcher import ProductMatcher
 from lenta_price_tags.ocr.docscope_provider import DocscopeOCRProvider
 from lenta_price_tags.ocr.mock_provider import MockOCRProvider
+from lenta_price_tags.ocr.paddle_provider import PaddleOCRProvider
 from lenta_price_tags.ocr.prompts import PRICE_TAG_OCR_PROMPT
 from lenta_price_tags.output.csv_writer import CsvWriter, compose_final_row
 from lenta_price_tags.output.debug_writer import DebugWriter
@@ -155,6 +156,10 @@ class PriceTagRecognitionPipeline:
             ocr_config = replace(ocr_config, device=self.config.runtime.device)
         if ocr_config.provider.casefold() == "mock":
             return MockOCRProvider()
+        if ocr_config.provider.casefold() == "docscope":
+            return DocscopeOCRProvider(ocr_config)
+        if ocr_config.provider.casefold() in {"paddle", "paddleocr", "ppocr"}:
+            return PaddleOCRProvider(ocr_config)
         return DocscopeOCRProvider(ocr_config)
 
     def _group_video(self, video: Path, run_dir: Path) -> list[PriceTagGroup]:

CV-Lenta-product-main/src/lenta_price_tags/ui/gradio_app.py

@@ -50,7 +50,8 @@ def build_demo():
                 "image_enhancement": {"variants": ["original"]},
                 "ocr": {
                     "provider": ocr_provider,
-                    "device": "cuda",
+                    "device": "cuda" if ocr_provider == "docscope" else "cpu",
+                    "model_name": "cyrillic_PP-OCRv5_mobile_rec",
                     "batch_size": int(ocr_batch_size),
                     "max_new_tokens": 256,
                     "zerogpu_duration_sec": 180,
@@ -78,7 +79,7 @@ def build_demo():
             video = gr.File(label="Video", file_types=[".mp4", ".mov", ".avi", ".mkv", ".webm"], type="filepath")
             db = gr.File(label="Product DB CSV", file_types=[".csv"], type="filepath")
         with gr.Row():
-            ocr = gr.Radio(["docscope", "mock"], value="docscope", label="OCR")
+            ocr = gr.Radio(["paddle", "docscope", "mock"], value="paddle", label="OCR")
             input_rotate = gr.Dropdown([0, 90, 180, 270], value=270, label="Video rotation")
             ocr_batch_size = gr.Dropdown([1, 2, 4, 8, 16, 32], value=16, label="OCR batch size")
             button = gr.Button("Run", variant="primary")

CV-Lenta-product-main/tests/test_paddle_ocr_parser.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from lenta_price_tags.ocr.paddle_provider import parse_paddle_text_lines
+
+
+def test_parse_paddle_lines_extracts_price_barcode_and_name() -> None:
+    fields = parse_paddle_text_lines(
+        [
+            "Мед ПОТАПЫЧ натуральный липовый",
+            "415,79",
+            "316 99",
+            "-23%",
+            "4603552017456",
+            "370204501518",
+        ]
+    )
+
+    assert fields["product_name"] == "Мед ПОТАПЫЧ натуральный липовый"
+    assert fields["price_default"] == "415.79"
+    assert fields["price_card"] == "316.99"
+    assert fields["discount_amount"] == "-23%"
+    assert fields["barcode"] == "4603552017456"
+    assert fields["id_sku"] == "370204501518"
+
+
+def test_parse_paddle_lines_does_not_turn_sku_into_price() -> None:
+    fields = parse_paddle_text_lines(
+        [
+            "Вино HAUT MARIN Colombard Ugni-blanc",
+            "270 108 726 573",
+            "3760094282559",
+            "1104.99",
+        ]
+    )
+
+    assert fields["price_default"] == "1104.99"
+    assert fields["barcode"] == "3760094282559"
+    assert fields["id_sku"] == "270108726573"

requirements.txt

@@ -2,6 +2,8 @@ accelerate>=0.34
 gradio>=4.44
 numpy>=1.24
 opencv-python-headless>=4.10
+paddleocr>=3.5.0
+paddlepaddle>=3.3.0
 pillow>=10.0
 pydantic>=2.0
 qwen-vl-utils>=0.0.8