from __future__ import annotations

import re

from common.user_messages import friendly_popup_text


# 实时日志只给用户看任务进度和最终失败原因；完整细节仍写入运行日志文件。
FAILURE_EXPORT_NOTICE = "请点击“导出日志”，把日志压缩包发给维护人员，简要说明情况即可。"
CONTACT_LOG_ALLOWED_AUTH_CODES = {"WW", "KHJLWW"}
_CONTACT_LOG_KEYWORDS = (
    "官方交流群",
    "联系方式",
    "天翼云4.5折交流群",
    "华为云8.8折交流群",
    "火山引擎交流群",
    "技术微信",
    "159500025",
    "916383784",
    "1051985214",
    "lywwyjs",
)
_CONTACT_LOG_COMPANY_LINE = "洛阳蛙蛙云计算"
_CONTACT_LOG_SEPARATOR = "+--------------------------------------------------+"
_IP_PREFIX_RE = re.compile(r"^\[[^\]]+\]\s*")
_current_manager_auth_code_value = ""
_contact_log_suppression_remaining = 0

_REGION_TOKEN_LABELS = {
    "cn-beijing": "北京",
    "cn-shanghai": "上海",
    "cn-guangzhou": "广州",
    "cn-north-4": "北京",
    "cn-east-3": "上海",
    "cn-south-1": "广州",
    "cn-southwest-2": "贵阳",
    "cn-north-9": "乌兰察布",
    "cn-east-4": "华东",
    "cn-east-5": "青岛",
}

_RESOURCE_TOKEN_REPLACEMENTS = (
    (r"security_group", "安全组"),
    (r"network_interface", "网卡"),
    (r"subnet", "子网"),
    (r"vpc", "VPC"),
    (r"ecs", "ECS"),
    (r"eip", "EIP"),
)

_STRONG_FAILURE_PATTERNS = (
    r"任务.*失败",
    r"一键.*失败",
    r"购买.*失败",
    r"搭建.*失败",
    r"读取账号资源.*失败",
    r"资源.*查询.*失败",
    r"查询.*失败",
    r"清空.*失败",
    r"删除.*失败",
    r"退订.*失败",
    r"释放.*失败",
    r"重装.*失败",
    r"最终失败",
    r"失败原因",
    r"错误汇总",
    r"地区异常",
    r"HTTP Error",
    r"Forbidden",
    r"unknown encoding",
    r"not permitted",
)

_FLOW_KEYWORDS = (
    "=====",
    "步骤:",
    "步骤：",
    "正在",
    "检测到",
    "复用",
    "已有",
    "开始",
    "完成",
    "成功",
    "已启动",
    "任务",
    "流程",
    "购买",
    "搭建",
    "读取账号资源",
    "账号资源查询完成",
    "有资源地区",
    "发现",
    "查询资源",
    "退订",
    "删除",
    "释放",
    "重装",
    "导出",
    "生成结果",
    "售罄",
    "跳过该地区",
    "请",
    "请先",
    "请更换",
    "当前没有",
    "已取消",
)

_FAILURE_KEYWORDS = (
    "失败",
    "异常",
    "错误",
    "无法",
    "未获取",
    "未完成",
    "无可用",
    "超时",
    "超过",
    "Error",
    "Forbidden",
    "denied",
    "not permitted",
)

_NON_BLOCKING_KEYWORDS = (
    "不影响",
    "继续",
    "重试",
    "稍后",
    "暂未",
    "等待",
    "兜底",
    "按删除完成处理",
    "按释放完成处理",
    "按完成处理",
    "按成功处理",
    "已不存在",
    "仍在",
)

_NOISY_KEYWORDS = (
    "当前目录:",
    "当前目录：",
    "开始时间:",
    "开始时间：",
    "日志文件:",
    "日志文件：",
    "请求ID=",
    "请求ID:",
    "request_id=",
    "RequestId",
    "Traceback is disabled",
    "SSH入口探测",
    "SSH入口确认",
    "SSH端口暂未就绪",
    "等待 SSH 端口开放",
    "SOCKS5认证检查",
    "端口暂未就绪",
)

_TRANSIENT_NETWORK_PATTERNS = (
    r"接口连接中断或暂时失败",
    r"自动重试",
    r"ConnectionResetError",
    r"ConnectionException",
    r"HTTPSConnectionPool",
    r"Max retries exceeded",
    r"Connection aborted",
    r"ProtocolError",
    r"WinError\s*10054",
    r"\b10054\b",
    r"远程主机强迫关闭",
    r"Read timed out",
    r"ConnectTimeout",
    r"timeout",
)

_AUTH_OR_PERMISSION_PATTERNS = (
    r"Incorrect IAM authentication",
    r"verify ak sk signature failed",
    r"APIGW\.0301",
    r"\bHTTP\s*401\b",
    r"\b401\b.*(?:AK|SK|signature|AccessKey|SecurityKey)",
    r"\bHTTP(?: Error)?\s*403\b",
    r"Forbidden",
    r"Unauthorized",
)

_TECHNICAL_ERROR_PATTERNS = (
    *_TRANSIENT_NETWORK_PATTERNS,
    *_AUTH_OR_PERMISSION_PATTERNS,
    r"Traceback",
    r"canonical_request",
    r"x-sdk-date",
    r"request_id",
    r"unknown encoding",
    r"Tcl data",
    r"FileNotFoundError",
)

_FINAL_FAILURE_HINTS = (
    "失败",
    "最终失败",
    "失败原因",
    "错误汇总",
    "地区异常",
    "阶段失败",
    "任务结束，成功",
    "存在地区子任务失败",
    "未完成",
    "超时",
    "停止",
    "中止",
    "failed",
)

_NOTICE_TERMINAL_PATTERNS = (
    r"任务.*失败",
    r"最终失败",
    r"失败原因",
    r"错误汇总",
    r"地区异常",
    r"阶段失败",
    r"存在地区子任务失败",
)

_NOTICE_ACTION_FAILURE_PATTERNS = (
    r"一键.*失败",
    r"购买.*失败",
    r"搭建.*失败",
    r"读取账号资源.*失败",
    r"资源.*查询.*失败",
    r"清空.*失败",
    r"删除.*失败",
    r"退订.*失败",
    r"释放.*失败",
    r"重装.*失败",
)

_GENERIC_FAILURE_SUMMARY_PATTERNS = (
    r"存在地区子任务失败",
    r"请查看上方日志",
    r"任务已结束",
    r"子任务结束，退出码",
    r"阶段结束，退出码",
    r"购买进程退出码=\d+\s*$",
)

_REALTIME_ADVICE_PATTERNS = (
    r"；?如果连续出现[，,]?请点击“导出日志”，把日志压缩包发给维护人员。?",
    r"；?如果连续出现[，,]?请导出日志发给维护人员。?",
)

_ITEM_FAILURE_DETAIL_PATTERNS = (
    r"\[[^\]]+\].*失败[:：].*(?:HTTP=|返回=|RequestId|请求ID|Invalid[A-Za-z.]+|Error)",
    r"\[[^\]]+\].*失败[:：].*(?:[A-Z][A-Za-z0-9]*(?:\.[A-Za-z0-9_-]+)+|Quota exceeded|配额|售罄|No module named)",
    r"^[^:：\n]+[:：]\s*(?:HTTP=|返回=|.*RequestId|.*请求ID|.*Invalid[A-Za-z.]+|.*Error)",
    r"^[^:：\n]+[:：]\s*(?:.*[A-Z][A-Za-z0-9]*(?:\.[A-Za-z0-9_-]+)+|.*Quota exceeded|.*配额|.*售罄|.*No module named)",
)

_ITEM_FAILURE_DETAIL_TOKENS = (
    r"HTTP\s*=",
    r"返回\s*=",
    r"RequestId",
    r"请求ID",
    r"\bInvalid[A-Za-z.]+",
    r"\bError\b",
    r"Quota exceeded",
    r"配额",
    r"售罄",
    r"No module named",
    r"\b(?:EIP|VPC|ECS|IMS|EVS|BMS|IAM|APIGW|Openapi)[A-Za-z0-9_.-]*\b",
)

_FINAL_FAILURE_SUMMARY_PATTERNS = (
    r".*结束[，,]\s*成功\s+\d+/\d+[，,]\s*失败\s*[1-9]\d*",
    r".*结束[，,]\s*成功\s+\d+\s*/\s*\d+[，,]\s*失败\s*[1-9]\d*",
    r".*失败原因[:：].+",
    r".*错误汇总[:：].+",
    r".*任务失败[:：].+",
    r".*最终失败.*",
)


def _matches_any(patterns: tuple[str, ...], text: str) -> bool:
    return any(re.search(pattern, text, re.IGNORECASE) for pattern in patterns)


def set_manager_auth_code_for_log_filter(code: str) -> None:
    global _current_manager_auth_code_value, _contact_log_suppression_remaining
    _current_manager_auth_code_value = re.sub(r"\s+", "", str(code or "")).upper()
    _contact_log_suppression_remaining = 0


def _current_manager_auth_code() -> str:
    return _current_manager_auth_code_value


def _should_show_contact_log() -> bool:
    return _current_manager_auth_code() in CONTACT_LOG_ALLOWED_AUTH_CODES


def _contact_log_body(line: str) -> str:
    text = _IP_PREFIX_RE.sub("", str(line or "").strip()).strip()
    if text.startswith("|"):
        text = text[1:].strip()
    return text


def _is_contact_log_line(line: str) -> bool:
    global _contact_log_suppression_remaining
    if _should_show_contact_log():
        _contact_log_suppression_remaining = 0
        return False
    body = _contact_log_body(line)
    is_contact_body = body == _CONTACT_LOG_COMPANY_LINE or any(keyword in body for keyword in _CONTACT_LOG_KEYWORDS)
    if is_contact_body:
        _contact_log_suppression_remaining = 12
        return True
    if body == _CONTACT_LOG_SEPARATOR:
        _contact_log_suppression_remaining = max(_contact_log_suppression_remaining, 12)
        return True
    if _contact_log_suppression_remaining <= 0:
        return False
    if not body:
        _contact_log_suppression_remaining -= 1
        return True
    _contact_log_suppression_remaining = 0
    return False


def _strip_separator_marks(text: str) -> str:
    value = str(text or "").strip()
    if not value:
        return ""
    if re.fullmatch(r"[=\-_]{3,}", value):
        return ""
    match = re.match(r"^(\[[^\]]+\]\s*)?([=\-_]{2,})\s*(.*?)\s*\2$", value)
    if match:
        prefix = match.group(1) or ""
        body = match.group(3).strip()
        if not body:
            return ""
        return f"{prefix}{body}".strip()
    match = re.match(r"^(\[[^\]]+\]\s*)?([=\-_]{2,})\s*(.+?)\s*[=\-_]{2,}$", value)
    if match:
        prefix = match.group(1) or ""
        body = match.group(3).strip()
        if not body:
            return ""
        return f"{prefix}{body}".strip()
    return value


def _localize_display_tokens(text: str) -> str:
    result = str(text or "")
    for code, label in sorted(_REGION_TOKEN_LABELS.items(), key=lambda item: len(item[0]), reverse=True):
        result = re.sub(rf"\[{re.escape(code)}(?:-(?:one|two|\d+))?\]", f"[{label}]", result, flags=re.IGNORECASE)
        result = re.sub(rf"(?<![\w.-]){re.escape(code)}(?![\w.-])", label, result, flags=re.IGNORECASE)
    for token, label in _RESOURCE_TOKEN_REPLACEMENTS:
        result = re.sub(rf"(?<![\w.-]){token}(?![\w.-])", label, result, flags=re.IGNORECASE)
    return result


def clean_realtime_display_line(line: str) -> str:
    """只清洗界面实时日志显示文本；完整原文仍保留在日志文件和结果输出里。"""
    if _is_contact_log_line(line):
        return ""
    text = _strip_separator_marks(line)
    if not text:
        return ""
    text = re.sub(r"^(\[[^\]]+\]\s*)?自动购买:\s*", r"\1", text)
    text = _localize_display_tokens(text)
    return text.strip()


def _has_final_failure_hint(text: str) -> bool:
    lower = text.lower()
    return any(keyword.lower() in lower for keyword in _FINAL_FAILURE_HINTS)


def _has_terminal_failure_hint(text: str) -> bool:
    terminal_keywords = (
        "任务失败",
        "最终失败",
        "失败原因",
        "错误汇总",
        "地区异常",
        "阶段失败",
        "存在地区子任务失败",
        "未完成",
        "超时",
        "超过",
        "已停止",
        "停止",
        "中止",
        "failed",
    )
    lower = text.lower()
    return any(keyword.lower() in lower for keyword in terminal_keywords)


def _has_non_blocking_hint(text: str) -> bool:
    """识别不影响任务的中途失败提示，避免把“无法继续”误判成可继续。"""
    if re.search(r"(无法|不能|不可|未能|不再|停止|中止).{0,8}继续", text):
        return False
    return any(keyword in text for keyword in _NON_BLOCKING_KEYWORDS)


def _append_failure_notice(message: str) -> str:
    text = remove_failure_export_notice(str(message or "")).strip()
    if not text:
        text = "任务失败。"
    return f"{text}\n{FAILURE_EXPORT_NOTICE}"


def remove_failure_export_notice(message: str) -> str:
    """移除实时日志导出提示，供各云防重复显示时复用。"""
    text = str(message or "")
    lines = [line for line in text.splitlines() if line.strip() != FAILURE_EXPORT_NOTICE]
    return "\n".join(lines).strip()


def has_failure_export_notice(message: str) -> bool:
    """判断过滤后的实时日志是否包含导出日志提示。"""
    return FAILURE_EXPORT_NOTICE in str(message or "")


def _friendly_realtime_text(text: str) -> str:
    """弹窗友好文案用于实时日志时，去掉重复维护提示，由过滤器统一追加一次。"""
    friendly = friendly_popup_text(text)
    for pattern in _REALTIME_ADVICE_PATTERNS:
        friendly = re.sub(pattern, "。", friendly)
    return remove_failure_export_notice(friendly).strip()


def _technical_realtime_message(text: str) -> str:
    """把 SDK/HTTP 原文压缩为用户能看懂的实时日志；空字符串表示只写文件。"""
    if _matches_any(_AUTH_OR_PERMISSION_PATTERNS, text):
        if _has_final_failure_hint(text):
            return _friendly_realtime_text(text)
        return ""

    if _matches_any(_TRANSIENT_NETWORK_PATTERNS, text):
        if _has_non_blocking_hint(text) and not _has_terminal_failure_hint(text):
            return ""
        if _has_terminal_failure_hint(text):
            return (
                "任务失败：云平台连接被临时断开。\n"
                "建议稍后重试；如果连续失败，请检查本机网络，或减少同时运行的任务数量后再试。"
            )
        return ""

    if _matches_any(_TECHNICAL_ERROR_PATTERNS, text):
        if _has_final_failure_hint(text):
            return _friendly_realtime_text(text)
        return ""

    return ""


def _is_failure_line(text: str) -> bool:
    return bool(text.strip()) and (
        any(keyword in text for keyword in _FAILURE_KEYWORDS)
        or _matches_any(_STRONG_FAILURE_PATTERNS, text)
        or _matches_any(_TECHNICAL_ERROR_PATTERNS, text)
    )


def _is_non_blocking_failure_line(text: str) -> bool:
    if not _is_failure_line(text):
        return False
    if _matches_any(_NOTICE_TERMINAL_PATTERNS, text):
        return False
    return _has_non_blocking_hint(text)


def _is_generic_failure_summary(text: str) -> bool:
    """汇总类失败行只提示流程失败，不把它当成明确失败原因。"""
    return _matches_any(_GENERIC_FAILURE_SUMMARY_PATTERNS, text)


def _has_specific_failure_reason(text: str) -> bool:
    """只有带具体原因的失败行，才追加导出日志提示。"""
    if _is_generic_failure_summary(text):
        return False
    if _matches_any(_TECHNICAL_ERROR_PATTERNS, text):
        return True
    if _matches_any((r"失败原因", r"错误汇总", r"地区异常"), text):
        return True
    if re.search(r"(失败|异常|错误|超时|未完成|无法)[^：:\n]{0,24}[：:]\s*\S+", text):
        return True
    if "退出码" in text and any(keyword in text for keyword in ("原因", "通常", "不足", "未买齐", "不可用", "失败")):
        return not _is_generic_failure_summary(text)
    return False


def _is_item_failure_detail(text: str) -> bool:
    """识别单个资源/子任务的技术失败详情；界面延后到最终摘要统一展示原因。"""
    if not _matches_any(_ITEM_FAILURE_DETAIL_PATTERNS, text):
        return False
    if any(keyword in text for keyword in _FAILURE_KEYWORDS):
        return True
    return _matches_any(_ITEM_FAILURE_DETAIL_TOKENS, text)


def _is_final_failure_summary(text: str) -> bool:
    """识别任务最终失败摘要；只有这里才显示导出日志提示。"""
    if _is_item_failure_detail(text):
        return False
    if _matches_any(_FINAL_FAILURE_SUMMARY_PATTERNS, text):
        return True
    if _matches_any(_NOTICE_TERMINAL_PATTERNS, text) and _has_specific_failure_reason(text):
        return True
    return False


def _compact_failure_reason(text: str) -> str:
    """把单条云 API 技术错误压缩成用户能看懂的失败原因，完整原文仍保留在日志文件。"""
    raw = str(text or "").strip()
    label = raw
    if "]" in label:
        label = label.split("]", 1)[0].lstrip("[")
    if ":" in label:
        label = label.split(":", 1)[0].strip()
    if "：" in label:
        label = label.split("：", 1)[0].strip()

    code_match = (
        re.search(r'"Code"\s*:\s*"([^"]+)"', raw)
        or re.search(r"\bCode[=:]\s*([^;，,\s]+)", raw)
        or re.search(r"\b((?:EIP|VPC|ECS|IMS|EVS|BMS|IAM|APIGW|Openapi)[A-Za-z0-9_.-]*)\b", raw)
    )
    message_match = re.search(r'"Message"\s*:\s*"([^"]+)"', raw) or re.search(r"\bMessage[=:]\s*([^;]+)", raw)
    code = (code_match.group(1) if code_match else "").strip()
    message = (message_match.group(1) if message_match else "").strip()
    lower_text = raw.lower()
    lower_code = code.lower()

    generic_resource_codes = {"ecs", "eip", "vpc", "ims", "evs", "bms", "iam", "apigw", "openapi"}

    if "account forbidden" in lower_text or "freezen" in lower_text or "frozen" in lower_text:
        reason = "账号被冻结、信用不足或余额不足，云平台禁止继续购买资源"
    elif "credit not enough" in lower_text or "insufficient balance" in lower_text or "余额不足" in raw:
        reason = "账号余额或信用额度不足，云平台拒绝创建资源"
    elif "sold out" in lower_text or "售罄" in raw:
        reason = "当前地区资源售罄或规格不可用，请更换地区或规格后重试"
    elif ("ssh" in lower_text or "ssh" in raw.lower()) and ("timeout" in lower_text or "超时" in raw):
        reason = "SSH 连接超时，通常是云服务器未就绪、安全组未放行或公网 IP 不通"
    elif "quota exceeded" in lower_text and "subeni_attached_to_vm" in lower_text:
        reason = "辅助弹性网卡已达到当前云服务器可绑定上限，需要清理旧资源或重建该地区资源"
    elif "quota exceeded" in lower_text and "flavor_contain_subeni" in lower_text:
        reason = "当前云服务器规格不支持所需辅助弹性网卡数量，需要重建为匹配规格"
    elif "quota exceeded" in lower_text and "publicip" in lower_text:
        reason = "弹性公网IP配额不足，需要释放未使用公网IP或提升配额后重试"
    elif "quota exceeded" in lower_text:
        reason = "云平台资源配额不足，需要释放旧资源或提升配额后重试"
    elif "no module named" in lower_text:
        reason = "软件运行库缺失，请使用完整便携包重新解压运行"
    elif "invalidsecuritygroup.inuse" in lower_code or ("security group" in lower_text and "eni" in lower_text):
        reason = "安全组仍被网卡占用，需要先删除该 VPC 内网卡"
    elif "invalidsubnet.inuse" in lower_code or ("subnet" in lower_text and "network interface" in lower_text):
        reason = "子网仍被网卡占用，需要先删除该 VPC 内网卡"
    elif "invalidvpc.inuse" in lower_code:
        reason = "VPC 仍有子网/网卡等依赖，需要先清理 VPC 内资源"
    elif "invalidnetworkinterface" in lower_code and "inuse" in lower_code:
        reason = "网卡仍被实例或云资源占用"
    elif "invalidchargetype" in lower_code:
        reason = "云服务器计费类型不支持直接删除，需要走退订流程"
    elif code or message:
        if code.lower() in generic_resource_codes and not message:
            reason = re.sub(r"\s+", " ", raw)
            reason = re.sub(r"请求ID=.*$", "", reason).strip("；; ")
            reason = reason[:120]
        else:
            reason = f"{code}{('：' + message) if message else ''}"
    else:
        reason = re.sub(r"\s+", " ", raw)
        reason = re.sub(r"请求ID=.*$", "", reason).strip("；; ")
        reason = reason[:120]
    if label and not reason.startswith(f"{label}:") and not reason.startswith(f"{label}："):
        return f"{label}: {reason}"
    return reason


def _should_append_failure_notice(text: str) -> bool:
    """只有明确终止失败或失败原因行才提示导出日志，避免临时重试误导用户。"""
    if not _is_failure_line(text):
        return False
    if _is_item_failure_detail(text):
        return False
    if _is_non_blocking_failure_line(text):
        return False
    if not _has_specific_failure_reason(text):
        return False
    if _matches_any(_NOTICE_TERMINAL_PATTERNS, text):
        return True
    if _matches_any(_NOTICE_ACTION_FAILURE_PATTERNS, text):
        return not _has_non_blocking_hint(text)
    return _has_terminal_failure_hint(text)


def should_show_realtime_log_line(line: str) -> bool:
    """判断某一行是否显示在实时日志框，避免临时重试信息误导用户。"""
    if _is_contact_log_line(line):
        return False
    text = line.strip()
    if not text:
        return False
    if _matches_any(_TECHNICAL_ERROR_PATTERNS, text):
        return bool(_technical_realtime_message(text))
    if _matches_any(_STRONG_FAILURE_PATTERNS, text):
        return not _is_non_blocking_failure_line(text)
    if any(keyword in text for keyword in _NOISY_KEYWORDS):
        return False
    if ("复用" in text or "已有" in text) and not any(keyword in text for keyword in _FAILURE_KEYWORDS):
        return True
    has_failure = any(keyword in text for keyword in _FAILURE_KEYWORDS)
    has_non_blocking = _has_non_blocking_hint(text)
    if has_failure and has_non_blocking:
        return _has_terminal_failure_hint(text)
    if has_failure:
        return True
    if has_non_blocking:
        return False
    return any(keyword in text for keyword in _FLOW_KEYWORDS)


def filter_realtime_log_text(text: str) -> str:
    """过滤多行实时日志；返回空字符串表示这一批日志只写文件、不显示到界面。"""
    if not text:
        return ""
    visible: list[str] = []
    notice_needed = False
    for line in text.splitlines():
        line = line.rstrip()
        if _is_item_failure_detail(line):
            cleaned_detail = clean_realtime_display_line(remove_failure_export_notice(line))
            if cleaned_detail:
                visible.append(cleaned_detail)
            continue
        technical_message = _technical_realtime_message(line)
        if technical_message:
            for technical_line in technical_message.splitlines():
                cleaned_technical = clean_realtime_display_line(technical_line)
                if cleaned_technical:
                    visible.append(cleaned_technical)
            if _has_specific_failure_reason(line):
                notice_needed = True
            continue
        if should_show_realtime_log_line(line):
            if _should_append_failure_notice(line):
                notice_needed = True
            cleaned_line = remove_failure_export_notice(line) if has_failure_export_notice(line) else line
            display_line = clean_realtime_display_line(cleaned_line)
            if display_line:
                visible.append(display_line)
    if notice_needed and visible:
        joined = "\n".join(visible)
        if FAILURE_EXPORT_NOTICE not in joined:
            visible.append(FAILURE_EXPORT_NOTICE)
    return "\n".join(visible) + ("\n" if visible else "")


def format_realtime_log_text_without_filter(text: str) -> str:
    """只做界面显示清理，不按关键词隐藏日志行。"""
    if not text:
        return ""
    visible: list[str] = []
    for line in str(text or "").splitlines():
        cleaned_line = remove_failure_export_notice(line) if has_failure_export_notice(line) else line
        display_line = clean_realtime_display_line(cleaned_line)
        if display_line:
            visible.append(display_line)
    return "\n".join(visible) + ("\n" if visible else "")


def is_failure_message(text: str) -> bool:
    """判断一段任务完成文案是否表示失败。"""
    value = clean_realtime_display_line(remove_failure_export_notice(str(text or "")))
    if not value:
        return False
    return _is_failure_line(value) or _has_terminal_failure_hint(value)


def _normalize_failure_reason(reason: str, max_length: int = 180) -> str:
    text = clean_realtime_display_line(remove_failure_export_notice(str(reason or "")))
    text = re.sub(r"^\s*[-*]\s*", "", text).strip()
    text = re.sub(r"\s+", " ", text)
    text = text.strip("。；; ")
    if len(text) > max_length:
        text = text[:max_length].rstrip() + "..."
    return text


def _skip_failure_reason_line(line: str) -> bool:
    text = str(line or "").strip()
    if not text:
        return True
    if text == FAILURE_EXPORT_NOTICE:
        return True
    if re.fullmatch(r"[=\-_]{3,}", text):
        return True
    skip_prefixes = (
        "当前目录:",
        "当前目录：",
        "开始时间:",
        "开始时间：",
        "结束时间:",
        "结束时间：",
        "日志文件:",
        "日志文件：",
        "已导出运行结果:",
        "已导出云服务器绑定 IP:",
        "已生成账号资源查询结果:",
        "已生成退订资源查询结果:",
        "任务已结束，可手动关闭窗口",
        "正在自动刷新",
        "失败项:",
    )
    if any(text.startswith(prefix) for prefix in skip_prefixes):
        return True
    if re.search(r"接口连接中断或暂时失败，自动重试\s+\d+/\d+:\s*HTTP Error 404:\s*Not Found", text):
        return True
    return False


def _explicit_failure_reason_from_line(line: str) -> str:
    cleaned = clean_realtime_display_line(remove_failure_export_notice(line))
    if not cleaned or _skip_failure_reason_line(cleaned):
        return ""

    for marker in ("失败原因", "错误汇总", "地区异常", "最终失败", "任务失败"):
        match = re.search(rf"{marker}\s*[：:]\s*(.+)", cleaned, flags=re.IGNORECASE)
        if match:
            return _normalize_failure_reason(match.group(1))

    if _is_non_blocking_failure_line(cleaned) and not _has_terminal_failure_hint(cleaned):
        return ""
    if _is_generic_failure_summary(cleaned) and not _has_specific_failure_reason(cleaned):
        return ""

    if _is_item_failure_detail(cleaned):
        return _normalize_failure_reason(_compact_failure_reason(cleaned))
    if _has_specific_failure_reason(cleaned):
        return _normalize_failure_reason(_compact_failure_reason(cleaned))
    if _matches_any(_TECHNICAL_ERROR_PATTERNS, cleaned) and is_failure_message(cleaned):
        return _normalize_failure_reason(_compact_failure_reason(cleaned))
    if any(keyword in cleaned for keyword in ("售罄", "配额", "余额", "欠费", "冻结", "鉴权", "权限不足", "不可用")):
        return _normalize_failure_reason(cleaned)
    if re.search(r"(失败|异常|错误|超时|未完成|无法)[^：:\n]{0,28}[：:]\s*\S+", cleaned):
        return _normalize_failure_reason(cleaned)
    return ""


def _fallback_failure_reason_from_line(line: str) -> str:
    cleaned = clean_realtime_display_line(remove_failure_export_notice(line))
    if not cleaned or _skip_failure_reason_line(cleaned):
        return ""
    exit_code = re.search(r"退出码\s*([1-9]\d*)", cleaned)
    if exit_code:
        return f"子任务退出码 {exit_code.group(1)}，实时日志未返回更具体原因"
    if is_failure_message(cleaned) and not re.search(r"失败\s*0(?:[。.\s]|$)", cleaned):
        return _normalize_failure_reason(cleaned)
    return ""


def extract_failure_reason_from_text(text: str, *, max_reasons: int = 3) -> str:
    """从实时日志或弹窗文案尾部提取最适合给用户看的失败原因。"""
    lines = [line for line in str(text or "").splitlines() if line.strip()]
    if not lines:
        return ""

    reasons: list[str] = []
    seen: set[str] = set()
    fallback = ""
    for line in reversed(lines[-220:]):
        reason = _explicit_failure_reason_from_line(line)
        if reason:
            key = reason.casefold()
            if key not in seen:
                seen.add(key)
                reasons.append(reason)
                if len(reasons) >= max(1, max_reasons):
                    break
            continue
        if not fallback:
            fallback = _fallback_failure_reason_from_line(line)

    if reasons:
        return "；".join(reversed(reasons))
    return fallback


def append_failure_reason_to_message(message: str, reason: str) -> str:
    """给失败弹窗追加失败原因，避免重复追加和重复导出提示。"""
    raw = str(message or "").strip()
    if not raw or not is_failure_message(raw):
        return raw
    if not reason or "失败原因" in raw:
        return raw
    had_notice = has_failure_export_notice(raw)
    base = remove_failure_export_notice(raw).strip()
    normalized_reason = _normalize_failure_reason(reason, max_length=260)
    if not normalized_reason or normalized_reason in base:
        return raw
    updated = f"{base}\n\n失败原因：{normalized_reason}"
    if had_notice:
        updated = f"{updated}\n{FAILURE_EXPORT_NOTICE}"
    return updated


class RealtimeLogDisplayFilter:
    """按日志区域记录导出提示是否已显示，避免同一任务重复刷屏。"""

    def __init__(self) -> None:
        self._notice_contexts: set[str] = set()
        self._failure_reasons: dict[str, list[str]] = {}
        self._failure_reason_keys: dict[str, set[str]] = {}

    def reset(self, context: str | None = None) -> None:
        if context is None:
            self._notice_contexts.clear()
            self._failure_reasons.clear()
            self._failure_reason_keys.clear()
            return
        key = str(context or "default")
        self._notice_contexts.discard(key)
        self._failure_reasons.pop(key, None)
        self._failure_reason_keys.pop(key, None)

    def _record_failure_reason(self, context: str, line: str) -> None:
        reason = _compact_failure_reason(line)
        if not reason:
            return
        seen = self._failure_reason_keys.setdefault(context, set())
        if reason in seen:
            return
        seen.add(reason)
        self._failure_reasons.setdefault(context, []).append(reason)

    def limit_notice(self, filtered: str, context: str = "default") -> str:
        key = str(context or "default")
        if not filtered:
            return filtered

        visible: list[str] = []
        notice_needed = False
        for raw_line in str(filtered or "").splitlines():
            line = raw_line.rstrip()
            stripped = line.strip()
            if not stripped:
                continue
            if stripped == FAILURE_EXPORT_NOTICE:
                notice_needed = True
                continue
            if stripped == "失败项:":
                continue
            if _is_item_failure_detail(stripped):
                self._record_failure_reason(key, stripped)
                continue
            cleaned = remove_failure_export_notice(line) if has_failure_export_notice(line) else line
            if _is_final_failure_summary(cleaned):
                display_line = clean_realtime_display_line(cleaned)
                if display_line:
                    visible.append(display_line)
                reasons = self._failure_reasons.get(key) or []
                if reasons:
                    visible.append("失败原因：")
                    for reason in reasons[:8]:
                        display_reason = clean_realtime_display_line(reason)
                        if display_reason:
                            visible.append(f"- {display_reason}")
                    if len(reasons) > 8:
                        visible.append(f"- 另有 {len(reasons) - 8} 项失败，完整信息见导出日志。")
                notice_needed = True
                continue
            display_line = clean_realtime_display_line(cleaned)
            if display_line:
                visible.append(display_line)

        if notice_needed and visible and key not in self._notice_contexts:
            visible.append(FAILURE_EXPORT_NOTICE)
            self._notice_contexts.add(key)
        return "\n".join(visible) + ("\n" if visible else "")

    def filter(self, text: str, context: str = "default") -> str:
        """过滤原始日志，并限制同一日志区域只显示一次导出提示。"""
        return self.limit_notice(filter_realtime_log_text(text), context)
