src/ocr/paddle_engine.py

# -*- coding: utf-8 -*-
"""
PaddleOCR engine with singleton pattern and memory optimization.
Compatible with both PaddleOCR 2.x and 3.x APIs.
"""
import logging
import gc
import warnings

# Suppress warnings
warnings.filterwarnings('ignore')


class LocalPaddleEngine:
    """
    PaddleOCR engine with singleton pattern.
    Auto-detects PaddleOCR version and uses appropriate API.
    """
    _instance = None
    _initialized = False

    def __new__(cls, lang='korean', enable_paddle=True):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self, lang='korean', enable_paddle=True):
        if LocalPaddleEngine._initialized:
            return

        self.ocr = None
        self.enabled = enable_paddle
        self.lang = lang
        self._api_version = None

        if not enable_paddle:
            logging.info("PaddleOCR disabled by configuration")
            LocalPaddleEngine._initialized = True
            return

        try:
            from paddleocr import PaddleOCR
            import paddleocr
            version = getattr(paddleocr, '__version__', '2.0.0')
            major_version = int(version.split('.')[0])

            if major_version >= 3:
                self._api_version = '3.x'
                logging.info(f"Initializing PaddleOCR {version} (3.x API, lang={lang})...")
                self.ocr = PaddleOCR(
                    lang=lang,
                    use_doc_orientation_classify=False,
                    use_doc_unwarping=False,
                    use_textline_orientation=False,
                    text_detection_model_name='PP-OCRv5_mobile_det',
                    text_det_limit_side_len=1280,
                    text_det_limit_type='max',
                )
            else:
                self._api_version = '2.x'
                logging.info(f"Initializing PaddleOCR {version} (2.x API, lang={lang})...")
                self.ocr = PaddleOCR(
                    lang=lang,
                    use_angle_cls=False,
                    use_gpu=False,
                    det_limit_side_len=1280,
                    det_limit_type='max',
                )

            logging.info(f"PaddleOCR initialized successfully (API: {self._api_version}).")

        except ImportError:
            logging.warning("PaddleOCR not installed. Disabling Paddle engine.")
            self.ocr = None
            self.enabled = False
        except Exception as e:
            logging.error(f"Failed to init PaddleOCR: {e}")
            self.ocr = None
            self.enabled = False
        finally:
            LocalPaddleEngine._initialized = True

    def detect_text(self, image_path):
        """
        Detect text in an image using PaddleOCR.
        Auto-selects API based on installed version.

        Returns:
            dict: {
                'text': str (full combined text),
                'lines': list of (text, confidence) tuples,
                'avg_confidence': float
            }
        """
        if not self.enabled or not self.ocr:
            return {'text': '', 'lines': [], 'avg_confidence': 0.0, 'debug': 'engine disabled'}

        try:
            if self._api_version == '3.x':
                return self._detect_text_v3(image_path)
            else:
                return self._detect_text_v2(image_path)
        except Exception as e:
            import traceback
            logging.error(f"PaddleOCR detection failed: {e}")
            return {'text': '', 'lines': [], 'avg_confidence': 0.0, 'debug': f'exception: {e}\n{traceback.format_exc()}'}

    @staticmethod
    def _poly_to_rect(poly):
        """Convert 4-point polygon to (x1, y1, x2, y2) bounding rectangle."""
        try:
            if isinstance(poly, (list, tuple)) and len(poly) >= 4:
                xs = [p[0] for p in poly]
                ys = [p[1] for p in poly]
                return (min(xs), min(ys), max(xs), max(ys))
        except (TypeError, IndexError):
            pass
        return None

    def _detect_text_v3(self, image_path):
        """PaddleOCR 3.x API using predict(). Handles multiple result formats.
        Returns text + bounding boxes (ocr_results) for coordinate-based extraction."""
        result = self.ocr.predict(image_path)

        if not result:
            logging.warning("PaddleOCR 3.x predict() returned empty result")
            return {'text': '', 'lines': [], 'ocr_results': [], 'avg_confidence': 0.0, 'debug': 'predict() returned empty'}

        # Build debug info for diagnostics
        debug_info = []
        try:
            debug_info.append(f"result type={type(result).__name__}, len={len(result) if hasattr(result, '__len__') else 'N/A'}")
            first = result[0]
            debug_info.append(f"result[0] type={type(first).__name__}")
            if hasattr(first, '__dict__'):
                debug_info.append(f"result[0] attrs={list(first.__dict__.keys())[:10]}")
            if hasattr(first, 'json'):
                json_val = first.json
                if isinstance(json_val, dict):
                    debug_info.append(f"json keys={list(json_val.keys())}")
                    res = json_val.get('res', None)
                    if res is not None:
                        debug_info.append(f"json.res type={type(res).__name__}")
                        if isinstance(res, dict):
                            debug_info.append(f"json.res keys={list(res.keys())}")
                        elif isinstance(res, list) and res:
                            debug_info.append(f"json.res[0] type={type(res[0]).__name__}, len={len(res)}")
                            if isinstance(res[0], dict):
                                debug_info.append(f"json.res[0] keys={list(res[0].keys())}")
                else:
                    debug_info.append(f"json type={type(json_val).__name__}")
            first_str = str(first)[:300]
            debug_info.append(f"str(result[0])[:300]={first_str}")
        except Exception as e:
            debug_info.append(f"debug error: {e}")

        logging.info(f"PaddleOCR 3.x debug: {'; '.join(debug_info)}")

        texts = []
        scores = []
        bboxes = []  # parallel list of (x1,y1,x2,y2) or None

        # Strategy 1: result[0].json['res'] with rec_texts/rec_scores/dt_polys (PaddleOCR 3.0-3.2)
        try:
            ocr_result = result[0]
            if hasattr(ocr_result, 'json') and isinstance(ocr_result.json, dict):
                res = ocr_result.json.get('res', {})
                if isinstance(res, dict):
                    t = res.get('rec_texts', [])
                    s = res.get('rec_scores', [])
                    polys = res.get('dt_polys', res.get('det_boxes', []))
                    if t:
                        texts, scores = list(t), list(s)
                        for i in range(len(texts)):
                            poly = polys[i] if i < len(polys) else None
                            bboxes.append(self._poly_to_rect(poly) if poly is not None else None)
                        logging.info(f"Parsed with Strategy 1 (json.res): {len(texts)} lines, {sum(1 for b in bboxes if b)} bboxes")
        except Exception as e:
            logging.debug(f"Strategy 1 failed: {e}")

        # Strategy 2: result[0].json['res'] is a list of dicts with 'rec_text'/'rec_score'/'dt_poly'
        if not texts:
            try:
                ocr_result = result[0]
                if hasattr(ocr_result, 'json') and isinstance(ocr_result.json, dict):
                    res = ocr_result.json.get('res', [])
                    if isinstance(res, list):
                        for item in res:
                            if isinstance(item, dict):
                                t = item.get('rec_text', item.get('text', ''))
                                s = item.get('rec_score', item.get('score', item.get('confidence', 0.0)))
                                poly = item.get('dt_poly', item.get('det_box', item.get('bbox', None)))
                                if t:
                                    texts.append(t)
                                    scores.append(float(s))
                                    bboxes.append(self._poly_to_rect(poly) if poly is not None else None)
                        if texts:
                            logging.info(f"Parsed with Strategy 2 (json.res list): {len(texts)} lines, {sum(1 for b in bboxes if b)} bboxes")
            except Exception as e:
                logging.debug(f"Strategy 2 failed: {e}")

        # Strategy 3: result is list of (bbox, (text, confidence)) tuples (PaddleOCR 3.3+)
        if not texts:
            try:
                for item in result:
                    if isinstance(item, (list, tuple)) and len(item) >= 2:
                        bbox_points = item[0]
                        text_info = item[1]
                        if isinstance(text_info, (list, tuple)) and len(text_info) >= 2:
                            texts.append(str(text_info[0]))
                            scores.append(float(text_info[1]))
                            bboxes.append(self._poly_to_rect(bbox_points))
                if texts:
                    logging.info(f"Parsed with Strategy 3 (bbox tuples): {len(texts)} lines, {sum(1 for b in bboxes if b)} bboxes")
            except Exception as e:
                logging.debug(f"Strategy 3 failed: {e}")

        # Strategy 4: result is generator/iterable of result objects with 'rec' attribute
        if not texts:
            try:
                for item in result:
                    if hasattr(item, 'rec'):
                        for rec in item.rec:
                            t = getattr(rec, 'text', '') or (rec[0] if isinstance(rec, (list, tuple)) else '')
                            s = getattr(rec, 'score', 0.0) or (rec[1] if isinstance(rec, (list, tuple)) and len(rec) > 1 else 0.0)
                            if t:
                                texts.append(str(t))
                                scores.append(float(s))
                                bboxes.append(None)
                if texts:
                    logging.info(f"Parsed with Strategy 4 (rec attr): {len(texts)} lines")
            except Exception as e:
                logging.debug(f"Strategy 4 failed: {e}")

        # All strategies failed
        if not texts:
            debug_str = '; '.join(debug_info)
            logging.warning(f"All parsing strategies failed. Debug: {debug_str}")
            return {'text': '', 'lines': [], 'ocr_results': [], 'avg_confidence': 0.0, 'debug': debug_str}

        full_text = '\n'.join(texts)
        lines = list(zip(texts, scores))
        avg_conf = sum(scores) / len(scores) if scores else 0.0

        # Build ocr_results with bbox for coordinate-based extraction
        ocr_results = []
        for i, (t, s) in enumerate(zip(texts, scores)):
            bbox = bboxes[i] if i < len(bboxes) else None
            if bbox:
                ocr_results.append({'text': t, 'confidence': s, 'bbox': bbox})

        logging.info(f"OCR extracted {len(texts)} text lines, {len(ocr_results)} with bbox, avg confidence: {avg_conf:.3f}")
        return {'text': full_text, 'lines': lines, 'ocr_results': ocr_results, 'avg_confidence': avg_conf}

    def _detect_text_v2(self, image_path):
        """PaddleOCR 2.x API using ocr(). Returns text + bounding boxes."""
        result = self.ocr.ocr(image_path, cls=False)

        if not result or not result[0]:
            return {'text': '', 'lines': [], 'ocr_results': [], 'avg_confidence': 0.0}

        texts = []
        scores = []
        ocr_results = []  # (text, confidence, bbox) for coordinate-based extraction
        for line in result[0]:
            bbox_points = line[0]  # [[x1,y1],[x2,y2],[x3,y3],[x4,y4]]
            text, score = line[1]
            texts.append(text)
            scores.append(score)

            # Convert 4-point bbox to (x1, y1, x2, y2) rectangle
            xs = [p[0] for p in bbox_points]
            ys = [p[1] for p in bbox_points]
            bbox = (min(xs), min(ys), max(xs), max(ys))
            ocr_results.append({
                'text': text,
                'confidence': score,
                'bbox': bbox,  # (x1, y1, x2, y2)
            })

        if not texts:
            return {'text': '', 'lines': [], 'ocr_results': [], 'avg_confidence': 0.0}

        full_text = '\n'.join(texts)
        lines = list(zip(texts, scores))
        avg_conf = sum(scores) / len(scores) if scores else 0.0

        return {'text': full_text, 'lines': lines, 'ocr_results': ocr_results, 'avg_confidence': avg_conf}

    @classmethod
    def cleanup(cls):
        """Cleanup PaddleOCR resources."""
        if cls._instance and cls._instance.ocr:
            del cls._instance.ocr
            cls._instance.ocr = None
        cls._instance = None
        cls._initialized = False
        gc.collect()