wordloop/backend/services/word_scan_service.py

from __future__ import annotations

import re
import shutil
import subprocess
import tempfile
from pathlib import Path

from fastapi import HTTPException

from services.llm_service import LlmServiceError, llm_service


_CJK_RE = re.compile(r"[\u4e00-\u9fff]")
_LATIN_RE = re.compile(r"[A-Za-z]")


def _heuristic_parse_line(line: str) -> dict[str, str] | None:
    text = line.strip()
    if not text or len(text) < 2:
        return None
    if not _CJK_RE.search(text) or not _LATIN_RE.search(text):
        return None

    parts = re.split(r"[\t|｜/／:：\-—–]+", text)
    parts = [p.strip() for p in parts if p.strip()]
    if len(parts) < 2:
        parts = re.split(r"\s{2,}", text)
        parts = [p.strip() for p in parts if p.strip()]
    if len(parts) < 2:
        return None

    left, right = parts[0], parts[1]
    left_has_cjk = bool(_CJK_RE.search(left))
    right_has_cjk = bool(_CJK_RE.search(right))
    if left_has_cjk and not right_has_cjk:
        return {"en": right, "zh": left}
    if right_has_cjk and not left_has_cjk:
        return {"en": left, "zh": right}
    return None


def _heuristic_parse_text(text: str) -> list[dict[str, str]]:
    pairs: list[dict[str, str]] = []
    seen: set[tuple[str, str]] = set()
    for line in text.splitlines():
        parsed = _heuristic_parse_line(line)
        if not parsed:
            continue
        key = (parsed["en"].lower(), parsed["zh"])
        if key in seen:
            continue
        seen.add(key)
        pairs.append(parsed)
    return pairs


def _to_word_create_items(pairs: list[dict[str, str]]) -> list[dict[str, str]]:
    items: list[dict[str, str]] = []
    for pair in pairs:
        en = pair["en"].strip()
        zh = pair["zh"].strip()
        if not en or not zh:
            continue
        items.append(
            {
                "source_text": en,
                "target_text": zh,
                "source_lang": "en",
                "target_lang": "zh",
            }
        )
    return items


class WordScanService:
    def _tesseract_bin(self) -> str | None:
        return shutil.which("tesseract")

    def ocr_image_bytes(self, image_bytes: bytes, *, suffix: str = ".png") -> str:
        tesseract = self._tesseract_bin()
        if not tesseract:
            raise HTTPException(
                status_code=503,
                detail="服务器未安装 OCR（tesseract），暂无法识别照片",
            )
        if not image_bytes:
            raise HTTPException(status_code=400, detail="图片为空")

        safe_suffix = suffix if suffix.startswith(".") else ".png"
        if safe_suffix.lower() not in {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif"}:
            safe_suffix = ".png"

        with tempfile.TemporaryDirectory() as tmp:
            image_path = Path(tmp) / f"scan{safe_suffix}"
            image_path.write_bytes(image_bytes)
            try:
                result = subprocess.run(
                    [tesseract, str(image_path), "stdout", "-l", "chi_sim+eng"],
                    capture_output=True,
                    timeout=60,
                    check=False,
                )
            except subprocess.TimeoutExpired as exc:
                raise HTTPException(status_code=504, detail="OCR 识别超时，请换一张更小的照片") from exc

            if result.returncode != 0:
                detail = result.stderr.decode("utf-8", errors="replace").strip()
                raise HTTPException(status_code=422, detail=detail or "OCR 识别失败")

            text = result.stdout.decode("utf-8", errors="replace").strip()
            if not text:
                raise HTTPException(status_code=422, detail="照片中未识别到文字")
            return text

    def scan_image(self, image_bytes: bytes, *, suffix: str = ".png") -> list[dict[str, str]]:
        text = self.ocr_image_bytes(image_bytes, suffix=suffix)
        return self.parse_scan_text(text)

    def parse_scan_text(self, text: str) -> list[dict[str, str]]:
        cleaned = text.strip()
        if not cleaned:
            raise HTTPException(status_code=400, detail="识别文本为空")

        pairs: list[dict[str, str]] = []
        if llm_service.is_configured():
            try:
                pairs = llm_service.parse_word_pairs_from_text(cleaned)
            except LlmServiceError:
                pairs = []

        if not pairs:
            pairs = _heuristic_parse_text(cleaned)

        items = _to_word_create_items(pairs)
        if not items:
            raise HTTPException(
                status_code=422,
                detail="未能从照片中识别出单词，请换一张更清晰的照片或手动添加",
            )
        return items


word_scan_service = WordScanService()