jorgasisten

import asyncio
import logging
import re
from dataclasses import dataclass
from typing import Any

from jorgasisten.core.config import Settings, get_settings
from jorgasisten.schemas import AssistantDecision, OperationResult, SheetRow, SheetSchema
from jorgasisten.services.ai_client import OpenAICompatibleClient
from jorgasisten.services.conversation_context import (
    DataConversationContext,
    apply_conversation_context,
    detect_requested_field_aliases,
)
from jorgasisten.services.formatters import build_grounded_data_answer, format_operation_error
from jorgasisten.services.operations import OperationExecutor
from jorgasisten.services.sheet_router import (
    build_read_filters,
    enrich_read_decision,
    fallback_decision,
    find_header_by_alias_priority,
    is_likely_data_question,
    score_schema,
    should_prefer_sheet_read,
)
from jorgasisten.services.sheets import GoogleSheetsClient, resolve_header, score_row_against_filters
from jorgasisten.services.text import compact_search_token_sequence, safe_string, tokenize

logger = logging.getLogger(__name__)


@dataclass(slots=True)
class ResolvedReadRows:
    sheet_name: str
    rows: list[SheetRow]
    filters: dict[str, str]
    supports_requested_field: bool


class AssistantService:
    MAX_CROSS_SHEET_MATCHES = 500
    IDENTITY_HEADER_ALIASES = [
        "kode barang",
        "kode produk",
        "kode",
        "sku",
        "serial unit",
        "serial",
        "invoice",
        "nama barang",
        "nama produk",
        "produk",
        "barang",
        "item",
        "nama",
        "customer",
        "pelanggan",
    ]

    def __init__(
        self,
        sheets: GoogleSheetsClient | None = None,
        ai_client: OpenAICompatibleClient | None = None,
        settings: Settings | None = None,
    ) -> None:
        self.settings = settings or get_settings()
        self.sheets = sheets or GoogleSheetsClient(self.settings)
        self.ai_client = ai_client or OpenAICompatibleClient(self.settings)
        self.executor = OperationExecutor(self.sheets, self.settings)

    async def process_message(
        self,
        message: str,
        actor_role: str,
        actor_label: str = "",
        conversation_context: DataConversationContext | None = None,
    ) -> OperationResult:
        if not self.sheets.is_configured():
            decision = fallback_decision(message, [])
            return OperationResult(
                ok=False,
                user_message=(
                    "google sheets belum aktif, pren. isi GOOGLE_SHEETS_SPREADSHEET_ID di .env "
                    "atau tambah spreadsheet lewat menu Spreadsheet. credential service account tetap wajib ada yaa."
                ),
                decision=decision,
            )

        try:
            logger.info("Process message: '%s' from actor_role='%s' (label='%s')", message, actor_role, actor_label)
            # A data answer must start from a new Google Sheets snapshot. This deliberately
            # invalidates the short-lived performance cache before each user request.
            refresh = getattr(self.sheets, "refresh_runtime_caches", None)
            if callable(refresh):
                await asyncio.to_thread(refresh)
            catalog = await asyncio.to_thread(self.sheets.get_catalog)
            local_decision = fallback_decision(message, catalog)
            is_data_request = local_decision.operation == "summarize" or is_likely_data_question(message, catalog)

            if not is_data_request and self._is_potential_search_query(message):
                search_result = await asyncio.to_thread(self._try_search_all_sheets, message, catalog)
                if search_result:
                    sheet_name, filters = search_result
                    local_decision = AssistantDecision(
                        intent="tanya_data",
                        sheet_name=sheet_name,
                        operation="read",
                        confidence=85,
                        filters=filters,
                        reason="Pencarian lintas-sheet menemukan data yang sesuai.",
                    )
                    is_data_request = True

            # Read-only data questions never depend on the model decision. The model may
            # still assist with controlled write parsing, but it cannot invent read answers.
            if is_data_request:
                decision = local_decision
                if decision.operation == "summarize":
                    decision = decision.model_copy(update={"operation": "read", "intent": "tanya_data"})
                logger.info("Using deterministic Google Sheets read decision: %s", decision)
            else:
                if local_decision.operation == "read":
                    return OperationResult(
                        ok=False,
                        user_message="Informasi tersebut belum tersedia di Google Sheets. Silakan hubungi admin.",
                        decision=local_decision,
                    )
                ai_decision = await self.ai_client.create_decision(message, catalog)
                logger.info("AI Decision: %s", ai_decision)
                decision = ai_decision or local_decision
                
            # Convert a free-form lookup into a read only after a real Sheets match exists.
            if decision.operation == "chat" and self._is_potential_search_query(message):
                search_results = await asyncio.to_thread(
                    self._try_search_all_sheets,
                    message,
                    catalog
                )
                if not search_results:
                    refreshed_catalog = await asyncio.to_thread(self._refresh_catalog)
                    if refreshed_catalog is not None:
                        catalog = refreshed_catalog
                        search_results = await asyncio.to_thread(
                            self._try_search_all_sheets,
                            message,
                            catalog,
                        )
                if search_results:
                    sheet_name, filters = search_results
                    decision = AssistantDecision(
                        intent="tanya_data",
                        sheet_name=sheet_name,
                        operation="read",
                        confidence=85,
                        filters=filters,
                        reason="Diubah dari chat ke read karena kata kunci cocok dengan data di Google Sheets.",
                    )
                    logger.info(
                        "Converted general 'chat' query to contextual 'read' on sheet '%s': %s",
                        sheet_name,
                        decision,
                    )
                
            decision = enrich_read_decision(message, decision, catalog)
            spreadsheet_id = safe_string(getattr(self.sheets, "spreadsheet_id", ""))
            decision = apply_conversation_context(
                message,
                decision,
                catalog,
                conversation_context,
                spreadsheet_id=spreadsheet_id,
            )
            if decision.operation == "ask_clarification" and should_prefer_sheet_read(message, catalog):
                refreshed_decision = await asyncio.to_thread(
                    self._refresh_read_decision,
                    message,
                    conversation_context,
                    spreadsheet_id,
                )
                if refreshed_decision is not None:
                    decision, catalog = refreshed_decision
            logger.info("Final Decision to execute: %s", decision)

            if decision.operation == "chat":
                return OperationResult(
                    ok=False,
                    user_message="Informasi tersebut belum tersedia di Google Sheets. Silakan hubungi admin.",
                    decision=decision,
                )
            
            result = await asyncio.to_thread(
                self.executor.execute,
                decision,
                actor_role,
                actor_label=actor_label,
            )
            logger.info(
                "Execution Result: ok=%s, rows_count=%s, msg='%s'",
                result.ok,
                len(result.rows) if result.rows else 0,
                result.user_message,
            )
            return await self._finalize_result(message, decision, result, catalog)
        except Exception as exc:
            logger.exception("Assistant processing failed: %s", exc)
            decision = fallback_decision(message, [])
            return OperationResult(ok=False, user_message=format_operation_error(exc), decision=decision)

    async def confirm_decision(
        self,
        decision: AssistantDecision,
        actor_role: str,
        prepared_create_data: dict[str, str] | None = None,
        actor_label: str = "",
    ) -> OperationResult:
        try:
            return await asyncio.to_thread(
                self.executor.execute,
                decision,
                actor_role,
                confirmed=True,
                prepared_create_data=prepared_create_data,
                actor_label=actor_label,
            )
        except Exception as exc:
            logger.exception("Confirmed operation failed: %s", exc)
            fallback = fallback_decision("", [])
            return OperationResult(ok=False, user_message=format_operation_error(exc), decision=fallback)

    async def process_decision(
        self,
        decision: AssistantDecision,
        actor_role: str,
        actor_label: str = "",
    ) -> OperationResult:
        try:
            return await asyncio.to_thread(
                self.executor.execute,
                decision,
                actor_role,
                actor_label=actor_label,
            )
        except Exception as exc:
            logger.exception("Structured decision processing failed: %s", exc)
            fallback = fallback_decision("", [])
            return OperationResult(ok=False, user_message=format_operation_error(exc), decision=fallback)

    async def _finalize_result(
        self,
        user_message: str,
        decision: AssistantDecision,
        result: OperationResult,
        catalog: list[SheetSchema],
    ) -> OperationResult:
        if not result.ok or result.requires_confirmation:
            return result
        if decision.operation == "read":
            resolved = await asyncio.to_thread(
                self._resolve_read_rows,
                user_message,
                decision,
                result.rows,
                catalog,
            )
            if resolved is not None:
                result.rows = resolved.rows
                result.decision = decision.model_copy(
                    update={"sheet_name": resolved.sheet_name, "filters": resolved.filters}
                )
                if not resolved.supports_requested_field:
                    result.user_message = self._build_missing_requested_field_message(user_message)
                    return result
            if not result.rows:
                refreshed = await asyncio.to_thread(self._refresh_read_result, user_message, decision)
                if refreshed is not None:
                    result.rows = refreshed.rows
                    result.decision = decision.model_copy(
                        update={"sheet_name": refreshed.sheet_name, "filters": refreshed.filters}
                    )
                    catalog = await asyncio.to_thread(self.sheets.get_catalog)

            if not result.rows:
                query_term = ""
                if decision.filters:
                    query_term = decision.filters.get("query") or list(decision.filters.values())[0]
                
                if query_term:
                    result.user_message = self._build_not_found_message(query_term)
                else:
                    result.user_message = self._build_not_found_message("")
                return result

            result.user_message = build_grounded_data_answer(
                user_message,
                result.decision.sheet_name,
                result.rows,
                max_rows=max(len(result.rows), self.settings.max_result_rows),
            )
            return result
        return result

    def _resolve_read_rows(
        self,
        user_message: str,
        decision: AssistantDecision,
        rows: list[SheetRow],
        catalog: list[SheetSchema],
    ) -> ResolvedReadRows | None:
        requested_aliases = detect_requested_field_aliases(user_message)
        candidates: list[tuple[int, ResolvedReadRows]] = []
        ranked_schemas = self._rank_read_schemas(user_message, decision, catalog, requested_aliases)
        if requested_aliases:
            field_schemas = [
                schema
                for schema in ranked_schemas
                if find_header_by_alias_priority(schema.headers, requested_aliases)
            ]
            if field_schemas:
                ranked_schemas = field_schemas

        for schema in ranked_schemas:
            filter_variants = self._filter_variants_for_schema(
                user_message,
                decision,
                schema.headers,
                requested_aliases,
            )
            if self._requires_full_sheet_scan(user_message) and {} not in filter_variants:
                # Aggregate/latest questions may contain only sheet/header words and no
                # row identity. Keep a full-sheet candidate as a deterministic fallback.
                filter_variants.append({})

            for filters in filter_variants:
                if filters:
                    candidate_rows = self.sheets.find_rows(
                        schema.title,
                        filters,
                        limit=max(self.settings.max_result_rows, self.MAX_CROSS_SHEET_MATCHES),
                    )
                else:
                    candidate_rows = self.sheets.read_rows(schema.title)
                if not candidate_rows:
                    continue

                supports_requested_field = self._rows_support_requested_field(candidate_rows, requested_aliases)
                resolved = ResolvedReadRows(
                    sheet_name=schema.title,
                    rows=candidate_rows,
                    filters=filters,
                    supports_requested_field=supports_requested_field,
                )
                best_row_score = max(score_row_against_filters(row, filters) for row in candidate_rows)
                score = score_schema(user_message, schema) + min(best_row_score, 240)
                score += self._requested_field_specificity(schema.headers, user_message)
                if schema.title == decision.sheet_name:
                    score += 10
                if supports_requested_field:
                    score += 1000
                if any(key != "query" for key in filters):
                    score += 25
                if set(filters) == {"query"} and len(filters["query"].split()) <= 2:
                    score -= 5
                if len(candidate_rows) == 1:
                    score += 10
                candidates.append((score, resolved))

        if candidates:
            candidates.sort(key=lambda item: (item[0], -len(item[1].rows), len(item[1].filters)), reverse=True)
            return candidates[0][1]

        initial_support = self._rows_support_requested_field(rows, requested_aliases)
        if rows:
            return ResolvedReadRows(
                sheet_name=decision.sheet_name,
                rows=rows,
                filters={key: safe_string(value) for key, value in decision.filters.items() if safe_string(value)},
                supports_requested_field=initial_support,
            )
        return None

    def _refresh_read_result(
        self,
        user_message: str,
        decision: AssistantDecision,
    ) -> ResolvedReadRows | None:
        refresh = getattr(self.sheets, "refresh_runtime_caches", None)
        if not callable(refresh):
            return None
        refresh()
        refreshed_catalog = self.sheets.get_catalog()
        return self._resolve_read_rows(user_message, decision, [], refreshed_catalog)

    def _refresh_read_decision(
        self,
        user_message: str,
        conversation_context: DataConversationContext | None,
        spreadsheet_id: str,
    ) -> tuple[AssistantDecision, list[SheetSchema]] | None:
        refresh = getattr(self.sheets, "refresh_runtime_caches", None)
        if not callable(refresh):
            return None
        refresh()
        refreshed_catalog = self.sheets.get_catalog()
        refreshed_decision = enrich_read_decision(
            user_message,
            fallback_decision(user_message, refreshed_catalog),
            refreshed_catalog,
        )
        refreshed_decision = apply_conversation_context(
            user_message,
            refreshed_decision,
            refreshed_catalog,
            conversation_context,
            spreadsheet_id=spreadsheet_id,
        )
        return refreshed_decision, refreshed_catalog

    def _refresh_catalog(self) -> list[SheetSchema] | None:
        refresh = getattr(self.sheets, "refresh_runtime_caches", None)
        if not callable(refresh):
            return None
        refresh()
        return self.sheets.get_catalog()

    def _rank_read_schemas(
        self,
        user_message: str,
        decision: AssistantDecision,
        catalog: list[SheetSchema],
        requested_aliases: list[str],
    ) -> list[SheetSchema]:
        ranked: list[tuple[int, SheetSchema]] = []
        for schema in catalog:
            score = score_schema(user_message, schema)
            if schema.title == decision.sheet_name:
                score += 40
            if requested_aliases and find_header_by_alias_priority(schema.headers, requested_aliases):
                score += 30
            score += self._requested_field_specificity(schema.headers, user_message)
            if any(resolve_header(key, schema.headers) for key in decision.filters if key != "query"):
                score += 15
            ranked.append((score, schema))
        ranked.sort(key=lambda item: item[0], reverse=True)
        return [schema for _, schema in ranked]

    def _requested_field_specificity(self, headers: list[str], user_message: str) -> int:
        """Prefer the exact stock field asked for over related operational columns.

        For example, a request for remaining stock must choose ``Stok Akhir``
        before ``Kuantiti`` (location allocation) or ``Stok Minimum`` (MOQ).
        """
        message_tokens = tokenize(user_message)
        if not message_tokens:
            return 0

        best_score = 0
        for header in headers:
            header_tokens = tokenize(header)
            if not header_tokens:
                continue

            score = 0
            is_stock_header = bool(header_tokens & {"stok", "stock"})
            is_quantity_header = bool(header_tokens & {"qty", "jumlah", "kuantiti", "kuantitas"})
            if message_tokens & {"sisa", "akhir"}:
                if is_stock_header and header_tokens & {"sisa", "akhir"}:
                    score = 650
                elif is_stock_header and header_tokens & {"minimum", "min"}:
                    score = 40
                elif is_stock_header:
                    score = 180
                elif is_quantity_header:
                    score = 20
            elif message_tokens & {"awal"}:
                score = 500 if is_stock_header and "awal" in header_tokens else 0
            elif message_tokens & {"masuk"}:
                score = 500 if "masuk" in header_tokens else 0
            elif message_tokens & {"keluar"}:
                score = 500 if "keluar" in header_tokens else 0
            elif message_tokens & {"stok", "stock"}:
                if is_stock_header and not (header_tokens & {"minimum", "min"}):
                    score = 300
                elif is_quantity_header:
                    score = 50
            best_score = max(best_score, score)
        return best_score

    def _filters_for_schema(
        self,
        user_message: str,
        decision: AssistantDecision,
        headers: list[str],
    ) -> dict[str, str]:
        schema_filters = build_read_filters(user_message, SheetSchema(title="", headers=headers))
        if schema_filters:
            return {key: safe_string(value) for key, value in schema_filters.items() if safe_string(value)}

        remapped_filters: dict[str, str] = {}
        for key, value in decision.filters.items():
            if key == "query":
                continue
            header = resolve_header(key, headers)
            value_text = safe_string(value)
            if header and value_text:
                remapped_filters[header] = value_text
        if remapped_filters:
            return remapped_filters
        return {key: safe_string(value) for key, value in decision.filters.items() if safe_string(value)}

    def _filter_variants_for_schema(
        self,
        user_message: str,
        decision: AssistantDecision,
        headers: list[str],
        requested_aliases: list[str],
    ) -> list[dict[str, str]]:
        variants: list[dict[str, str]] = []
        seen: set[tuple[tuple[str, str], ...]] = set()

        def add_variant(filters: dict[str, str]) -> None:
            normalized = {key: safe_string(value) for key, value in filters.items() if safe_string(value)}
            if not normalized:
                return
            fingerprint = tuple(sorted(normalized.items()))
            if fingerprint in seen:
                return
            seen.add(fingerprint)
            variants.append(normalized)

        add_variant(self._filters_for_schema(user_message, decision, headers))

        remapped_filters: dict[str, str] = {}
        for key, value in decision.filters.items():
            if key == "query":
                continue
            header = resolve_header(key, headers)
            value_text = safe_string(value)
            if header and value_text:
                remapped_filters[header] = value_text
        add_variant(remapped_filters)

        preferred_header = find_header_by_alias_priority(headers, self.IDENTITY_HEADER_ALIASES)
        for query in self._query_variants(user_message, decision.filters, requested_aliases, headers):
            add_variant({"query": query})
            if preferred_header and len(query.split()) >= 2:
                add_variant({preferred_header: query})

        return variants

    def _query_variants(
        self,
        user_message: str,
        decision_filters: dict[str, Any],
        requested_aliases: list[str],
        headers: list[str],
    ) -> list[str]:
        variants: list[str] = []
        seen: set[str] = set()

        def add_query(raw_query: str) -> None:
            query = " ".join(token for token in compact_search_token_sequence(raw_query) if token)
            if not query or query in seen:
                return
            seen.add(query)
            variants.append(query)

        base_query = safe_string(decision_filters.get("query"))
        add_query(base_query)
        add_query(user_message)

        stripped_tokens = self._strip_requested_field_tokens(
            compact_search_token_sequence(base_query or user_message),
            requested_aliases,
        )
        stripped_tokens = self._strip_query_directive_tokens(stripped_tokens, headers)
        add_query(" ".join(stripped_tokens))

        salient_tokens = [token for token in stripped_tokens if len(token) >= 4 or token.isdigit()]
        add_query(" ".join(salient_tokens))
        if len(salient_tokens) > 5:
            add_query(" ".join(salient_tokens[:5]))
        if len(stripped_tokens) > 6:
            add_query(" ".join(stripped_tokens[:6]))

        return variants

    def _strip_query_directive_tokens(self, tokens: list[str], headers: list[str]) -> list[str]:
        directive_tokens = {
            "bandingkan",
            "berapa",
            "bulan",
            "count",
            "data",
            "ini",
            "jumlah",
            "jumlahkan",
            "terbaru",
            "terakhir",
            "total",
        }
        header_tokens = {token for header in headers for token in compact_search_token_sequence(header)}
        return [token for token in tokens if token not in directive_tokens and token not in header_tokens]

    def _requires_full_sheet_scan(self, message: str) -> bool:
        tokens = set(compact_search_token_sequence(message))
        return bool(tokens & {"bandingkan", "count", "jumlah", "jumlahkan", "terbaru", "terakhir", "total"})

    def _strip_requested_field_tokens(self, tokens: list[str], requested_aliases: list[str]) -> list[str]:
        excluded_tokens: set[str] = set()
        for alias in requested_aliases:
            excluded_tokens.update(compact_search_token_sequence(alias))
        return [token for token in tokens if token not in excluded_tokens]

    def _rows_support_requested_field(self, rows: list[SheetRow], requested_aliases: list[str]) -> bool:
        if not rows or not requested_aliases:
            return bool(rows)
        return find_header_by_alias_priority(list(rows[0].values.keys()), requested_aliases) is not None

    def _build_missing_requested_field_message(self, user_message: str) -> str:
        requested_aliases = detect_requested_field_aliases(user_message)
        field_label = requested_aliases[0] if requested_aliases else "field itu"
        return (
            f"datanya ada, tapi kolom {field_label} belum tersedia di sheet yang paling relevan, pren. "
            "coba cek struktur kolomnya atau tanyain field lain yang memang ada."
        )

    def _build_not_found_message(self, query_term: str) -> str:
        return (
            "Data yang Anda tanyakan belum ditemukan di Google Sheets. "
            "Silakan periksa kembali nama, kode, atau informasi yang digunakan."
        )

    def _rows_for_answer(self, rows: list[SheetRow], limit: int = 12) -> list[dict[str, str]]:
        compact_rows: list[dict[str, str]] = []
        for row in rows[:limit]:
            compact_rows.append(
                {
                    key: safe_string(value)
                    for key, value in row.values.items()
                    if safe_string(value)
                }
            )
        return compact_rows

    def _is_potential_search_query(self, message: str) -> bool:
        cleaned = message.strip().lower()
        if cleaned.startswith("/"):
            return False
        if cleaned.startswith(("jelaskan ", "kenapa ", "mengapa ", "bagaimana cara ", "bikinin ", "buatkan ")):
            return False
        
        from jorgasisten.services.sheet_router import GREETING_KEYWORDS
        if cleaned in GREETING_KEYWORDS:
            return False
            
        words = [w for w in re.findall(r"[a-z0-9]+", cleaned) if w]
        if not words:
            return False
            
        if len(words) > 40:
            return False
            
        return True

    def _try_search_all_sheets(self, message: str, catalog: list[SheetSchema]) -> tuple[str, dict[str, Any]] | None:
        try:
            decision = fallback_decision(message, catalog)
            resolved = self._resolve_read_rows(message, decision, [], catalog)
            if resolved is None:
                return None
            logger.info("Found matching data in sheet '%s' for query '%s'", resolved.sheet_name, message)
            return resolved.sheet_name, resolved.filters
        except Exception as exc:
            logger.debug("Failed global read search for '%s': %s", message, exc)
            return None