Blame for bot/byte_semantics_quality.py

import re

from bot.byte_semantics_base import (
    is_current_events_prompt,
    is_follow_up_prompt,
    is_high_risk_current_events_prompt,
    is_serious_technical_prompt,
)
from bot.byte_semantics_constants import (
    ANALYTICAL_STYLE_TERMS,
    BRIEF_STYLE_TERMS,
    EXISTENCE_QUESTION_TERMS,
    GENERIC_RESPONSE_HINT_TERMS,
    OVERCONFIDENT_HINT_TERMS,
    PLAYFUL_STYLE_TERMS,
    QUALITY_SAFE_FALLBACK,
    QUALITY_STOPWORDS,
    QUESTION_OPENING_TERMS,
    RECOMMENDATION_HINT_TERMS,
    REWRITE_REASON_GUIDANCE,
)
from bot.byte_semantics_current_events import (
    build_server_time_anchor_instruction,
    build_verifiable_prompt,
    has_current_events_confidence_label,
    has_current_events_source_anchor,
    has_current_events_temporal_anchor,
    has_current_events_uncertainty,
    is_canonical_high_risk_fallback,
)


def build_research_priority_instruction(prompt: str) -> str:
    if not is_serious_technical_prompt(prompt):
        return ""
    return (
        "Instrucoes de pesquisa aprofundada: priorize evidencia recente (2025-2026) e fontes confiaveis "
        "(universidades, orgaos de saude, journals). "
        "Quando existir contexto em alta relacionado ao tema, incorpore esse contexto explicitamente. "
        "Se houver contribuicao relevante do Brasil no tema, mencione de forma objetiva."
    )


def build_anti_generic_contract_instruction(prompt: str) -> str:
    clean_prompt = (prompt or "").strip()
    if not clean_prompt:
        return ""
    return (
        "Contrato anti-generico: responda o caso especifico da pergunta, sem texto enciclopedico amplo. "
        "Nao use frases vagas como 'depende', 'em geral' ou 'cada caso' sem antes entregar resposta objetiva."
    )


def build_direct_answer_instruction(prompt: str) -> str:
    normalized = " ".join((prompt or "").lower().split()).strip(" ?!.,:")
    if not normalized:
        return ""
    is_question = "?" in (prompt or "")
    starts_like_question = normalized.startswith(QUESTION_OPENING_TERMS)
    if not is_question and not starts_like_question:
        return ""

    instructions = [
        "Formato de resposta obrigatorio: responda a pergunta principal na primeira linha, sem introducao."
    ]
    if normalized.startswith(EXISTENCE_QUESTION_TERMS):
        instructions.append("Se for pergunta de existencia, comece com 'Sim,' ou 'Nao,'.")
    if any(term in normalized for term in RECOMMENDATION_HINT_TERMS):
        instructions.append("Se houver opcoes concretas, cite 1 a 3 nomes diretamente relevantes.")
    instructions.append("Evite texto generico e nao desvie do foco da pergunta.")
    return " ".join(instructions)


def build_adaptive_ai_instruction(prompt: str) -> str:
    lowered = (prompt or "").lower()
    if any(term in lowered for term in BRIEF_STYLE_TERMS):
        return "Estilo de resposta: ultra objetivo, no maximo 3 linhas."
    if any(term in lowered for term in ANALYTICAL_STYLE_TERMS):
        return "Estilo de resposta: analise pratica em ate 4 linhas, com 2 ou 3 pontos claros."
    if any(term in lowered for term in PLAYFUL_STYLE_TERMS):
        return "Estilo de resposta: humor leve e inteligente, sem perder precisao factual."
    return ""


def _quality_tokenize(text: str) -> list[str]:
    tokens = re.findall(r"[a-z0-9]{3,}", (text or "").lower())
    normalized: list[str] = []
    for token in tokens:
        if len(token) > 4 and token.endswith("s"):
            token = token[:-1]
        normalized.append(token)
    return normalized


def _extract_focus_terms(prompt: str, max_terms: int = 10) -> list[str]:
    terms: list[str] = []
    for token in _quality_tokenize(prompt):
        if token in QUALITY_STOPWORDS:
            continue
        if token.isdigit() and len(token) < 4:
            continue
        if token in terms:
            continue
        terms.append(token)
        if len(terms) >= max_terms:
            break
    return terms


def _count_focus_overlap(focus_terms: list[str], answer_tokens: set[str]) -> int:
    if not focus_terms or not answer_tokens:
        return 0
    overlap = 0
    for focus_term in focus_terms:
        if focus_term in answer_tokens:
            overlap += 1
            continue
        focus_prefix = focus_term[:5]
        if len(focus_prefix) < 4:
            continue
        if any(
            token.startswith(focus_prefix) or focus_term.startswith(token[:5])
            for token in answer_tokens
            if len(token) >= 4
        ):
            overlap += 1
    return overlap


def build_quality_prompt_script(prompt: str, server_time_instruction: str | None = None) -> str:
    clean_prompt = (prompt or "").strip()
    is_current = is_current_events_prompt(clean_prompt)
    is_high_risk_current = is_high_risk_current_events_prompt(clean_prompt)
    active_server_time_instruction = (
        server_time_instruction or build_server_time_anchor_instruction()
    )
    lines = [
        "Script de qualidade obrigatorio:\n"
        "1) Responda a pergunta principal na primeira linha, sem introducao.\n"
        "2) Inclua 1 ou 2 detalhes concretos ligados ao pedido (nome, data, local, obra ou numero).\n"
        "3) Maximize densidade informacional: cada linha precisa trazer dado novo.\n"
        "4) Corte frases vagas/repetitivas que nao agregam informacao.\n"
        "5) Se houver baixa confianca, admita incerteza de forma explicita e peca 1 fonte."
    ]
    if is_current:
        lines.append(f"6) {active_server_time_instruction}")
        lines.append(
            "7) Para tema atual, separe explicitamente o que esta confirmado agora e o que ainda e rumor."
        )
        if is_high_risk_current:
            lines.append(
                "8) Para noticia/anuncio atual, inclua as linhas finais: 'Confianca: alta|media|baixa' e 'Fonte: ...'."
            )
            lines.append(
                f"9) Se faltar confirmacao robusta, use exatamente: '{QUALITY_SAFE_FALLBACK}'"
            )
        else:
            lines.append(
                f"8) Se faltar confirmacao robusta, use exatamente: '{QUALITY_SAFE_FALLBACK}'"
            )
    else:
        lines.append("6) Mantenha foco total no objeto perguntado.")
        lines.append("7) Evite qualquer afirmacao absoluta sem base verificavel.")
    lines.append(
        "10) Nao termine com pergunta aberta, exceto se faltar dado essencial para responder."
    )
    return "\n".join(lines)


def is_low_quality_answer(prompt: str, answer: str) -> tuple[bool, str]:
    clean_prompt = " ".join((prompt or "").split()).strip()
    answer_lines = [
        line.strip()
        for line in (answer or "").replace("\r\n", "\n").replace("\r", "\n").split("\n")
        if line.strip()
    ]
    if not answer_lines:
        return True, "resposta_vazia"
    clean_answer = "\n".join(answer_lines)
    compact_answer = " ".join(clean_answer.split()).strip()
    if not compact_answer:
        return True, "resposta_vazia"

    normalized_prompt = " ".join(clean_prompt.lower().split()).strip(" ?!.,:")
    lowered_answer = clean_answer.lower()
    lowered_compact_answer = compact_answer.lower()
    generic_hits = sum(1 for term in GENERIC_RESPONSE_HINT_TERMS if term in lowered_answer)
    overconfident_hits = sum(1 for term in OVERCONFIDENT_HINT_TERMS if term in lowered_answer)
    has_uncertainty = has_current_events_uncertainty(clean_answer)
    has_temporal_anchor = has_current_events_temporal_anchor(clean_answer)
    has_source_anchor = has_current_events_source_anchor(clean_answer)
    has_confidence_label = has_current_events_confidence_label(clean_answer)

    focus_terms = _extract_focus_terms(clean_prompt)
    answer_tokens = set(_quality_tokenize(compact_answer))
    overlap_count = _count_focus_overlap(focus_terms, answer_tokens)
    first_line = answer_lines[0].strip().lower()

    starts_too_generic = lowered_answer.startswith(
        ("depende", "em geral", "de forma geral", "geralmente", "na maioria dos casos")
    )
    prompt_is_complex = len(clean_prompt) >= 40
    answer_is_short = len(clean_answer) < 70
    follow_up_prompt = is_follow_up_prompt(clean_prompt)
    current_events_prompt = is_current_events_prompt(clean_prompt)
    current_events_high_risk = is_high_risk_current_events_prompt(clean_prompt)
    existence_question = normalized_prompt.startswith(EXISTENCE_QUESTION_TERMS)

    if (
        "conexao com o modelo instavel" in lowered_answer
        or "conexão com o modelo instável" in lowered_answer
    ):
        return True, "modelo_indisponivel"
    if generic_hits >= 2 and overlap_count <= 1:
        return True, "resposta_generica"
    if starts_too_generic and overlap_count <= 1:
        return True, "abertura_generica"
    if existence_question and not first_line.startswith(
        ("sim", "nao", "não", "existe", "tem", "ha ", "há ")
    ):
        return True, "resposta_existencia_sem_posicao"
    if prompt_is_complex and answer_is_short and overlap_count <= 1:
        return True, "resposta_curta_sem_substancia"
    if len(focus_terms) >= 3 and overlap_count == 0 and (generic_hits > 0 or answer_is_short):
        return True, "off_topic"
    if (
        current_events_prompt
        and not follow_up_prompt
        and len(clean_prompt) >= 18
        and not has_temporal_anchor
        and not has_uncertainty
    ):
        return True, "tema_atual_sem_ancora_temporal"
    if (
        current_events_high_risk
        and has_uncertainty
        and not is_canonical_high_risk_fallback(clean_answer)
    ):
        return True, "incerteza_fora_fallback_canonico"
    if current_events_high_risk and not follow_up_prompt and not has_confidence_label:
        return True, "tema_atual_sem_confianca_explicita"
    if (
        current_events_high_risk
        and not follow_up_prompt
        and not has_uncertainty
        and not has_source_anchor
        and len(clean_answer) >= 120
    ):
        return True, "tema_atual_sem_base_verificavel"
    if (
        current_events_prompt
        and has_uncertainty
        and "link/fonte" not in lowered_answer
        and "fonte" not in lowered_answer
    ):
        return True, "incerteza_sem_pedido_de_fonte"
    if lowered_compact_answer.endswith("?") and not has_uncertainty and not follow_up_prompt:
        return True, "termina_com_pergunta_aberta"
    if overconfident_hits > 0 and not has_temporal_anchor and not has_uncertainty:
        return True, "confianca_absoluta_sem_base"
    return False, ""


def build_quality_rewrite_prompt(
    prompt: str, draft_answer: str, reason: str, server_time_instruction: str | None = None
) -> str:
    clean_prompt = (prompt or "").strip()
    clean_draft = (draft_answer or "").strip()
    safe_reason = (reason or "qualidade_insuficiente").strip()
    active_server_time_instruction = (
        server_time_instruction or build_server_time_anchor_instruction()
    )
    reason_fix = REWRITE_REASON_GUIDANCE.get(
        safe_reason, "Ajuste a resposta para ficar objetiva, verificavel e aderente a pergunta."
    )
    return (
        f"Pergunta original: {clean_prompt}\n"
        f"Rascunho anterior reprovado ({safe_reason}): {clean_draft}\n"
        "Reescreva para chat Twitch com as regras:\n"
        f"- {active_server_time_instruction}\n"
        f"- Correcao alvo: {reason_fix}\n"
        "- Primeira linha responde direto.\n"
        "- No maximo 4 linhas.\n"
        "- Alta densidade: cada linha deve adicionar informacao nova.\n"
        "- Inclua 1 ou 2 fatos concretos e relevantes.\n"
        "- Para tema atual, diferencie confirmado agora vs rumor.\n"
        "- Nao termine com pergunta aberta.\n"
        "- Nao use frase vaga/generica.\n"
        "- Se nao der para confirmar, use exatamente: "
        f"'{QUALITY_SAFE_FALLBACK}'"
    )


def build_llm_enhanced_prompt(prompt: str, server_time_instruction: str | None = None) -> str:
    clean_prompt = (prompt or "").strip()
    if not clean_prompt:
        return ""

    serious_mode = is_serious_technical_prompt(clean_prompt)
    active_server_time_instruction = (
        server_time_instruction or build_server_time_anchor_instruction()
    )
    enriched_prompt = build_verifiable_prompt(
        clean_prompt,
        concise_mode=not serious_mode,
        server_time_instruction=active_server_time_instruction,
    )
    extra_instructions: list[str] = []
    if active_server_time_instruction not in enriched_prompt:
        extra_instructions.append(active_server_time_instruction)

    anti_generic_instruction = build_anti_generic_contract_instruction(clean_prompt)
    if anti_generic_instruction:
        extra_instructions.append(anti_generic_instruction)
    direct_answer_instruction = build_direct_answer_instruction(clean_prompt)
    if direct_answer_instruction:
        extra_instructions.append(direct_answer_instruction)

    if is_follow_up_prompt(clean_prompt):
        extra_instructions.append(
            "Instrucoes de continuidade: trate esta pergunta como follow-up e resolva referencias vagas "
            "(ex.: ele, ela, isso) usando o historico recente do chat."
        )

    research_instruction = build_research_priority_instruction(clean_prompt)
    if research_instruction:
        extra_instructions.append(research_instruction)

    if serious_mode:
        extra_instructions.append(
            "Para tema tecnico serio e relevante, responda em uma unica mensagem. "
            "Contrato obrigatorio: no maximo 4 linhas, alta densidade, sem separador de multiparte."
        )

    if not is_current_events_prompt(clean_prompt):
        adaptive_instruction = build_adaptive_ai_instruction(clean_prompt)
        if adaptive_instruction:
            extra_instructions.append(adaptive_instruction)

    quality_script = build_quality_prompt_script(
        clean_prompt, server_time_instruction=active_server_time_instruction
    )
    if quality_script:
        extra_instructions.append(quality_script)

    if not extra_instructions:
        return enriched_prompt
    return f"{enriched_prompt}\n" + "\n".join(extra_instructions)