from datetime import datetime
from difflib import SequenceMatcher
from pathlib import Path
import re
import sys
import time
import unicodedata

import psycopg2
from psycopg2.extras import DictCursor, Json


ROOT_DIR = Path(__file__).resolve().parents[1]
CONN_DIR = ROOT_DIR / "conn"
if str(CONN_DIR) not in sys.path:
    sys.path.insert(0, str(CONN_DIR))

import config
import mysql5
from ai_index import clean_text, ensure_ai_schema, extract_codes, refresh_feedback_summary


MIN_SCORE = 0.07
CONFIDENCE_MIN_SCORE = {
    "bassa": MIN_SCORE,
    "media": 0.18,
    "alta": 0.35,
}
DEFAULT_LIMIT = 5
DIVERSITY_POOL_MULTIPLIER = 8
MAX_SOLUTION_TOKEN_SIMILARITY = 0.62
MAX_SOLUTION_TOKEN_OVERLAP = 0.65
MAX_SOLUTION_TEXT_SIMILARITY = 0.70
_SCHEMA_READY = False
TOKEN_RE = re.compile(r"\b\w{3,}\b", re.IGNORECASE)
SIGNATURE_SPACE_RE = re.compile(r"\s+")


def get_pg_connection():
    return psycopg2.connect(**config.POSTGRES_CONFIG)


def ensure_runtime_schema():
    global _SCHEMA_READY
    if not _SCHEMA_READY:
        ensure_ai_schema()
        _SCHEMA_READY = True


def _id_value(value):
    if value is None or value == "":
        return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _normalize_int_list(value):
    if value is None:
        return []
    if not isinstance(value, (list, tuple, set)):
        value = [value]

    values = []
    for item in value:
        parsed = _id_value(item)
        if parsed is not None and parsed not in values:
            values.append(parsed)
    return values


def _query_text(ticket):
    parts = [
        ticket.get("subject"),
        ticket.get("problem"),
        ticket.get("product"),
        ticket.get("area"),
        ticket.get("subarea"),
        ticket.get("competence"),
        ticket.get("program"),
        ticket.get("program_type"),
    ]
    codes = extract_codes(*parts)
    cleaned = clean_text(" ".join(str(p) for p in parts if p))
    return " ".join(part for part in (cleaned, " ".join(codes)) if part)


def _confidence(score):
    if score >= CONFIDENCE_MIN_SCORE["alta"]:
        return "Alta"
    if score >= CONFIDENCE_MIN_SCORE["media"]:
        return "Media"
    return "Bassa"


def _min_score_from_filters(filters):
    filters = filters or {}
    if filters.get("min_score") is not None:
        try:
            return float(filters["min_score"])
        except (TypeError, ValueError):
            return MIN_SCORE

    confidence = str(filters.get("min_confidence") or "").lower()
    return CONFIDENCE_MIN_SCORE.get(confidence, MIN_SCORE)


def _solution_tokens(solution):
    text = clean_text(solution).lower()
    return set(TOKEN_RE.findall(text))


def _solution_signature(solution):
    text = clean_text(solution).lower()
    text = "".join(
        char for char in unicodedata.normalize("NFKD", text)
        if not unicodedata.combining(char)
    )
    text = re.sub(r"\b(gentile|buongiorno|ciao|salve)\b", " ", text)
    text = re.sub(r"\b(cliente|mauro|martina|lisa|cinzia|nadia)\b", " ", text)
    return SIGNATURE_SPACE_RE.sub(" ", text).strip()


def _jaccard_similarity(left, right):
    if not left or not right:
        return 0
    return len(left & right) / len(left | right)


def _overlap_similarity(left, right):
    if not left or not right:
        return 0
    return len(left & right) / min(len(left), len(right))


def _matches_filter(row, filters, key, row_key):
    values = _normalize_int_list((filters or {}).get(key))
    return values and row.get(row_key) in values


def _passes_strict_metadata_filters(row, filters):
    filters = filters or {}
    filter_columns = (
        ("product_ids", "product_id"),
        ("area_ids", "area_id"),
        ("subarea_ids", "subarea_id"),
        ("competence_ids", "competence_id"),
        ("client_ids", "client_id"),
    )
    for key, column in filter_columns:
        values = _normalize_int_list(filters.get(key))
        if not values:
            values = _normalize_int_list(filters.get(column))
        if values and row.get(column) not in values:
            return False

    excluded_clients = _normalize_int_list(filters.get("exclude_client_ids"))
    if not excluded_clients:
        excluded_clients = _normalize_int_list(filters.get("exclude_client_id"))
    if excluded_clients and row.get("client_id") in excluded_clients:
        return False

    return True


def _diverse_suggestions(rows, query_ticket, limit, filters=None):
    selected = []
    selected_solution_tokens = []
    selected_solution_signatures = []
    duplicates_skipped = 0
    min_score = _min_score_from_filters(filters)

    for row in rows:
        row_dict = dict(row)
        if not _passes_strict_metadata_filters(row_dict, filters):
            continue
        if row_dict["total_score"] < min_score:
            continue

        solution_tokens = _solution_tokens(row_dict.get("solution_preview"))
        solution_signature = _solution_signature(row_dict.get("solution_preview"))
        is_duplicate = any(
            _jaccard_similarity(solution_tokens, existing_tokens) >= MAX_SOLUTION_TOKEN_SIMILARITY
            or _overlap_similarity(solution_tokens, existing_tokens) >= MAX_SOLUTION_TOKEN_OVERLAP
            or SequenceMatcher(None, solution_signature, existing_signature).ratio() >= MAX_SOLUTION_TEXT_SIMILARITY
            for existing_tokens, existing_signature in zip(selected_solution_tokens, selected_solution_signatures)
        )
        if is_duplicate:
            duplicates_skipped += 1
            continue

        row_dict["confidence"] = _confidence(row_dict["total_score"])
        row_dict["reasons"] = _reasons(row_dict, query_ticket, filters)
        row_dict["duplicates_skipped_before"] = duplicates_skipped
        selected.append(row_dict)
        selected_solution_tokens.append(solution_tokens)
        selected_solution_signatures.append(solution_signature)

        if len(selected) >= limit:
            break

    return selected


def _reasons(row, query_ticket=None, filters=None):
    query_ticket = query_ticket or {}
    filters = filters or {}
    reasons = []

    if row["lexical_score"] >= 0.08:
        reasons.append("termini esatti")
    if row["trigram_score"] >= 0.12:
        reasons.append("testo simile")
    if row["code_score"] > 0:
        reasons.append("stesso codice")

    if _matches_filter(row, filters, "product_ids", "product_id"):
        reasons.append("filtro prodotto")
    elif row["product_id"] is not None and row["product_id"] == _id_value(query_ticket.get("product_id")):
        reasons.append("stesso prodotto")

    if _matches_filter(row, filters, "subarea_ids", "subarea_id"):
        reasons.append("filtro sottoarea")
    elif row["subarea_id"] is not None and row["subarea_id"] == _id_value(query_ticket.get("subarea_id")):
        reasons.append("stessa sottoarea")
    elif _matches_filter(row, filters, "area_ids", "area_id"):
        reasons.append("filtro area")
    elif row["area_id"] is not None and row["area_id"] == _id_value(query_ticket.get("area_id")):
        reasons.append("stessa area")

    if _matches_filter(row, filters, "competence_ids", "competence_id"):
        reasons.append("filtro competenza")
    elif row["competence_id"] is not None and row["competence_id"] == _id_value(query_ticket.get("competence_id")):
        reasons.append("stessa competenza")

    if _matches_filter(row, filters, "client_ids", "client_id"):
        reasons.append("stesso cliente")
    if row["feedback_score"] > 0:
        reasons.append("feedback positivo")
    if filters.get("recent_months"):
        reasons.append("periodo recente")
    if not reasons:
        reasons.append("somiglianza generale")
    return reasons[:5]


def _format_date(value):
    if not value:
        return "-"
    if hasattr(value, "strftime"):
        return value.strftime("%Y-%m-%d")
    return str(value)


def _format_suggestion(query_ticket_id, suggestion, rank):
    ticket_id = suggestion["ticket_id"]
    link = f"https://tsnew.sanmarcoweb.com/it/ticket/index/index/operation/view/id/{ticket_id}"
    product = suggestion.get("product") or "-"
    area = suggestion.get("area") or "-"
    subarea = suggestion.get("subarea") or "-"
    competence = suggestion.get("competence") or "-"
    client = suggestion.get("client") or "-"
    confidence = suggestion["confidence"]
    reasons = ", ".join(suggestion["reasons"])
    solution = suggestion.get("solution_preview") or "Soluzione non disponibile"
    diversity_note = ""
    if suggestion.get("duplicates_skipped_before"):
        diversity_note = f"Alternative simili saltate: {suggestion['duplicates_skipped_before']}\n"

    feedback_line = ""
    if query_ticket_id:
        feedback_line = f"Feedback: /utile_{query_ticket_id}_{ticket_id} /non_utile_{query_ticket_id}_{ticket_id}\n"

    return (
        f"\n\n----- {rank}. Confidenza {confidence} -----\n"
        f"[ /Ticket_dettaglio_{ticket_id} ]\n"
        f"{link}\n"
        f"Cliente: {client}\n"
        f"Prodotto: {product} | Competenza: {competence}\n"
        f"Area: {area} | Sottoarea: {subarea}\n"
        f"Apertura: {_format_date(suggestion.get('opened_at'))}\n"
        f"Perche: {reasons}\n"
        f"{diversity_note}"
        f"Score: {suggestion['total_score']:.3f}\n"
        f"SOLUZIONE: {solution}\n"
        f"{feedback_line}"
    )


def _has_metadata_filters(filters):
    filters = filters or {}
    return any(
        _normalize_int_list(filters.get(key))
        for key in ("product_ids", "area_ids", "subarea_ids", "competence_ids", "client_ids")
    )


def _metadata_candidate_clause(ticket, filters=None):
    clauses = []
    params = {}
    for field in ("product_id", "area_id", "subarea_id", "competence_id"):
        value = _id_value((ticket or {}).get(field))
        params[field] = value
        if value is not None:
            clauses.append(f"i.{field} = %({field})s")
    if _has_metadata_filters(filters):
        clauses.append("TRUE")
    if not clauses:
        clauses.append("FALSE")
    return " OR ".join(clauses), params


def _strict_filter_clause(filters):
    filters = filters or {}
    clauses = []
    params = {}

    filter_columns = (
        ("product_ids", "product_id"),
        ("area_ids", "area_id"),
        ("subarea_ids", "subarea_id"),
        ("competence_ids", "competence_id"),
        ("client_ids", "client_id"),
    )
    for key, column in filter_columns:
        values = _normalize_int_list(filters.get(key))
        if not values:
            values = _normalize_int_list(filters.get(column))
        if values:
            clauses.append(f"i.{column} = ANY(%({key})s::int[])")
            params[key] = values

    excluded_clients = _normalize_int_list(filters.get("exclude_client_ids"))
    if not excluded_clients:
        excluded_clients = _normalize_int_list(filters.get("exclude_client_id"))
    if excluded_clients:
        clauses.append("NOT (i.client_id = ANY(%(exclude_client_ids)s::int[]))")
        params["exclude_client_ids"] = excluded_clients

    if filters.get("require_code_match"):
        clauses.append("i.codes && %(query_codes)s::text[]")

    recent_months = _id_value(filters.get("recent_months"))
    if recent_months is not None and recent_months > 0:
        clauses.append("i.opened_at >= now() - (%(recent_days)s::int * interval '1 day')")
        params["recent_days"] = recent_months * 31

    return " AND ".join(clauses) if clauses else "TRUE", params


def _safe_filters(filters):
    safe = {}
    for key, value in (filters or {}).items():
        if value is None or value is False or value == []:
            continue
        if isinstance(value, (list, tuple, set)):
            safe[key] = list(value)
        else:
            safe[key] = value
    return safe


def _search(query_kind, query_text, query_ticket_id=None, query_ticket=None, user_id=None, limit=DEFAULT_LIMIT, offset=0, filters=None, log_results=True):
    ensure_runtime_schema()
    query_text = clean_text(query_text)
    if not query_text:
        return []

    query_ticket = query_ticket or {}
    filters = filters or {}
    limit = max(int(limit or DEFAULT_LIMIT), 1)
    offset = max(int(offset or 0), 0)
    pool_target = offset + limit
    query_codes = extract_codes(query_text)
    metadata_clause, metadata_params = _metadata_candidate_clause(query_ticket, filters)
    strict_clause, strict_params = _strict_filter_clause(filters)

    params = {
        "query_ticket_id": int(query_ticket_id) if query_ticket_id is not None else None,
        "query_text": query_text,
        "problem_text": clean_text(query_ticket.get("problem") or query_text),
        "subject_text": clean_text(query_ticket.get("subject") or query_text),
        "candidate_limit": max(pool_target * 50, 250),
        "final_limit": max(pool_target * DIVERSITY_POOL_MULTIPLIER * 2, 40),
        "query_product_id": _id_value(query_ticket.get("product_id")),
        "query_area_id": _id_value(query_ticket.get("area_id")),
        "query_subarea_id": _id_value(query_ticket.get("subarea_id")),
        "query_competence_id": _id_value(query_ticket.get("competence_id")),
        "query_codes": query_codes,
    }
    params.update(metadata_params)
    params.update(strict_params)

    sql = f"""
        WITH q AS (
            SELECT websearch_to_tsquery('simple', %(query_text)s) AS tsq
        ),
        lexical_candidates AS (
            SELECT i.ticket_id
            FROM ticket_ai_index i, q
            WHERE (%(query_ticket_id)s IS NULL OR i.ticket_id <> %(query_ticket_id)s)
              AND ({strict_clause})
              AND i.search_vector @@ q.tsq
            ORDER BY ts_rank_cd(i.search_vector, q.tsq) DESC
            LIMIT %(candidate_limit)s
        ),
        metadata_candidates AS (
            SELECT i.ticket_id
            FROM ticket_ai_index i
            WHERE (%(query_ticket_id)s IS NULL OR i.ticket_id <> %(query_ticket_id)s)
              AND ({strict_clause})
              AND ({metadata_clause})
            ORDER BY i.opened_at DESC NULLS LAST
            LIMIT %(candidate_limit)s
        ),
        code_candidates AS (
            SELECT i.ticket_id
            FROM ticket_ai_index i
            WHERE (%(query_ticket_id)s IS NULL OR i.ticket_id <> %(query_ticket_id)s)
              AND ({strict_clause})
              AND i.codes && %(query_codes)s::text[]
            ORDER BY i.opened_at DESC NULLS LAST
            LIMIT %(candidate_limit)s
        ),
        candidate_ids AS (
            SELECT ticket_id FROM lexical_candidates
            UNION
            SELECT ticket_id FROM code_candidates
            UNION
            SELECT ticket_id FROM metadata_candidates
        ),
        scored AS (
            SELECT
                i.ticket_id,
                i.subject_text,
                i.problem_text,
                i.solution_preview,
                i.competence_id,
                i.competence,
                i.product_id,
                i.product,
                i.area_id,
                i.area,
                i.subarea_id,
                i.subarea,
                i.client_id,
                i.client,
                i.opened_at,
                ts_rank_cd(i.search_vector, q.tsq) AS lexical_score,
                GREATEST(
                    similarity(i.problem_text, %(problem_text)s),
                    similarity(i.subject_text, %(subject_text)s)
                ) AS trigram_score,
                CASE WHEN i.codes && %(query_codes)s::text[] THEN 0.18 ELSE 0 END AS code_score,
                (
                    CASE WHEN i.product_id IS NOT DISTINCT FROM %(query_product_id)s THEN 0.045 ELSE 0 END +
                    CASE WHEN i.subarea_id IS NOT DISTINCT FROM %(query_subarea_id)s THEN 0.035 ELSE 0 END +
                    CASE WHEN i.area_id IS NOT DISTINCT FROM %(query_area_id)s THEN 0.020 ELSE 0 END +
                    CASE WHEN i.competence_id IS NOT DISTINCT FROM %(query_competence_id)s THEN 0.015 ELSE 0 END
                ) AS metadata_score,
                LEAST(
                    GREATEST(COALESCE(fs.positive_count, 0) - COALESCE(fs.negative_count, 0), 0) * 0.015,
                    0.12
                ) AS feedback_score
            FROM candidate_ids c
            JOIN ticket_ai_index i ON i.ticket_id = c.ticket_id
            CROSS JOIN q
            LEFT JOIN ticket_ai_feedback_summary fs ON fs.ticket_id = i.ticket_id
        )
        SELECT
            *,
            (
                lexical_score * 0.62 +
                trigram_score * 0.28 +
                code_score +
                metadata_score +
                feedback_score
            ) AS total_score
        FROM scored
        ORDER BY total_score DESC, lexical_score DESC, opened_at DESC NULLS LAST
        LIMIT %(final_limit)s
    """

    started_at = time.perf_counter()
    connection = get_pg_connection()
    cursor = connection.cursor(cursor_factory=DictCursor)
    cursor.execute(sql, params)
    rows = cursor.fetchall()
    all_suggestions = _diverse_suggestions(rows, query_ticket, pool_target, filters)
    suggestions = all_suggestions[offset:offset + limit]
    elapsed_ms = int((time.perf_counter() - started_at) * 1000)

    if log_results:
        _log_search(cursor, user_id, query_kind, query_ticket_id, query_text, filters, len(all_suggestions), elapsed_ms)
        if query_kind == "ticket" and query_ticket_id is not None and suggestions:
            _log_suggestions(cursor, user_id, int(query_ticket_id), query_text, suggestions, rank_offset=offset)
        connection.commit()

    cursor.close()
    connection.close()
    return suggestions


def search_similar_tickets(query_ticket_id, user_id=None, limit=DEFAULT_LIMIT, offset=0, filters=None, log_results=True):
    query_ticket = mysql5.get_ticket_context(query_ticket_id)
    if not query_ticket:
        return []

    query_text = _query_text(query_ticket)
    return _search(
        "ticket",
        query_text,
        query_ticket_id=int(query_ticket_id),
        query_ticket=query_ticket,
        user_id=user_id,
        limit=limit,
        offset=offset,
        filters=filters,
        log_results=log_results,
    )


def search_similar_text(query_text, user_id=None, limit=DEFAULT_LIMIT, offset=0, filters=None, log_results=True):
    query_text = clean_text(query_text)
    if not query_text:
        return []
    return _search(
        "text",
        query_text,
        query_ticket_id=None,
        query_ticket={},
        user_id=user_id,
        limit=limit,
        offset=offset,
        filters=filters,
        log_results=log_results,
    )


def _log_search(cursor, user_id, query_kind, query_ticket_id, query_text, filters, result_count, elapsed_ms):
    cursor.execute(
        """
        INSERT INTO ticket_ai_search_log (
            user_id,
            query_kind,
            query_ticket_id,
            query_text,
            filters,
            result_count,
            elapsed_ms
        ) VALUES (%s, %s, %s, %s, %s, %s, %s)
        """,
        (
            user_id,
            query_kind,
            int(query_ticket_id) if query_ticket_id is not None else None,
            query_text,
            Json(_safe_filters(filters)),
            int(result_count),
            int(elapsed_ms),
        ),
    )


def _log_suggestions(cursor, user_id, query_ticket_id, query_text, suggestions, rank_offset=0):
    for rank, suggestion in enumerate(suggestions, start=1 + rank_offset):
        cursor.execute(
            """
            INSERT INTO ticket_ai_query_log (
                user_id,
                query_ticket_id,
                suggested_ticket_id,
                rank,
                total_score,
                lexical_score,
                trigram_score,
                code_score,
                metadata_score,
                feedback_score,
                confidence,
                query_text,
                reasons
            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            """,
            (
                user_id,
                query_ticket_id,
                suggestion["ticket_id"],
                rank,
                float(suggestion["total_score"]),
                float(suggestion["lexical_score"]),
                float(suggestion["trigram_score"]),
                float(suggestion["code_score"]),
                float(suggestion["metadata_score"]),
                float(suggestion["feedback_score"]),
                suggestion["confidence"],
                query_text,
                suggestion["reasons"],
            ),
        )


def record_feedback(user_id, query_ticket_id, suggested_ticket_id, feedback, note=None):
    ensure_runtime_schema()
    if feedback not in ("useful", "not_useful", "favorite", "opened"):
        raise ValueError("feedback must be useful, not_useful, favorite, or opened")

    connection = get_pg_connection()
    cursor = connection.cursor()
    cursor.execute(
        """
        INSERT INTO ticket_ai_feedback (
            user_id,
            query_ticket_id,
            suggested_ticket_id,
            feedback,
            note
        ) VALUES (%s, %s, %s, %s, %s)
        """,
        (user_id, int(query_ticket_id), int(suggested_ticket_id), feedback, note),
    )
    connection.commit()
    cursor.close()
    connection.close()
    refresh_feedback_summary()
    return True


def record_favorite(user_id, ticket_id, query_ticket_id=None, note=None):
    ensure_runtime_schema()
    ticket_id = int(ticket_id)
    query_ticket_id = int(query_ticket_id) if query_ticket_id is not None else ticket_id

    connection = get_pg_connection()
    cursor = connection.cursor()
    cursor.execute(
        """
        SELECT 1
        FROM ticket_ai_feedback
        WHERE user_id = %s
          AND suggested_ticket_id = %s
          AND feedback = 'favorite'
        LIMIT 1
        """,
        (user_id, ticket_id),
    )
    if cursor.fetchone():
        cursor.close()
        connection.close()
        return False

    cursor.execute(
        """
        INSERT INTO ticket_ai_feedback (
            user_id,
            query_ticket_id,
            suggested_ticket_id,
            feedback,
            note
        ) VALUES (%s, %s, %s, 'favorite', %s)
        """,
        (user_id, query_ticket_id, ticket_id, note),
    )
    connection.commit()
    cursor.close()
    connection.close()
    refresh_feedback_summary()
    return True


def get_user_favorites(user_id, limit=20):
    ensure_runtime_schema()
    connection = get_pg_connection()
    cursor = connection.cursor(cursor_factory=DictCursor)
    cursor.execute(
        """
        SELECT *
        FROM (
            SELECT DISTINCT ON (f.suggested_ticket_id)
                f.suggested_ticket_id AS ticket_id,
                f.query_ticket_id,
                f.created_at,
                i.product,
                i.area,
                i.subject_text
            FROM ticket_ai_feedback f
            LEFT JOIN ticket_ai_index i ON i.ticket_id = f.suggested_ticket_id
            WHERE f.user_id = %s
              AND f.feedback = 'favorite'
            ORDER BY f.suggested_ticket_id, f.created_at DESC
        ) favorites
        ORDER BY created_at DESC
        LIMIT %s
        """,
        (user_id, int(limit)),
    )
    rows = [dict(row) for row in cursor.fetchall()]
    cursor.close()
    connection.close()
    return rows


def get_user_bot_settings(user_id):
    ensure_runtime_schema()
    connection = get_pg_connection()
    cursor = connection.cursor(cursor_factory=DictCursor)
    cursor.execute(
        """
        INSERT INTO ticket_bot_user_settings (user_id)
        VALUES (%s)
        ON CONFLICT (user_id) DO NOTHING
        """,
        (int(user_id),),
    )
    cursor.execute(
        """
        SELECT *
        FROM ticket_bot_user_settings
        WHERE user_id = %s
        """,
        (int(user_id),),
    )
    settings = dict(cursor.fetchone())
    connection.commit()
    cursor.close()
    connection.close()
    return settings


def update_user_bot_settings(user_id, **updates):
    allowed = {
        "notifications_enabled",
        "quiet_start_hour",
        "quiet_end_hour",
        "digest_minutes",
        "high_gravity_only",
        "last_notification_at",
    }
    clean_updates = {key: value for key, value in updates.items() if key in allowed}
    if not clean_updates:
        return get_user_bot_settings(user_id)

    get_user_bot_settings(user_id)
    assignments = [f"{key} = %s" for key in clean_updates]
    values = list(clean_updates.values())
    values.append(int(user_id))

    connection = get_pg_connection()
    cursor = connection.cursor()
    cursor.execute(
        f"""
        UPDATE ticket_bot_user_settings
        SET {", ".join(assignments)},
            updated_at = now()
        WHERE user_id = %s
        """,
        values,
    )
    connection.commit()
    cursor.close()
    connection.close()
    return get_user_bot_settings(user_id)


def set_notifications_enabled(user_id, enabled):
    return update_user_bot_settings(user_id, notifications_enabled=bool(enabled))


def set_quiet_hours(user_id, start_hour, end_hour):
    return update_user_bot_settings(
        user_id,
        quiet_start_hour=max(0, min(23, int(start_hour))),
        quiet_end_hour=max(0, min(23, int(end_hour))),
    )


def clear_quiet_hours(user_id):
    return update_user_bot_settings(user_id, quiet_start_hour=None, quiet_end_hour=None)


def set_digest_minutes(user_id, minutes):
    return update_user_bot_settings(user_id, digest_minutes=max(0, int(minutes)))


def set_high_gravity_only(user_id, enabled):
    return update_user_bot_settings(user_id, high_gravity_only=bool(enabled))


def _is_quiet_now(settings, now=None):
    start = settings.get("quiet_start_hour")
    end = settings.get("quiet_end_hour")
    if start is None or end is None or start == end:
        return False

    hour = (now or datetime.now()).hour
    if start < end:
        return start <= hour < end
    return hour >= start or hour < end


def should_send_notifications(user_id, now=None):
    settings = get_user_bot_settings(user_id)
    now = now or datetime.now()
    if not settings.get("notifications_enabled", True):
        return False
    if _is_quiet_now(settings, now=now):
        return False

    digest_minutes = int(settings.get("digest_minutes") or 0)
    last_notification_at = settings.get("last_notification_at")
    if digest_minutes > 0 and last_notification_at:
        elapsed_seconds = (now - last_notification_at).total_seconds()
        if elapsed_seconds < digest_minutes * 60:
            return False
    return True


def mark_notification_sent(user_id):
    return update_user_bot_settings(user_id, last_notification_at=datetime.now())


def get_ai_stats():
    ensure_runtime_schema()
    connection = get_pg_connection()
    cursor = connection.cursor(cursor_factory=DictCursor)
    cursor.execute("SELECT COUNT(*) AS indexed_count, MAX(indexed_at) AS last_indexed_at FROM ticket_ai_index")
    index_stats = dict(cursor.fetchone())
    cursor.execute(
        """
        SELECT
            COUNT(*) AS search_count,
            COALESCE(AVG(elapsed_ms), 0)::integer AS avg_elapsed_ms,
            COALESCE(SUM(CASE WHEN result_count = 0 THEN 1 ELSE 0 END), 0) AS no_result_count
        FROM ticket_ai_search_log
        """
    )
    search_stats = dict(cursor.fetchone())
    cursor.execute("SELECT feedback, COUNT(*) AS count FROM ticket_ai_feedback GROUP BY feedback")
    feedback_counts = {row["feedback"]: row["count"] for row in cursor.fetchall()}
    cursor.close()
    connection.close()
    return {
        **index_stats,
        **search_stats,
        "feedback_counts": feedback_counts,
    }


def get_no_result_searches(limit=10):
    ensure_runtime_schema()
    connection = get_pg_connection()
    cursor = connection.cursor(cursor_factory=DictCursor)
    cursor.execute(
        """
        SELECT created_at, user_id, query_kind, query_ticket_id, query_text, filters, elapsed_ms
        FROM ticket_ai_search_log
        WHERE result_count = 0
        ORDER BY created_at DESC
        LIMIT %s
        """,
        (int(limit),),
    )
    rows = [dict(row) for row in cursor.fetchall()]
    cursor.close()
    connection.close()
    return rows


def get_negative_feedback(limit=10):
    ensure_runtime_schema()
    connection = get_pg_connection()
    cursor = connection.cursor(cursor_factory=DictCursor)
    cursor.execute(
        """
        SELECT
            f.created_at,
            f.user_id,
            f.query_ticket_id,
            f.suggested_ticket_id,
            i.product,
            i.area,
            i.subject_text
        FROM ticket_ai_feedback f
        LEFT JOIN ticket_ai_index i ON i.ticket_id = f.suggested_ticket_id
        WHERE f.feedback = 'not_useful'
        ORDER BY f.created_at DESC
        LIMIT %s
        """,
        (int(limit),),
    )
    rows = [dict(row) for row in cursor.fetchall()]
    cursor.close()
    connection.close()
    return rows


def get_top_useful(limit=10):
    ensure_runtime_schema()
    connection = get_pg_connection()
    cursor = connection.cursor(cursor_factory=DictCursor)
    cursor.execute(
        """
        SELECT
            fs.ticket_id,
            fs.positive_count,
            fs.negative_count,
            i.product,
            i.area,
            i.subject_text
        FROM ticket_ai_feedback_summary fs
        LEFT JOIN ticket_ai_index i ON i.ticket_id = fs.ticket_id
        ORDER BY fs.positive_count DESC, fs.negative_count ASC
        LIMIT %s
        """,
        (int(limit),),
    )
    rows = [dict(row) for row in cursor.fetchall()]
    cursor.close()
    connection.close()
    return rows


def start_guessing_manual(ticket, user_id=None, filters=None, limit=DEFAULT_LIMIT, offset=0):
    suggestions = search_similar_tickets(ticket, user_id=user_id, limit=limit, offset=offset, filters=filters)
    return [_format_suggestion(int(ticket), suggestion, rank + offset) for rank, suggestion in enumerate(suggestions, start=1)]