"""技术支持分析闭环。

本模块冻结检索引用，校验 Agent 输出，并在策略和权限共同允许时执行受控自动解决。
"""

import re
from copy import deepcopy
from difflib import SequenceMatcher

from django.conf import settings
from django.db import transaction
from django.utils import timezone

from apps.agents.models import AgentDefinition, AgentRun, AgentVersion
from apps.agents.services import queue_agent_run
from apps.audit.models import OperationAuditLog
from apps.knowledge.embeddings import KnowledgeEmbeddingError
from apps.knowledge.models import KnowledgeItem
from apps.knowledge.services import retrieve_knowledge
from apps.solutions.models import Solution
from apps.tickets.models import Ticket, TicketActorSource, TicketSupportAnalysis
from apps.tickets.replies import ensure_customer_reply_task
from apps.tickets.services import (
    TicketDataError,
    TicketStatusError,
    adopt_solution_for_ticket,
    get_allowed_transitions,
)

SUPPORT_AGENT_CODE = "support-analysis"
RETRIEVAL_POLICY_VERSION = "published_scope_hybrid_rerank_v2"
SUPPORT_OUTPUT_FIELDS = {
    "analysis_summary",
    "troubleshooting_steps",
    "candidate_solution_ids",
    "customer_reply_draft",
    "citations",
    "auto_resolve_recommended",
}


class SupportAnalysisError(ValueError):
    pass


def _normalize(value: str) -> str:
    return re.sub(r"\s+", " ", value.casefold()).strip()


def _terms(value: str) -> set[str]:
    normalized = _normalize(value)
    words = set(re.findall(r"[a-z0-9][a-z0-9_.:-]{1,}|[\u4e00-\u9fff]{2,}", normalized))
    chinese = "".join(re.findall(r"[\u4e00-\u9fff]", normalized))
    words.update(chinese[index : index + 2] for index in range(max(len(chinese) - 1, 0)))
    return {term for term in words if term}


def _text_score(query: str, content: str) -> float:
    query_terms = _terms(query)
    content_terms = _terms(content)
    keyword_score = (
        len(query_terms & content_terms) / len(query_terms | content_terms)
        if query_terms and content_terms
        else 0.0
    )
    title_score = SequenceMatcher(None, _normalize(query[:200]), _normalize(content[:200])).ratio()
    return min(keyword_score * 0.7 + title_score * 0.3, 1.0)


def _excerpt(content: str, query: str, limit: int = 240) -> str:
    normalized = content.strip()
    if len(normalized) <= limit:
        return normalized
    terms = sorted(_terms(query), key=len, reverse=True)
    lowered = normalized.casefold()
    position = next((lowered.find(term) for term in terms if term in lowered), 0)
    start = max(position - limit // 3, 0)
    end = min(start + limit, len(normalized))
    return f"{'…' if start else ''}{normalized[start:end]}{'…' if end < len(normalized) else ''}"


def build_ticket_support_query(ticket: Ticket) -> str:
    return "\n".join(
        value
        for value in [
            ticket.title,
            ticket.description,
            ticket.problem_performance,
            ticket.error_information,
            ticket.related_system,
            ticket.product_name,
            ticket.module_name,
            ticket.environment,
            ticket.attempted_actions,
            ticket.customer_request,
            *ticket.tags,
        ]
        if value
    ).strip()


def _scope_matches(
    ticket: Ticket,
    *,
    category_id: int | None = None,
    category_code: str = "",
    related_system: str = "",
    environment: str = "",
) -> bool:
    if ticket.category_id and category_id and ticket.category_id != category_id:
        return False
    if category_code and ticket.category_id:
        if _normalize(category_code) != _normalize(ticket.category.code):
            return False
    if related_system and ticket.related_system:
        if _normalize(related_system) != _normalize(ticket.related_system):
            return False
    if environment and ticket.environment:
        if _normalize(environment) != _normalize(ticket.environment):
            return False
    return True


def retrieve_support_references(ticket: Ticket, *, limit: int = 8) -> tuple[str, list[dict]]:
    """检索并冻结支持分析可引用的知识、方案和历史工单候选。"""

    query = build_ticket_support_query(ticket)
    if not query:
        raise SupportAnalysisError("工单缺少可用于检索的问题信息")
    candidates = []
    try:
        hybrid_results = retrieve_knowledge(
            query,
            limit=max(limit * 8, 64),
            published_only=True,
            scope={
                "category_code": ticket.category.code if ticket.category_id else "",
                "related_system": ticket.related_system,
                "environment": ticket.environment,
            },
        )
    except KnowledgeEmbeddingError as error:
        raise SupportAnalysisError(str(error)) from error
    hybrid_by_solution = {
        item["metadata"].get("solution_id"): item
        for item in hybrid_results
        if item["metadata"].get("solution_id")
    }
    hybrid_by_item = {
        item["metadata"].get("item_id"): item
        for item in hybrid_results
        if item["metadata"].get("item_id")
    }
    hybrid_by_history = {
        item["metadata"].get("historical_ticket_id"): item
        for item in hybrid_results
        if item["metadata"].get("historical_ticket_id")
    }

    solutions = Solution.objects.filter(status="published").select_related("category")
    for solution in solutions[:500]:
        if not _scope_matches(
            ticket,
            category_id=solution.category_id,
            related_system=solution.related_system,
            environment=solution.environment_scope,
        ):
            continue
        content = "\n".join(
            [
                solution.title,
                solution.problem_signature,
                solution.root_cause,
                solution.solution,
                solution.risk_note,
            ]
        )
        hybrid_result = hybrid_by_solution.get(solution.id)
        score = (
            float(hybrid_result["score"])
            if hybrid_result is not None
            else _text_score(query, content)
        )
        if score <= 0:
            continue
        candidates.append(
            {
                "source_type": "solution",
                "source_id": solution.id,
                "source_version": solution.updated_at.isoformat(),
                "title": solution.title,
                "excerpt": _excerpt(content, query),
                "score": score,
                "applicable_scope": {
                    "category_code": solution.category.code if solution.category_id else None,
                    "related_system": solution.related_system,
                    "environment": solution.environment_scope,
                },
                "auto_resolve_policy": {
                    "allow_auto_resolve": solution.allow_auto_resolve,
                    "requires_human_action": solution.requires_human_action,
                    "requires_external_confirmation": solution.requires_external_confirmation,
                    "requires_production_change": solution.requires_production_change,
                },
                "retrieval": hybrid_result["metadata"]["retrieval"]
                if hybrid_result is not None
                else {
                    "channels": ["unindexed_text_fallback"],
                    "text_score": round(score, 4),
                    "vector_score": 0.0,
                    "scope_score": 1.0,
                },
            }
        )

    items = KnowledgeItem.objects.filter(
        status="published",
        knowledge_base__enabled=True,
    ).select_related("knowledge_base")
    for item in items[:500]:
        if not _scope_matches(
            ticket,
            category_code=item.category_code,
            related_system=item.related_system,
            environment=item.environment_scope,
        ):
            continue
        content = "\n".join(
            [
                item.title,
                item.problem,
                item.applicable_scope,
                item.error_code,
                item.root_cause,
                item.troubleshooting_steps,
                item.solution,
                item.risk_note,
            ]
        )
        hybrid_result = hybrid_by_item.get(item.id)
        score = (
            float(hybrid_result["score"])
            if hybrid_result is not None
            else _text_score(query, content)
        )
        if score <= 0:
            continue
        candidates.append(
            {
                "source_type": "knowledge",
                "source_id": item.id,
                "source_version": item.updated_at.isoformat(),
                "title": item.title,
                "excerpt": _excerpt(content, query),
                "score": score,
                "applicable_scope": {"description": item.applicable_scope},
                "retrieval": hybrid_result["metadata"]["retrieval"]
                if hybrid_result is not None
                else {
                    "channels": ["unindexed_text_fallback"],
                    "text_score": round(score, 4),
                    "vector_score": 0.0,
                    "scope_score": 1.0,
                },
            }
        )

    history = (
        Ticket.objects.filter(
            status__in=["resolved", "closed"],
            archived_at__isnull=True,
        )
        .exclude(id=ticket.id)
        .select_related("category", "resolution")
    )
    for historical in history[:500]:
        if not _scope_matches(
            ticket,
            category_id=historical.category_id,
            related_system=historical.related_system,
            environment=historical.environment,
        ):
            continue
        resolution = getattr(historical, "resolution", None)
        content = "\n".join(
            [
                historical.title,
                historical.description,
                historical.problem_performance,
                historical.error_information,
                resolution.root_cause if resolution else "",
                resolution.solution_text if resolution else "",
            ]
        )
        hybrid_result = hybrid_by_history.get(historical.id)
        score = (
            float(hybrid_result["score"])
            if hybrid_result is not None
            else _text_score(query, content)
        )
        if score <= 0:
            continue
        candidates.append(
            {
                "source_type": "historical_ticket",
                "source_id": historical.id,
                "source_version": historical.updated_at.isoformat(),
                "title": historical.title,
                "excerpt": _excerpt(content, query),
                "score": score,
                "applicable_scope": {
                    "category_code": historical.category.code
                    if historical.category_id
                    else None,
                    "related_system": historical.related_system,
                    "environment": historical.environment,
                },
                "retrieval": hybrid_result["metadata"]["retrieval"]
                if hybrid_result is not None
                else {
                    "channels": ["unindexed_text_fallback"],
                    "text_score": round(score, 4),
                    "vector_score": 0.0,
                    "scope_score": 1.0,
                },
            }
        )

    candidates.sort(key=lambda item: (item["score"], item["source_version"]), reverse=True)
    references = []
    for rank, item in enumerate(candidates[:limit], start=1):
        reference = deepcopy(item)
        reference["rank"] = rank
        reference["score"] = round(float(reference["score"]), 4)
        references.append(reference)
    return query, references


def validate_support_agent_contract(version: AgentVersion) -> None:
    last_step = version.steps.select_related("skill_version").order_by("-order").first()
    if last_step is None:
        raise SupportAnalysisError("支持分析 Agent 没有可执行步骤")
    output_schema = last_step.skill_version.output_schema
    if not isinstance(output_schema, dict) or output_schema.get("type") != "object":
        raise SupportAnalysisError("支持分析 Agent 的最终输出必须是对象")
    required = set(output_schema.get("required", []))
    missing = sorted(SUPPORT_OUTPUT_FIELDS - required)
    if missing:
        raise SupportAnalysisError(
            f"支持分析 Agent 最终输出契约缺少字段：{', '.join(missing)}"
        )


def validate_support_analysis_output(run: AgentRun, output) -> None:
    if run.definition.code != SUPPORT_AGENT_CODE:
        return
    if not isinstance(output, dict):
        raise SupportAnalysisError("支持分析结果必须是对象")
    missing = sorted(SUPPORT_OUTPUT_FIELDS - set(output))
    if missing:
        raise SupportAnalysisError(f"支持分析结果缺少字段：{', '.join(missing)}")
    try:
        analysis = run.ticket_support_analysis
    except TicketSupportAnalysis.DoesNotExist as error:
        raise SupportAnalysisError("支持分析运行未关联真实工单") from error
    valid_ranks = {item.get("rank") for item in analysis.references}
    citations = output.get("citations")
    if not isinstance(citations, list) or any(
        not isinstance(item, int) or item not in valid_ranks for item in citations
    ):
        raise SupportAnalysisError("支持分析结果包含无效引用")
    if analysis.references and not citations:
        raise SupportAnalysisError("支持分析结果必须引用至少一条检索依据")
    valid_solution_ids = {
        item.get("source_id")
        for item in analysis.references
        if item.get("source_type") == "solution"
    }
    candidate_solution_ids = output.get("candidate_solution_ids")
    if not isinstance(candidate_solution_ids, list) or any(
        not isinstance(item, int) or item not in valid_solution_ids
        for item in candidate_solution_ids
    ):
        raise SupportAnalysisError("支持分析结果引用了检索范围外的解决方案")
    if output.get("auto_resolve_recommended"):
        eligible_solution_ids = {
            item["source_id"]
            for item in analysis.references
            if item.get("source_type") == "solution"
            and item.get("score", 0) >= 0.85
            and item.get("auto_resolve_policy", {}).get("allow_auto_resolve")
            and not item.get("auto_resolve_policy", {}).get("requires_human_action")
            and not item.get("auto_resolve_policy", {}).get("requires_external_confirmation")
            and not item.get("auto_resolve_policy", {}).get("requires_production_change")
        }
        if not eligible_solution_ids.intersection(candidate_solution_ids):
            raise SupportAnalysisError("自动解决建议不满足阈值、范围或风险策略")


def _user_has_permission(user, permission: str) -> bool:
    return bool(
        user
        and user.is_authenticated
        and (user.is_superuser or user.has_perm(f"system.{permission}"))
    )


def _automation_decision(
    analysis: TicketSupportAnalysis,
    output: dict,
) -> tuple[dict, Solution | None]:
    ticket = analysis.ticket
    run = analysis.run
    threshold = min(max(float(settings.TICKET_AUTO_RESOLVE_MIN_SCORE), 0.0), 1.0)
    candidate_ids = output.get("candidate_solution_ids", [])
    references = sorted(
        (
            item
            for item in analysis.references
            if item.get("source_type") == "solution"
            and item.get("source_id") in candidate_ids
        ),
        key=lambda item: float(item.get("score", 0)),
        reverse=True,
    )
    reference = references[0] if references else None
    solution = (
        Solution.objects.select_related("category").filter(id=reference["source_id"]).first()
        if reference
        else None
    )
    existing_resolution = getattr(ticket, "resolution", None)
    checks = {
        "agent_recommended": output.get("auto_resolve_recommended") is True,
        "global_policy_enabled": bool(settings.TICKET_AUTO_RESOLVE_ENABLED),
        "initiator_can_resolve": _user_has_permission(run.created_by, "ticket.resolve"),
        "agent_action_allowed": "ticket.resolve"
        in run.config_snapshot.get("agent", {}).get("allowed_actions", []),
        "candidate_present": solution is not None,
        "published": bool(solution and solution.status == "published"),
        "score_at_least_threshold": bool(
            reference and float(reference.get("score", 0)) >= threshold
        ),
        "scope_matches": bool(
            solution
            and _scope_matches(
                ticket,
                category_id=solution.category_id,
                related_system=solution.related_system,
                environment=solution.environment_scope,
            )
        ),
        "version_unchanged": bool(
            solution
            and reference
            and reference.get("source_version") == solution.updated_at.isoformat()
        ),
        "risk_policy_allows": bool(
            solution
            and solution.allow_auto_resolve
            and not solution.requires_human_action
            and not solution.requires_external_confirmation
            and not solution.requires_production_change
        ),
        "ticket_not_archived": ticket.archived_at is None,
        "status_transition_allowed": "resolved"
        in get_allowed_transitions(
            ticket.status,
            ticket.category.code if ticket.category_id else None,
        ),
        "no_conflicting_resolution": existing_resolution is None
        or existing_resolution.linked_solution_id in {None, solution.id if solution else None},
    }
    reason_labels = {
        "agent_recommended": "agent_not_recommended",
        "global_policy_enabled": "global_policy_disabled",
        "initiator_can_resolve": "initiator_permission_denied",
        "agent_action_allowed": "agent_action_not_allowed",
        "candidate_present": "candidate_solution_missing",
        "published": "solution_not_published",
        "score_at_least_threshold": "score_below_threshold",
        "scope_matches": "scope_mismatch",
        "version_unchanged": "solution_version_changed",
        "risk_policy_allows": "risk_policy_blocked",
        "ticket_not_archived": "ticket_archived",
        "status_transition_allowed": "status_transition_blocked",
        "no_conflicting_resolution": "existing_resolution_conflict",
    }
    reasons = [reason_labels[key] for key, passed in checks.items() if not passed]
    decision = {
        "policy_version": "support_auto_resolve_v1",
        "status": "eligible" if not reasons else "manual_recommendation",
        "threshold": threshold,
        "solution_id": solution.id if solution else None,
        "score": round(float(reference.get("score", 0)), 4) if reference else 0.0,
        "checks": checks,
        "reasons": reasons,
        "applied": False,
        "evaluated_at": timezone.now().isoformat(),
    }
    return decision, solution


@transaction.atomic
def process_support_analysis_result(run: AgentRun) -> TicketSupportAnalysis | None:
    """验证运行产物并生成可信分析，自动动作仍需通过权限与策略判定。"""

    if run.definition.code != SUPPORT_AGENT_CODE or run.status != "succeeded":
        return None
    analysis = (
        TicketSupportAnalysis.objects.select_for_update(of=("self",))
        .select_related(
            "ticket__category",
            "ticket__resolution",
            "run__created_by",
            "run__agent_version",
        )
        .get(run=run)
    )
    if analysis.processed_at is not None:
        return analysis
    output = run.final_output if isinstance(run.final_output, dict) else {}
    ensure_customer_reply_task(
        analysis,
        str(output.get("customer_reply_draft") or ""),
    )
    decision, solution = _automation_decision(analysis, output)
    if decision["status"] == "eligible" and solution is not None:
        try:
            adopt_solution_for_ticket(
                analysis.ticket,
                solution,
                run.created_by,
                remark="支持分析 Agent 通过统一策略自动采纳解决方案",
                source=TicketActorSource.AGENT,
                agent_run_id=run.id,
            )
        except (TicketDataError, TicketStatusError):
            decision["status"] = "manual_recommendation"
            decision["reasons"].append("domain_service_rejected")
        else:
            decision["status"] = "auto_resolved"
            decision["applied"] = True
            analysis.applied_solution = solution
    analysis.automation_decision = decision
    analysis.processed_at = timezone.now()
    analysis.save(
        update_fields=[
            "automation_decision",
            "applied_solution",
            "processed_at",
        ]
    )
    analysis.ticket.events.create(
        event_type="support_automation_evaluated",
        to_value={
            "status": decision["status"],
            "solution_id": decision["solution_id"],
            "applied": decision["applied"],
            "reasons": decision["reasons"],
        },
        operator=run.created_by,
        source=TicketActorSource.AGENT,
        agent_run_id=run.id,
        metadata={"policy_version": decision["policy_version"]},
    )
    OperationAuditLog.objects.create(
        user=run.created_by,
        module="tickets",
        action="support_automation_evaluated",
        target_type="ticket",
        target_id=str(analysis.ticket_id),
        request_snapshot={
            "agent_run_public_id": str(run.public_id),
            "decision": decision,
        },
        result="success",
    )
    return analysis


@transaction.atomic
def start_ticket_support_analysis(ticket: Ticket, *, created_by=None) -> TicketSupportAnalysis:
    """为工单创建唯一活动分析并投递绑定正式 Agent 版本的运行。"""

    ticket = (
        Ticket.objects.select_for_update(of=("self",))
        .select_related("category")
        .get(id=ticket.id)
    )
    if ticket.archived_at is not None:
        raise SupportAnalysisError("已归档工单不能启动支持分析")
    if ticket.status in {"resolved", "closed"}:
        raise SupportAnalysisError("已结束工单不能启动新的支持分析")
    active = (
        TicketSupportAnalysis.objects.select_related("run")
        .filter(ticket=ticket, run__status__in=["queued", "running"])
        .first()
    )
    if active is not None:
        return active
    definition = AgentDefinition.objects.select_related(
        "active_version__model_deployment__provider"
    ).filter(code=SUPPORT_AGENT_CODE).first()
    if definition is None or definition.active_version_id is None:
        raise SupportAnalysisError("尚未发布支持分析 Agent")
    version = definition.active_version
    validate_support_agent_contract(version)
    query, references = retrieve_support_references(ticket)
    retrieval_channels = sorted(
        {
            channel
            for reference in references
            for channel in reference.get("retrieval", {}).get("channels", [])
        }
    )
    embedding_models = sorted(
        {
            reference.get("retrieval", {}).get("query_embedding_model")
            for reference in references
            if reference.get("retrieval", {}).get("query_embedding_model")
        }
    )
    input_payload = {
        "ticket": {
            "id": ticket.id,
            "ticket_no": ticket.ticket_no,
            "title": ticket.title,
            "description": ticket.description,
            "category_code": ticket.category.code if ticket.category_id else None,
            "environment": ticket.environment,
            "related_system": ticket.related_system,
            "product_name": ticket.product_name,
            "module_name": ticket.module_name,
            "attempted_actions": ticket.attempted_actions,
            "error_information": ticket.error_information,
            "customer_request": ticket.customer_request,
            "current_result": ticket.current_result,
        },
        "query": query,
        "references": references,
    }
    run = queue_agent_run(
        version,
        input_payload,
        created_by=created_by,
        config_snapshot_extra={
            "business_object": {"type": "ticket", "id": ticket.id},
            "retrieval": {
                "policy_version": RETRIEVAL_POLICY_VERSION,
                "allowed_statuses": {
                    "knowledge": ["published"],
                    "solution": ["published"],
                    "historical_ticket": ["resolved", "closed"],
                },
                "reference_count": len(references),
                "channels": retrieval_channels,
                "embedding_models": embedding_models,
            },
            "automation": {
                "policy_version": "support_auto_resolve_v1",
                "enabled": bool(settings.TICKET_AUTO_RESOLVE_ENABLED),
                "minimum_score": float(settings.TICKET_AUTO_RESOLVE_MIN_SCORE),
            },
        },
        defer_until_commit=True,
    )
    analysis = TicketSupportAnalysis.objects.create(
        ticket=ticket,
        run=run,
        query=query,
        references=references,
        retrieval_policy=run.config_snapshot["retrieval"],
        created_by=created_by,
    )
    ticket.events.create(
        event_type="support_analysis_started",
        to_value={
            "run_public_id": str(run.public_id),
            "reference_count": len(references),
        },
        operator=created_by,
        source=TicketActorSource.HUMAN,
        agent_run_id=run.id,
        metadata={"retrieval_policy_version": RETRIEVAL_POLICY_VERSION},
    )
    OperationAuditLog.objects.create(
        user=created_by,
        module="tickets",
        action="support_analysis_started",
        target_type="ticket",
        target_id=str(ticket.id),
        request_snapshot={
            "agent_run_public_id": str(run.public_id),
            "reference_count": len(references),
            "retrieval_policy_version": RETRIEVAL_POLICY_VERSION,
        },
        result="success",
    )
    return analysis