"""
Domain Term Extractor for RAG Optimization.

Extracts domain-specific terms from text chunks and Excel records to enable
efficient pre-filtering during vector similarity search.
"""

import asyncio
import json
import math
import os
import re
from dataclasses import dataclass, field
from typing import Any, Optional, Union
from uuid import UUID, uuid4

import structlog
from openai import AzureOpenAI, AsyncAzureOpenAI

from src.core.config import settings
from src.llm import get_async_chat_client, get_chat_client, get_chat_model_name, strip_thinking

logger = structlog.get_logger(__name__)

# Environment configuration
MAX_TERMS_PER_CHUNK = int(os.getenv("DOMAIN_TERM_MAX_PER_CHUNK", "50"))
LLM_THRESHOLD = int(os.getenv("DOMAIN_TERM_LLM_THRESHOLD", "2"))
ENABLE_CJK_SEGMENTATION = os.getenv("ENABLE_CJK_SEGMENTATION", "true").lower() == "true"
ENABLE_PERFORMANCE_MONITORING = os.getenv("ENABLE_PERFORMANCE_MONITORING", "false").lower() == "true"

# TEXTIQ-389 confidence constants (SYS-FR-47).
RULE_CONFIDENCE = 0.95
LLM_FALLBACK_CONFIDENCE = 0.70


@dataclass
class DomainTaxonomy:
    """Domain taxonomy with categories and term patterns.

    Defines high-level domain categories and their associated terms for
    rule-based extraction and category inference.
    """

    categories: dict[str, list[str]] = field(default_factory=lambda: {
        "finance": [
            "revenue", "ebitda", "margin", "profit", "loss", "income", "expense",
            "balance", "asset", "liability", "equity", "cash flow", "investment",
            "dividend", "earnings", "debt", "credit", "budget", "forecast", "valuation",
            "roi", "quarter", "quarterly", "annual", "fiscal", "financial"
        ],
        "hr": [
            "employee", "salary", "compensation", "benefit", "payroll", "hire",
            "termination", "performance", "review", "training", "onboarding",
            "headcount", "workforce", "talent", "recruitment", "retention",
            "leave", "vacation", "pto", "overtime", "bonus", "promotion"
        ],
        "engineering": [
            "code", "deploy", "build", "test", "debug", "bug", "feature", "api",
            "database", "server", "client", "frontend", "backend", "infrastructure",
            "architecture", "design", "implementation", "sprint", "ticket", "release",
            "version", "commit", "branch", "merge", "pull request", "ci/cd"
        ],
        "legal": [
            "contract", "agreement", "clause", "term", "condition", "compliance",
            "regulation", "law", "statute", "liability", "indemnity", "warranty",
            "intellectual property", "patent", "trademark", "copyright", "license",
            "litigation", "dispute", "settlement", "amendment", "exhibit"
        ],
        "operations": [
            "process", "workflow", "procedure", "supply chain", "inventory",
            "logistics", "vendor", "supplier", "procurement", "fulfillment",
            "delivery", "shipment", "warehouse", "distribution", "capacity",
            "throughput", "efficiency", "quality", "sla", "kpi"
        ],
        "marketing": [
            "campaign", "brand", "customer", "lead", "conversion", "engagement",
            "analytics", "traffic", "channel", "social media", "content", "seo",
            "sem", "advertising", "promotion", "funnel", "segment", "target",
            "awareness", "retention", "churn"
        ]
    })

    def get_all_terms(self) -> set[str]:
        """Get all terms across all categories."""
        all_terms = set()
        for terms in self.categories.values():
            all_terms.update(term.lower() for term in terms)
        return all_terms

    def infer_category(self, terms: list[str]) -> str:
        """Infer domain category from extracted terms.

        Args:
            terms: List of extracted terms

        Returns:
            Category name with highest term match count, or "general" if no clear match
        """
        if not terms:
            return "general"

        terms_lower = {term.lower() for term in terms}
        category_scores = {}

        for category, category_terms in self.categories.items():
            category_terms_lower = {term.lower() for term in category_terms}
            matches = len(terms_lower & category_terms_lower)
            if matches > 0:
                category_scores[category] = matches

        if not category_scores:
            return "general"

        # Return category with highest score
        return max(category_scores.items(), key=lambda x: x[1])[0]


# LLM prompt for term extraction
TERM_EXTRACTION_PROMPT = """Extract domain-specific terms from the following text.

Focus on:
- Business domain keywords (finance, HR, legal, engineering, etc.)
- Technical terminology specific to the content
- Important entities and concepts
- Industry-specific jargon

Rules:
- Extract 3-15 terms (or fewer if content is sparse)
- Use lowercase
- Include both single words and short phrases (max 3 words)
- Avoid generic words like "the", "is", "and"
- Focus on nouns and noun phrases

Text:
{content}

Output ONLY a JSON array of terms:
["term1", "term2", "term3"]
"""


class DomainTermExtractor:
    """Extracts domain-specific terms from chunks for RAG pre-filtering."""

    def __init__(
        self,
        taxonomy: Optional[DomainTaxonomy] = None,
        client: Optional[AzureOpenAI] = None
    ):
        """Initialize the domain term extractor.

        Args:
            taxonomy: DomainTaxonomy instance (uses default if None)
            client: Optional AzureOpenAI client (lazy-loaded if None)
        """
        self.taxonomy = taxonomy or DomainTaxonomy()
        self._client = client
        self._client_initialized = client is not None
        self.logger = structlog.get_logger(__name__)
        self._sudachi_tokenizer = None
        self._sudachi_init_attempted = False

    def _get_client(self):
        """Get or create LLM client lazily.

        Lazy initialization prevents creating HTTP clients before process forking,
        which causes issues in Celery/Airflow workers.
        """
        if not self._client_initialized:
            self._client = get_chat_client()
            self._client_initialized = True

        return self._client

    def extract_from_chunk(
        self,
        content: str,
        metadata: Optional[dict[str, Any]] = None
    ) -> dict[str, Any]:
        """Extract domain terms from a text chunk.

        Uses hybrid approach:
        1. Rule-based extraction (fast keyword matching)
        2. LLM enhancement (for complex/ambiguous content)

        Args:
            content: Text content of the chunk
            metadata: Optional metadata dict (e.g., {"page_number": 1})

        Returns:
            Dict with:
                - domain_terms: List of extracted terms (max MAX_TERMS_PER_CHUNK)
                - domain_category: Inferred category string
                - extraction_method: "rules" or "llm"

        Example:
            >>> extractor = DomainTermExtractor()
            >>> result = extractor.extract_from_chunk("Q4 EBITDA increased...")
            >>> result
            {"domain_terms": ["ebitda", "q4", "revenue"],
             "domain_category": "finance",
             "extraction_method": "rules"}
        """
        if not content or not content.strip():
            return {
                "domain_terms": [],
                "domain_category": "general",
                "extraction_method": "none"
            }

        # Try rule-based extraction first
        rule_terms = self._extract_by_rules(content)
        terms: list[str] = list(rule_terms)
        confidences: dict[str, float] = {t: RULE_CONFIDENCE for t in rule_terms}
        extraction_method = "rules"

        # Fall back to LLM if too few terms found
        if len(terms) < LLM_THRESHOLD:
            self.logger.info(
                "using_llm_extraction",
                terms_found=len(terms),
                threshold=LLM_THRESHOLD
            )
            for llm_term, llm_conf in self._llm_extract(content):
                terms.append(llm_term)
                # Max across sources (rule=0.95 vs LLM-computed).
                confidences[llm_term] = max(confidences.get(llm_term, 0.0), llm_conf)
            extraction_method = "llm"

        # Deduplicate (preserving order) and limit
        terms = list(dict.fromkeys(terms))[:MAX_TERMS_PER_CHUNK]

        # Infer category
        category = self.taxonomy.infer_category(terms)

        return {
            "domain_terms": terms,
            "domain_category": category,
            "domain_confidences": {t: confidences[t] for t in terms},
            "extraction_method": extraction_method,
        }

    def extract_from_excel_record(
        self,
        headers: list[str],
        row_data: dict[str, Any],
        sheet_name: Optional[str] = None
    ) -> dict[str, Any]:
        """Extract domain terms from an Excel record.

        Extracts terms from both column headers and cell values.

        Args:
            headers: List of column header strings (flat or hierarchical)
            row_data: Dict mapping headers to cell values
            sheet_name: Optional sheet name

        Returns:
            Dict with domain_terms, domain_category, extraction_method

        Example:
            >>> extractor = DomainTermExtractor()
            >>> result = extractor.extract_from_excel_record(
            ...     headers=["Employee ID", "Salary", "Department"],
            ...     row_data={"Employee ID": "E123", "Salary": 75000, "Department": "HR"}
            ... )
            >>> result["domain_category"]
            "hr"
        """
        # Combine headers and values into searchable text
        content_parts = []

        # Add sheet name if available
        if sheet_name:
            content_parts.append(sheet_name)

        # Add all headers
        content_parts.extend(headers)

        # Add string values (skip numeric values for term extraction)
        for value in row_data.values():
            if isinstance(value, str) and value.strip():
                content_parts.append(value)

        combined_content = " ".join(content_parts)

        # Use standard chunk extraction
        return self.extract_from_chunk(combined_content)

    def _extract_by_rules(self, content: str) -> list[str]:
        """Extract terms using rule-based keyword matching.

        Fast pattern matching against taxonomy terms using fuzzy matching.

        Args:
            content: Text content

        Returns:
            List of matched terms
        """
        content_lower = content.lower()
        matched_terms = []

        # Match against all taxonomy terms
        for term in self.taxonomy.get_all_terms():
            # Use word boundary matching to avoid partial matches
            pattern = r'\b' + re.escape(term) + r'\b'
            if re.search(pattern, content_lower):
                matched_terms.append(term)

        return matched_terms

    def _get_sudachi_tokenizer(self):
        """Get or create SudachiPy tokenizer lazily.

        Lazy initialization prevents creating tokenizer before process forking,
        which causes issues in Celery/Airflow workers.

        The _sudachi_init_attempted flag tracks whether initialization has been
        attempted (regardless of success) to avoid repeated failed attempts.

        Returns:
            Tokenizer instance or None if initialization fails
        """
        if not self._sudachi_init_attempted:
            try:
                from sudachipy import tokenizer, dictionary

                tokenizer_obj = dictionary.Dictionary().create()
                self._sudachi_tokenizer = tokenizer_obj
                self._sudachi_init_attempted = True

            except ImportError as e:
                self.logger.warning(
                    "sudachi_init_failed",
                    error="sudachipy not installed",
                    error_type="ImportError",
                    fallback="segmentation_disabled",
                    action_required="install sudachipy and sudachidict_core packages"
                )
                self._sudachi_init_attempted = True
            except (FileNotFoundError, PermissionError) as e:
                self.logger.error(
                    "sudachi_init_failed",
                    error=str(e),
                    error_type=type(e).__name__,
                    fallback="segmentation_disabled",
                    action_required="check dictionary files exist and have correct permissions"
                )
                self._sudachi_init_attempted = True
            except Exception as e:
                self.logger.error(
                    "sudachi_init_failed",
                    error=str(e),
                    error_type=type(e).__name__,
                    fallback="segmentation_disabled",
                    action_required="check logs and sudachipy installation"
                )
                self._sudachi_init_attempted = True

        return self._sudachi_tokenizer

    def _is_japanese(self, text: str) -> bool:
        """Detect if text contains Japanese characters.

        Uses heuristics to distinguish Japanese from Chinese:
        - Hiragana/Katakana presence strongly indicates Japanese
        - Kanji-only text with sufficient Hiragana/Katakana ratio is Japanese
        - Pure Kanji is treated conservatively (could be Chinese)

        Args:
            text: Text to analyze

        Returns:
            True if Japanese characters detected with high confidence, False otherwise
        """
        if not text:
            return False

        # Unicode ranges for Japanese scripts
        # Hiragana: U+3040-U+309F (unique to Japanese)
        # Katakana: U+30A0-U+30FF (unique to Japanese)
        # Kanji: U+4E00-U+9FFF (shared with Chinese)

        # Check for Hiragana or Katakana - strong indicators of Japanese
        hiragana_katakana_pattern = r'[\u3040-\u309F\u30A0-\u30FF]'
        if re.search(hiragana_katakana_pattern, text):
            return True

        # If only Kanji present, check if ratio is significant
        # to avoid false positives on Chinese text
        kanji_pattern = r'[\u4E00-\u9FFF]'
        kanji_matches = re.findall(kanji_pattern, text)
        if kanji_matches:
            # Only treat as Japanese if Kanji makes up >30% of text
            # This helps filter out English text with occasional Kanji
            kanji_ratio = len(kanji_matches) / len(text)
            return kanji_ratio > 0.3

        return False

    def _segment_japanese(self, text: str) -> str:
        """Segment Japanese text using SudachiPy.

        Args:
            text: Japanese text to segment

        Returns:
            Space-separated segmented text, or original text if segmentation fails
        """
        tokenizer_obj = self._get_sudachi_tokenizer()

        if tokenizer_obj is None:
            return text  # Fallback to original if tokenizer unavailable

        try:
            # Use Mode C (long unit) to preserve compound words
            # Mode is specified here during tokenization, not at tokenizer creation
            from sudachipy.tokenizer import Tokenizer
            mode = Tokenizer.SplitMode.C

            # Measure segmentation performance (if enabled)
            if ENABLE_PERFORMANCE_MONITORING:
                import time
                start_time = time.time()

            tokens = [m.surface() for m in tokenizer_obj.tokenize(text, mode)]

            if ENABLE_PERFORMANCE_MONITORING:
                elapsed_ms = (time.time() - start_time) * 1000

            # Validate segmentation output
            if not tokens or all(len(t.strip()) == 0 for t in tokens):
                self.logger.warning(
                    "japanese_segmentation_invalid",
                    reason="empty_tokens",
                    fallback="original_text"
                )
                return text

            # Check if segmentation is meaningful (not all single-char splits)
            single_char_count = sum(1 for t in tokens if len(t.strip()) == 1)
            if len(tokens) > 5 and single_char_count / len(tokens) > 0.9:
                self.logger.warning(
                    "japanese_segmentation_invalid",
                    reason="excessive_single_char_splits",
                    single_char_ratio=round(single_char_count / len(tokens), 2),
                    fallback="original_text"
                )
                return text

            segmented = " ".join(tokens)

            # Log success with optional performance metrics
            log_data = {
                "char_count": len(text),
                "token_count": len(tokens)
            }
            if ENABLE_PERFORMANCE_MONITORING:
                log_data["elapsed_ms"] = round(elapsed_ms, 2)

            self.logger.info("japanese_segmentation_success", **log_data)

            return segmented

        except Exception as e:
            self.logger.warning(
                "japanese_segmentation_failed",
                error=str(e),
                error_type=type(e).__name__,
                fallback="original_text",
                action_required="check if error is transient (retry) or indicates tokenizer issue"
            )
            return text  # Graceful fallback

    def _llm_extract(self, content: str) -> list[tuple[str, float]]:
        """Extract terms using LLM for complex/ambiguous content.

        Returns (term, confidence) pairs. Confidence is `exp(min(token_logprobs))`
        over the tokens spanning each JSON-quoted term in the response, with
        ``LLM_FALLBACK_CONFIDENCE`` (0.70) on any alignment failure.
        """
        import time

        result_text = ""
        try:
            client = self._get_client()

            # Truncate content if too long (max ~2000 tokens)
            # Do this BEFORE segmentation to avoid expansion issues
            if len(content) > 8000:
                content = content[:8000] + "..."

            # PREPROCESSING: Segment Japanese text if enabled
            if ENABLE_CJK_SEGMENTATION:
                if self._is_japanese(content):
                    self.logger.info(
                        "japanese_text_detected",
                        char_count=len(content)
                    )
                    content = self._segment_japanese(content)
            else:
                self.logger.debug("cjk_segmentation_disabled", reason="env_var_false")

            prompt = TERM_EXTRACTION_PROMPT.format(content=content)

            # Add rate limiting: small delay between API calls to avoid overwhelming the service
            time.sleep(0.1)  # 100ms delay between calls

            start_time = time.time()
            response = client.chat.completions.create(
                model=get_chat_model_name(),
                messages=[
                    {"role": "system", "content": "You are a domain term extraction expert."},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.0,
                max_tokens=500,
                logprobs=True,
                timeout=30.0  # 30 second timeout per call
            )
            elapsed = time.time() - start_time

            if elapsed > 10.0:
                self.logger.warning("llm_call_slow", elapsed_seconds=round(elapsed, 2))

            result_text = strip_thinking(response.choices[0].message.content).strip()

            # Parse JSON array response (empty after think-stripping = no terms found)
            if not result_text:
                self.logger.warning("llm_empty_after_think_strip")
                return []
            raw_terms = json.loads(result_text)

            if not isinstance(raw_terms, list):
                self.logger.warning(
                    "llm_returned_non_list",
                    result=result_text,
                    action_required="verify prompt is correct; may be transient LLM issue - retry recommended"
                )
                return []

            original_terms = [t for t in raw_terms if isinstance(t, str) and t]
            confidences = self._confidence_from_logprobs(response, original_terms)
            scored = [(t.lower(), confidences.get(t.lower(), LLM_FALLBACK_CONFIDENCE)) for t in original_terms]

            self.logger.info("llm_extraction_success", terms_count=len(scored))

            return scored

        except json.JSONDecodeError as e:
            self.logger.error(
                "llm_json_parse_error",
                error=str(e),
                result=result_text,
                action_required="likely transient LLM formatting issue - retry recommended; if persistent, check prompt engineering"
            )
            return []
        except Exception as e:
            error_type = type(e).__name__
            # Categorize error types for actionable guidance
            if "Auth" in error_type or "Permission" in error_type or "API" in error_type:
                action = "check Azure OpenAI credentials and endpoint configuration"
                likely_transient = False
            elif "Timeout" in error_type or "Connection" in error_type:
                action = "network or service issue - retry recommended"
                likely_transient = True
            else:
                action = "investigate error type and context; check logs for details"
                likely_transient = None

            self.logger.error(
                "llm_extraction_failed",
                error=str(e),
                error_type=error_type,
                likely_transient=likely_transient,
                action_required=action
            )
            return []

    def _confidence_from_logprobs(
        self,
        response: Any,
        terms: list[str],
    ) -> dict[str, float]:
        """Compute per-term confidence from chat-completion logprobs.

        Aggregation is weakest-link: ``exp(min(token_logprobs))`` over the tokens
        that span each JSON-quoted term in the response. Missing logprobs or
        unalignable terms fall back to ``LLM_FALLBACK_CONFIDENCE`` (0.70).

        Returns a dict keyed by the **lowercased** term.
        """
        fallback = {t.lower(): LLM_FALLBACK_CONFIDENCE for t in terms}

        try:
            lp = response.choices[0].logprobs
            content_tokens = getattr(lp, "content", None) if lp is not None else None
            if not content_tokens:
                self.logger.warning("llm_logprobs_missing", fallback=LLM_FALLBACK_CONFIDENCE)
                return fallback

            # Build concatenated text + char-offset → token-index mapping.
            parts: list[str] = []
            logprob_values: list[float] = []
            char_to_token: list[int] = []
            for idx, tok in enumerate(content_tokens):
                text = getattr(tok, "token", "") or ""
                if not isinstance(text, str):
                    return fallback
                parts.append(text)
                logprob_values.append(float(getattr(tok, "logprob", 0.0)))
                char_to_token.extend([idx] * len(text))

            full_lower = "".join(parts).lower()

            out: dict[str, float] = {}
            for term in terms:
                term_lower = term.lower()
                needle = f'"{term_lower}"'
                pos = full_lower.find(needle)
                if pos < 0:
                    self.logger.warning(
                        "llm_logprob_term_unalignable",
                        term=term_lower,
                        fallback=LLM_FALLBACK_CONFIDENCE,
                    )
                    out[term_lower] = LLM_FALLBACK_CONFIDENCE
                    continue

                # Skip opening quote; end_char points at closing quote (exclusive).
                start_char = pos + 1
                end_char = pos + len(needle) - 1
                if start_char >= len(char_to_token) or end_char <= start_char:
                    out[term_lower] = LLM_FALLBACK_CONFIDENCE
                    continue

                start_tok = char_to_token[start_char]
                end_tok = char_to_token[end_char - 1] + 1
                slice_ = logprob_values[start_tok:end_tok]
                if not slice_:
                    out[term_lower] = LLM_FALLBACK_CONFIDENCE
                    continue

                conf = math.exp(min(slice_))
                out[term_lower] = max(0.0, min(1.0, conf))

            return out

        except (AttributeError, TypeError, ValueError, IndexError):
            return fallback

    def _get_async_client(self):
        """Get or create async LLM client lazily."""
        if not hasattr(self, '_async_client') or self._async_client is None:
            self._async_client = get_async_chat_client()

        return self._async_client

    async def _llm_extract_async(self, content: str) -> list[tuple[str, float]]:
        """Async version of _llm_extract. Returns (term, confidence) pairs."""
        try:
            client = self._get_async_client()

            if len(content) > 8000:
                content = content[:8000] + "..."

            if ENABLE_CJK_SEGMENTATION and self._is_japanese(content):
                content = self._segment_japanese(content)

            prompt = TERM_EXTRACTION_PROMPT.format(content=content)

            response = await asyncio.wait_for(
                client.chat.completions.create(
                    model=get_chat_model_name(),
                    messages=[
                        {"role": "system", "content": "You are a domain term extraction expert."},
                        {"role": "user", "content": prompt}
                    ],
                    temperature=0.0,
                    max_tokens=500,
                    logprobs=True,
                ),
                timeout=30.0,
            )

            result_text = strip_thinking(response.choices[0].message.content).strip()
            if not result_text:
                return []
            raw_terms = json.loads(result_text)

            if not isinstance(raw_terms, list):
                return []

            original_terms = [t for t in raw_terms if isinstance(t, str) and t]
            confidences = self._confidence_from_logprobs(response, original_terms)
            return [(t.lower(), confidences.get(t.lower(), LLM_FALLBACK_CONFIDENCE)) for t in original_terms]

        except Exception as e:
            self.logger.warning(
                "async_llm_extraction_failed",
                error=str(e),
                error_type=type(e).__name__,
            )
            return []

    async def extract_from_chunks_async(
        self,
        chunks: list[dict[str, Any]],
        doc_type: str = "excel",
    ) -> list[dict[str, Any]]:
        """Extract domain terms from multiple chunks concurrently.

        Uses rule-based extraction first, then dispatches LLM fallback calls
        concurrently for chunks that need it.

        Args:
            chunks: List of chunk dicts
            doc_type: Document type ("excel" or text type)

        Returns:
            List of dicts with domain_category and domain_terms per chunk
        """
        concurrency = settings.llm_concurrency
        semaphore = asyncio.Semaphore(concurrency)

        self.logger.info(
            "async_domain_extraction_starting",
            total_chunks=len(chunks),
            concurrency=concurrency,
        )

        async def extract_single(idx: int, chunk: dict) -> dict[str, Any]:
            """Extract domain terms from a single chunk."""
            try:
                if doc_type == "excel":
                    content = chunk.get("content", {})
                    if isinstance(content, dict):
                        headers = list(content.keys())
                        sheet_name = chunk.get("_source", {}).get("sheet", "")
                        content_parts = []
                        if sheet_name:
                            content_parts.append(sheet_name)
                        content_parts.extend(headers)
                        for v in content.values():
                            if isinstance(v, str) and v.strip():
                                content_parts.append(v)
                        combined = " ".join(content_parts)
                    else:
                        combined = str(content)
                else:
                    text = chunk.get("text") or chunk.get("content", "")
                    combined = str(text)

                if not combined or not combined.strip():
                    return {"domain_terms": [], "domain_category": "general", "domain_confidences": {}}

                # Rule-based first (fast, no I/O)
                rule_terms = self._extract_by_rules(combined)
                terms: list[str] = list(rule_terms)
                confidences: dict[str, float] = {t: RULE_CONFIDENCE for t in rule_terms}

                # LLM fallback if too few terms — with concurrency control
                if len(terms) < LLM_THRESHOLD:
                    async with semaphore:
                        for llm_term, llm_conf in await self._llm_extract_async(combined):
                            terms.append(llm_term)
                            # Max across sources (rule vs LLM).
                            confidences[llm_term] = max(confidences.get(llm_term, 0.0), llm_conf)

                terms = list(dict.fromkeys(terms))[:MAX_TERMS_PER_CHUNK]
                category = self.taxonomy.infer_category(terms)

                return {
                    "domain_terms": terms,
                    "domain_category": category,
                    "domain_confidences": {t: confidences[t] for t in terms},
                }

            except Exception as e:
                self.logger.warning(
                    "async_domain_extraction_failed",
                    chunk_idx=idx,
                    error=str(e),
                )
                return {"domain_terms": [], "domain_category": "general", "domain_confidences": {}}

        import logging
        logging.getLogger("httpx").setLevel(logging.WARNING)

        from tqdm.asyncio import tqdm_asyncio

        tasks = [extract_single(idx, chunk) for idx, chunk in enumerate(chunks)]
        results = await tqdm_asyncio.gather(
            *tasks, desc="Extracting domain terms"
        )

        # Log category distribution
        category_counts: dict[str, int] = {}
        for data in results:
            cat = data["domain_category"]
            category_counts[cat] = category_counts.get(cat, 0) + 1

        self.logger.info(
            "async_domain_extraction_complete",
            total_chunks=len(chunks),
            categories=category_counts,
        )

        return list(results)

    async def extract_from_nodes_async(
        self,
        nodes: list,
        doc_type: str = "word",
    ) -> tuple[set, dict[str, int], dict[str, float]]:
        """Extract domain terms from LlamaIndex nodes concurrently.

        Mutates node.metadata in place with domain_category and domain_terms.
        Uses rule-based extraction first, then dispatches LLM fallback calls
        concurrently for nodes that need it.

        Args:
            nodes: List of LlamaIndex TextNode/BaseNode objects
            doc_type: Document type ("excel", "word", "pdf", etc.)

        Returns:
            Tuple of (all_terms set, category_counts dict, term_confidences dict).
            ``term_confidences`` holds the per-term max confidence across nodes.
        """
        from collections import Counter

        concurrency = settings.llm_concurrency
        semaphore = asyncio.Semaphore(concurrency)

        self.logger.info(
            "async_node_domain_extraction_starting",
            total_nodes=len(nodes),
            concurrency=concurrency,
        )

        async def extract_single(idx: int, node) -> dict[str, Any]:
            """Extract domain terms from a single node."""
            try:
                text = node.text if hasattr(node, "text") else str(node)
                metadata = node.metadata if hasattr(node, "metadata") else {}

                if not text or not text.strip():
                    return {"domain_terms": [], "domain_category": "general", "domain_confidences": {}}

                if doc_type == "excel":
                    sheet_name = metadata.get("sheet_name", "")
                    combined = f"{sheet_name} {text}" if sheet_name else text
                else:
                    combined = text

                # Rule-based first (fast, no I/O)
                rule_terms = self._extract_by_rules(combined)
                terms: list[str] = list(rule_terms)
                confidences: dict[str, float] = {t: RULE_CONFIDENCE for t in rule_terms}

                # LLM fallback if too few terms — with concurrency control
                if len(terms) < LLM_THRESHOLD:
                    async with semaphore:
                        for llm_term, llm_conf in await self._llm_extract_async(combined):
                            terms.append(llm_term)
                            # Max across sources (rule vs LLM).
                            confidences[llm_term] = max(confidences.get(llm_term, 0.0), llm_conf)

                terms = list(dict.fromkeys(terms))[:MAX_TERMS_PER_CHUNK]
                category = self.taxonomy.infer_category(terms)

                return {
                    "domain_terms": terms,
                    "domain_category": category,
                    "domain_confidences": {t: confidences[t] for t in terms},
                }

            except Exception as e:
                self.logger.warning(
                    "async_node_domain_extraction_failed",
                    node_idx=idx,
                    error=str(e),
                )
                return {"domain_terms": [], "domain_category": "general", "domain_confidences": {}}

        import logging
        logging.getLogger("httpx").setLevel(logging.WARNING)

        from tqdm.asyncio import tqdm_asyncio

        tasks = [extract_single(idx, node) for idx, node in enumerate(nodes)]
        results = await tqdm_asyncio.gather(
            *tasks, desc="Extracting domain terms from nodes"
        )

        # Apply results to node metadata in place
        all_terms: set = set()
        category_counts = Counter()
        term_confidences: dict[str, float] = {}

        for node, result in zip(nodes, results):
            node.metadata["domain_category"] = result["domain_category"]
            node.metadata["domain_terms"] = result["domain_terms"]
            all_terms.update(result["domain_terms"])
            category_counts[result["domain_category"]] += 1
            for t, c in result.get("domain_confidences", {}).items():
                # Max across nodes (same term appearing in multiple chunks).
                term_confidences[t] = max(term_confidences.get(t, 0.0), c)

        self.logger.info(
            "async_node_domain_extraction_complete",
            total_nodes=len(nodes),
            categories=dict(category_counts),
        )

        return all_terms, dict(category_counts), term_confidences

    def detect_language(self, term: str) -> str:
        """Detect language of term using Unicode ranges.

        Args:
            term: Term to detect language for

        Returns:
            Language code: "ja", "vi", or "en"
        """
        # Check for Japanese characters (Hiragana, Katakana, CJK)
        if self._is_japanese(term):
            return "ja"

        # Check for Vietnamese diacritics
        vietnamese_chars = set("àáảãạăằắẳẵặâầấẩẫậèéẻẽẹêềếểễệìíỉĩịòóỏõọôồốổỗộơờớởỡợùúủũụưừứửữựỳýỷỹỵđ")
        if any(c in vietnamese_chars for c in term.lower()):
            return "vi"

        return "en"  # Default

    @staticmethod
    def _resolve_persist_confidence(
        term: str,
        confidence_by_term: Optional[dict[str, float]],
    ) -> float:
        """Per-row confidence lookup for ``persist_new_terms``.

        Legacy callers pass ``confidence_by_term=None`` → every row defaults to
        ``LLM_FALLBACK_CONFIDENCE`` (0.70). Otherwise we look up by lowercased
        term and clamp to [0, 1] defensively.
        """
        if confidence_by_term is None:
            return LLM_FALLBACK_CONFIDENCE
        conf = confidence_by_term.get(term.lower(), LLM_FALLBACK_CONFIDENCE)
        return max(0.0, min(1.0, float(conf)))

    def persist_new_terms(
        self,
        terms: list[str],
        domain_category: str,
        *,
        tenant_id: Union[str, UUID],
        node_id: Union[str, UUID],
        document_id: Optional[Union[str, UUID]] = None,
        confidence_by_term: Optional[dict[str, float]] = None,
    ) -> int:
        """Persist new terms to dictionary with concurrent async translations.

        Args:
            terms: List of terms to persist
            domain_category: Domain category for the terms
            tenant_id: Owning tenant (required — scopes RLS).
            node_id: Owning hierarchy node (required — scopes per-node dedup).
            document_id: Optional UUID or string representation of source document.
            confidence_by_term: Optional per-term confidence map (lowercased keys).
                Rows for terms absent from the map (including legacy callers that
                pass ``None``) default to ``LLM_FALLBACK_CONFIDENCE`` (0.70).

        Returns:
            Count of new terms added (0 if all duplicates or on IntegrityError)
        """
        try:
            from sqlalchemy import create_engine
            from sqlalchemy.exc import IntegrityError
            from sqlalchemy.orm import Session

            from src.db.models import DomainTermDictionary
            from src.extraction_v2.translation_service import TranslationService
            from src.knowledge.domain_dictionary_service import DomainDictionaryService
        except ImportError as e:
            self.logger.error(
                "import_error_in_persist_new_terms",
                error=str(e),
                missing_module=e.name if hasattr(e, 'name') else "unknown",
                action_required="ensure_dependencies_in_external_python_venv",
                hint="SQLAlchemy and application modules must be available in Airflow external_python environment"
            )
            raise

        sync_db_url = settings.database_url.replace('postgresql+asyncpg://', 'postgresql+psycopg2://')
        sync_db_url = sync_db_url.replace('@localhost:', '@postgres:')
        engine = create_engine(sync_db_url)
        try:
            with Session(engine) as session:
                service = DomainDictionaryService(
                    session,
                    tenant_id=str(tenant_id) if tenant_id else None,
                    node_id=str(node_id) if node_id else None,
                )
                translator = TranslationService()

                existing = service.get_existing_terms_set()

                # Filter to only new terms
                new_terms = [t for t in terms if t.lower() not in existing]
                if not new_terms:
                    return 0

                self.logger.info("translating_new_terms", count=len(new_terms))

                import logging
                logging.getLogger("httpx").setLevel(logging.WARNING)

                from tqdm.asyncio import tqdm_asyncio

                # Translate all new terms concurrently
                concurrency = getattr(settings, 'llm_concurrency', 5)
                semaphore = asyncio.Semaphore(concurrency)

                async def translate_one(term: str):
                    async with semaphore:
                        source_lang = self.detect_language(term)
                        return await translator.translate_term_async(term, source_lang)

                async def translate_all():
                    tasks = [translate_one(t) for t in new_terms]
                    return await tqdm_asyncio.gather(
                        *tasks, desc="Translating domain terms"
                    )

                translation_results = asyncio.run(translate_all())

                doc_id = None
                if document_id:
                    doc_id = UUID(document_id) if isinstance(document_id, str) else document_id

                new_entries = []
                for term, (translations, success) in zip(new_terms, translation_results):
                    t_id = UUID(str(tenant_id)) if tenant_id else None
                    n_id = UUID(str(node_id)) if node_id else None
                    conf = self._resolve_persist_confidence(term, confidence_by_term)
                    entry = DomainTermDictionary(
                        id=uuid4(),
                        term_original=term,
                        term_ja=translations["ja"],
                        term_en=translations["en"],
                        term_vi=translations["vi"],
                        source_language=self.detect_language(term),
                        domain_category=domain_category,
                        source_document_id=doc_id,
                        translation_failed=not success,
                        confidence=conf,
                        tenant_id=t_id,
                        node_id=n_id,
                    )
                    new_entries.append(entry)

                if new_entries:
                    try:
                        service.add_terms_batch(new_entries)
                        return len(new_entries)
                    except IntegrityError as e:
                        self.logger.warning(
                            "integrity_error_during_batch_insert",
                            error=str(e),
                            attempted_count=len(new_entries),
                            reason="likely_race_condition_duplicate_term"
                        )
                        return 0

                return 0
        finally:
            engine.dispose()
