core/english.py

"""
English vocabulary loader for SinCode v3.
Used for English passthrough detection in the decoder.
Loads purely from the 20k corpus file — no hardcoded word lists.
"""

import os
import logging
import requests
from typing import Set

from core.constants import ENGLISH_CORPUS_URL, MIN_ENGLISH_LEN

logger = logging.getLogger(__name__)


def _resolve_english_cache_path() -> str:
    override = os.getenv("SINCODE_ENGLISH_CACHE")
    if override:
        return override

    candidates = [
        os.path.join(os.getenv("HF_HOME", ""), "english_20k.txt") if os.getenv("HF_HOME") else "",
        os.path.join(os.getcwd(), "english_20k.txt"),
        os.path.join(os.getenv("TMPDIR", os.getenv("TEMP", "/tmp")), "english_20k.txt"),
    ]

    for path in candidates:
        if not path:
            continue
        parent = os.path.dirname(path) or "."
        try:
            os.makedirs(parent, exist_ok=True)
            with open(path, "a", encoding="utf-8"):
                pass
            return path
        except OSError:
            continue

    return "english_20k.txt"


ENGLISH_CORPUS_CACHE = _resolve_english_cache_path()


def load_english_vocab() -> Set[str]:
    vocab: Set[str] = set()

    if not os.path.exists(ENGLISH_CORPUS_CACHE) or os.path.getsize(ENGLISH_CORPUS_CACHE) == 0:
        try:
            logger.info("Downloading English corpus...")
            response = requests.get(ENGLISH_CORPUS_URL, timeout=10)
            response.raise_for_status()
            with open(ENGLISH_CORPUS_CACHE, "wb") as f:
                f.write(response.content)
        except (requests.RequestException, OSError) as exc:
            logger.warning("Could not download English corpus: %s", exc)
            return vocab

    try:
        with open(ENGLISH_CORPUS_CACHE, "r", encoding="utf-8") as f:
            vocab.update(
                w for line in f
                if (w := line.strip().lower()) and len(w) >= MIN_ENGLISH_LEN
            )
    except OSError as exc:
        logger.warning("Could not read English corpus file: %s", exc)

    logger.info("English vocabulary loaded: %d words", len(vocab))
    return vocab


ENGLISH_VOCAB: Set[str] = load_english_vocab()