"""Behaviour detection running inside the main service.

It scans 24h of F5 traffic with Graylog pivot aggregations, scores each IP by
behaviour only (no rule hits, no known list), and stores the result in SQLite.
A background thread keeps it refreshed; the web UI reads the stored result.
"""
import csv
import ipaddress
import json
import os
import statistics
import threading
import time

from .http import GraylogError
from .asn import AsnDB

WINDOW = 86400
INTERNAL = ("10.", "192.168.", "172.16.", "172.17.", "172.18.", "172.19.",
            "172.2", "172.3")

UA_SDK = ("curl", "wget", "python-requests", "python-urllib", "go-http-client",
          "okhttp", "java/", "apache-httpclient", "httpcomponents",
          "libredtail", "axios", "node-fetch", "aiohttp", "scrapy",
          "postmanruntime", "insomnia", "unityplayer", "dart/", "httpie",
          "guzzle", "faraday", "restsharp", "libwww", "eeo/", "academy_widget")
UA_BROWSER = ("chrome", "firefox", "safari", "edg/", "opr/", "gecko/20100101")

CRAWLER_TOKENS = ("sogou", "baiduspider", "googlebot", "bingbot", "yandex",
                  "bytespider", "ahrefs", "semrush", "mj12", "dotbot",
                  "petalbot")

RULE_TITLES = {
    "probe_java": "Spring/中间件探测", "probe_config": "配置/凭证探测",
    "probe_cms": "CMS 探测", "probe_iot": "IoT/路由器探测",
    "probe_devops": "DevOps/监控探测", "probe_webshell": "WebShell/上传探测",
    "exploit_payload": "漏洞利用 payload", "login_attack": "登录失败尝试",
    "login_burst": "登录接口高频", "bad_ua": "扫描器 UA",
    "bot_ua": "僵尸网络 UA", "not_found": "404 探测",
}

ASN_CACHE = {}


def _asn_csv(root):
    geo_dir = os.path.join(root, "data", "geo")
    try:
        names = sorted(name for name in os.listdir(geo_dir)
                       if name.startswith("dbip-asn-lite-") and name.endswith(".csv"))
    except FileNotFoundError:
        return None
    return os.path.join(geo_dir, names[-1]) if names else None


def _load_asn(path):
    if ASN_CACHE:
        return ASN_CACHE
    starts, ranges = [], []
    try:
        with open(path, encoding="utf-8", errors="replace") as fh:
            for row in csv.reader(fh):
                if len(row) < 4 or ":" in row[0]:
                    continue
                try:
                    first = int(ipaddress.IPv4Address(row[0]))
                    last = int(ipaddress.IPv4Address(row[1]))
                except ValueError:
                    continue
                starts.append(first)
                ranges.append((last, row[2], row[3]))
    except (FileNotFoundError, TypeError):
        pass
    order = sorted(range(len(starts)), key=lambda i: starts[i])
    ASN_CACHE.update({"starts": [starts[i] for i in order],
                      "ranges": [ranges[i] for i in order]})
    return ASN_CACHE


def shared_type(org):
    text = (org or "").lower()
    cloud = ("alibaba", "aliyun", "tencent", "huawei cloud", "aws", "amazon",
             "google", "microsoft", "digitalocean", "vultr", "linode", "hetzner",
             "ovh", "wowrack", "idc", "data center", "datacenter", "cloud",
             "hosting", "colo")
    if any(token in text for token in cloud):
        return "hosting"
    if "china mobile" in text or "中国移动" in text or "cmnet" in text:
        return "mobile"
    if ("china unicom" in text or "china telecom" in text or "chinanet" in text
            or "联通" in text or "电信" in text or "broadband" in text):
        return "broadband"
    if "education" in text or "cernet" in text or "教育" in text:
        return "education"
    return "unknown"


ORG_CN = (("china mobile", "中国移动"), ("china unicom", "中国联通"),
          ("china telecom", "中国电信"), ("chinanet", "中国电信"),
          ("cernet", "中国教育网"), ("education", "教育网"),
          ("alibaba", "阿里云"), ("aliyun", "阿里云"), ("huawei cloud", "华为云"),
          ("tencent", "腾讯云"), ("wowrack", "Wowrack 机房"),
          ("vmiss", "VMISS 机房"), ("idc", "IDC 机房"))

TYPE_TO_SHARED = {"mobile": "mobile", "broadband": "broadband",
                  "education": "education", "cloud": "hosting", "idc": "hosting"}

SHARE_LABEL = {"mobile": "移动网络", "broadband": "宽带 ISP",
               "education": "教育网", "hosting": "云主机/机房", "unknown": "未知网络"}


def org_cn(org):
    text = (org or "").lower()
    for token, name in ORG_CN:
        if token in text:
            return name
    return org or "未知"


def asn_info(ip, root):
    cache = _load_asn(_asn_csv(root))
    starts = cache.get("starts") or []
    if not starts:
        return {"asn": "", "org": "未知", "shared": "unknown"}
    try:
        value = int(ipaddress.IPv4Address(ip))
    except ValueError:
        return {"asn": "", "org": "未知", "shared": "unknown"}
    import bisect
    index = bisect.bisect_right(starts, value) - 1
    if index >= 0:
        last, asn, org = cache["ranges"][index]
        if last >= value:
            return {"asn": asn, "org": org, "shared": shared_type(org)}
    return {"asn": "", "org": "未知", "shared": "unknown"}


RANGE_OVERRIDE = None


def pivot(client, series, row_groups, column_groups=None, query="*",
          limit=300, timeout=180000):
    if RANGE_OVERRIDE:
        timerange = {"type": "absolute",
                     "from": RANGE_OVERRIDE["from"],
                     "to": RANGE_OVERRIDE["to"]}
    else:
        timerange = {"type": "relative", "range": WINDOW}
    body = {"queries": [{
        "id": "q1",
        "timerange": timerange,
        "query": {"type": "elasticsearch", "query_string": query},
        "filter": {"type": "or", "filters": [{"type": "stream",
                                               "id": "6ac83059a6878e26256fcacc"}]},
        "search_types": [{
            "type": "pivot", "id": "st1", "series": series,
            "row_groups": [{"type": "values", "field": f, "limit": limit}
                           for f in row_groups],
            "column_groups": column_groups or [], "sort": [], "rollup": True,
        }],
    }], "parameters": []}
    data = client.http.request("POST", "/api/views/search/sync",
                               params={"timeout": timeout}, body=body)
    return data["results"]["q1"]["search_types"]["st1"]


def rows_to_dict(search_type, series_id):
    out = {}
    for row in search_type.get("rows") or []:
        key = row.get("key") or []
        if not key:
            continue
        ip = str(key[0])
        for value in row.get("values") or []:
            if (value.get("key") or [None])[0] == series_id:
                try:
                    out[ip] = float(value.get("value") or 0)
                except (TypeError, ValueError):
                    out[ip] = 0.0
    return out


def hourly_matrix(search_type):
    out = {}
    for row in search_type.get("rows") or []:
        key = row.get("key") or []
        if not key:
            continue
        counts = []
        for value in row.get("values") or []:
            stamp = (value.get("key") or [None])[0]
            if not stamp or "cnt" not in (value.get("key") or []):
                continue
            try:
                counts.append(float(value.get("value") or 0))
            except (TypeError, ValueError):
                pass
        out[str(key[0])] = counts
    return out


def bucket_stats(counts):
    active = [c for c in counts if c > 0]
    if not counts:
        return {"coverage": 0.0, "cv": None, "peak_share": 0.0}
    cv = (statistics.pstdev(active) / statistics.fmean(active)
          if len(active) > 2 and statistics.fmean(active) else None)
    total = sum(counts)
    return {"coverage": len(active) / len(counts) * 100,
            "cv": cv,
            "peak_share": (max(counts) / total * 100) if total else 0.0}


def uri_template(uri):
    import re
    path = str(uri or "").split("?")[0]
    path = re.sub(
        r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
        r"[0-9a-fA-F]{12}", "{uuid}", path)
    path = re.sub(r"[0-9a-fA-F]{16,}", "{hex}", path)
    path = re.sub(r"\d+", "{n}", path)
    return path


def ua_family(ua):
    text = str(ua or "").strip()
    if not text:
        return "empty"
    lower = text.lower()
    if any(token in lower for token in UA_SDK):
        return "sdk"
    if lower.startswith("mozilla/") and any(t in lower for t in UA_BROWSER):
        return "browser"
    if lower.startswith("mozilla/5.0"):
        return "fake-browser"
    return "other"


def ua_garbage(ua_card, top_ua):
    if ua_card > 50:
        return True
    text = str(top_ua or "")
    if not text:
        return False
    printable = sum(1 for char in text if 32 <= ord(char) < 127 or ord(char) > 127)
    return printable / max(1, len(text)) < 0.85


def behavior_score(feat):
    total = feat["total"]
    if total < 10:
        return 0, ["请求过少"]
    score = 0
    signals = []
    pct4 = feat["pct4"]
    if pct4 >= 50:
        score += 30
        signals.append("高失败率 %.0f%%" % pct4)
    elif pct4 >= 30:
        score += 15
        signals.append("失败率 %.0f%%" % pct4)
    if feat["uri_card"] >= 20:
        score += 20
        signals.append("路径分散（%d 个 URI）" % feat["uri_card"])
    elif feat["uri_card"] >= 8:
        score += 8
    if feat["coverage"] <= 30 and feat["peak_share"] >= 25:
        score += 20
        signals.append("短时爆发（覆盖 %.0f%%、峰值 %.0f%%）"
                       % (feat["coverage"], feat["peak_share"]))
    if feat["login_ratio"] >= 20:
        score += 15
        signals.append("登录接口集中 %.0f%%" % feat["login_ratio"])
    if feat["host_card"] >= 2:
        score += 5
        signals.append("多站点（%d 个 Host）" % feat["host_card"])
    if feat["ua_card"] == 1:
        score += 5
    if feat["ua_family"] in ("sdk", "fake-browser", "other"):
        score += 10
        signals.append("自动化 UA（%s）" % feat["ua_family"])
    elif feat["ua_family"] == "browser":
        score -= 5
    if feat.get("four_xx", 0) >= 100 and pct4 >= 20:
        score += 10
        signals.append("失败请求 %d 次" % int(feat["four_xx"]))
    if feat["get_ratio"] < 30:
        score -= 25
        signals.append("非 GET 为主（业务接口）")
    if pct4 < 15 and feat["pct2"] >= 80:
        score -= 20
        signals.append("低失败率（采集/浏览）")
    if (feat["coverage"] >= 80 and feat["cv"] is not None
            and feat["cv"] <= 1.2 and feat["uri_templates"] <= 8):
        score -= 20
        signals.append("长期稳定单接口")
    if feat["uri_templates"] >= 15:
        score += 10
        signals.append("路径模板分散（%d 个）" % feat["uri_templates"])
    if feat["status_known"] < 10:
        score -= 10
        signals.append("状态字段缺失")
    return max(0, min(100, score)), signals


def humanize(item):
    parts = []
    if item["uri_card"] >= 20:
        parts.append("访问了 %s 个不同路径" % format(item["uri_card"], ","))
    if item["four_xx"] >= 20:
        parts.append("产生 %s 次失败请求（失败率 %.0f%%）"
                     % (format(int(item["four_xx"]), ","), item["pct4"]))
    elif item["pct4"] >= 20:
        parts.append("失败率 %.0f%%" % item["pct4"])
    if item["coverage"] <= 30 and item["peak_share"] >= 25:
        parts.append("流量集中在很短的时间内")
    if item["login_ratio"] >= 20:
        parts.append("大量请求集中在登录接口")
    if item["host_card"] >= 2:
        parts.append("同时访问了 %d 个站点" % item["host_card"])
    if item["ua_family"] in ("sdk", "fake-browser", "other"):
        parts.append("使用脚本/非浏览器客户端")
    behavior = item.get("behavior_type") or "异常自动化"
    tail = {"目录扫描": "符合目录扫描特征", "短时扫描": "符合短时扫描特征",
            "登录爆破": "符合登录爆破特征",
            "伪造爬虫 UA": "伪装成搜索引擎爬虫，且请求大量失败",
            "采集爬虫": "是采集爬虫", "异常自动化": "行为高度自动化，不像正常用户"}.get(
                behavior, "行为异常")
    if parts:
        return "24 小时内" + "，".join(parts) + "——" + tail + "。"
    return "24 小时内请求 %d 次，%s。" % (item["total"], tail)


def behavior_type(feat, top_ua):
    tokens = CRAWLER_TOKENS
    if any(token in (top_ua or "").lower() for token in tokens):
        return "伪造爬虫 UA" if feat["pct4"] >= 50 else "采集爬虫"
    if feat["login_ratio"] >= 20:
        return "登录爆破"
    if feat["pct4"] >= 50 and feat["uri_card"] >= 20:
        return "目录扫描"
    if feat["coverage"] <= 30 and feat["peak_share"] >= 25:
        return "短时扫描"
    if feat["uri_templates"] >= 100 and feat["pct2"] >= 70:
        return "内容抓取"
    return "异常自动化"


def empty_feat(ip):
    return {"ip": ip, "total": 0, "uri_card": 0, "ua_card": 0, "host_card": 0,
            "method_card": 0, "get_ratio": 0.0, "login_ratio": 0.0,
            "pct2": 0.0, "pct3": 0.0, "pct4": 0.0, "pct5": 0.0,
            "status_known": 0.0, "coverage": 0.0, "cv": None, "peak_share": 0.0,
            "uri_templates": 0, "ua_family": "", "ua_garbage": False,
            "top_uri": "", "top_ua": "", "four_xx": 0}


class BehaviorAnalyzer:
    def __init__(self, client, store, fields, filter_str=None, root=None,
                 interval=900, window=WINDOW, notifier=None):
        self.client = client
        self.store = store
        self.fields = fields
        self.filter_str = filter_str
        self.root = root or os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        self.interval = max(120, int(interval or 900))
        self.window = int(window or WINDOW)
        self.notifier = notifier
        self.last_run = None
        self.last_error = None
        self.runs = 0
        self.asn = None
        try:
            asn_path = os.path.join(self.root, "data", "geo", "asn.db")
            if os.path.exists(asn_path):
                self.asn = AsnDB(asn_path)
        except Exception:
            self.asn = None
        self.geo = None
        try:
            from .geo import GeoIP
            geo_path = os.path.join(self.root, "data", "geo", "geo.db")
            if os.path.exists(geo_path):
                self.geo = GeoIP(geo_path)
        except Exception:
            self.geo = None

    # ---------- data collection ----------
    def _details(self, ips):
        def run(ip):
            q = 'client_ip:"%s"' % ip
            detail = {"uri": [], "ua": [], "buckets5m": []}
            try:
                st = pivot(self.client, [{"type": "count", "id": "cnt"}],
                           ["uri"], query=q, limit=40)
                for row in st.get("rows") or []:
                    key = row.get("key") or []
                    value = (row.get("values") or [{}])[0].get("value")
                    if key:
                        detail["uri"].append((str(key[0]), int(float(value or 0))))
            except GraylogError:
                pass
            try:
                st = pivot(self.client, [{"type": "count", "id": "cnt"}],
                           ["ua"], query=q, limit=3)
                for row in st.get("rows") or []:
                    key = row.get("key") or []
                    value = (row.get("values") or [{}])[0].get("value")
                    if key:
                        detail["ua"].append((str(key[0]), int(float(value or 0))))
            except GraylogError:
                pass
            try:
                extra = {}
                if RANGE_OVERRIDE:
                    extra = {"from_ts": RANGE_OVERRIDE["from"],
                             "to_ts": RANGE_OVERRIDE["to"]}
                hist = self.client.histogram(query=q, range_seconds=self.window,
                                             interval="5m",
                                             filter_str=self.filter_str, **extra)
                from .client import normalize_histogram
                detail["buckets5m"] = [b["count"]
                                       for b in normalize_histogram(hist)]
            except GraylogError:
                pass
            return ip, detail

        from concurrent.futures import ThreadPoolExecutor
        out = {}
        with ThreadPoolExecutor(max_workers=8) as pool:
            for ip, detail in pool.map(run, ips):
                out[ip] = detail
        return out

    def analyze(self):
        from concurrent.futures import ThreadPoolExecutor
        fields = self.fields
        client = self.client

        st = pivot(client,
                   [{"type": "count", "id": "cnt"},
                    {"type": "card", "field": "uri", "id": "uri_card"},
                    {"type": "card", "field": "ua", "id": "ua_card"},
                    {"type": "card", "field": "host", "id": "host_card"},
                    {"type": "card", "field": "method", "id": "method_card"}],
                   ["client_ip"], limit=600)
        total = rows_to_dict(st, "cnt")
        uri_card = rows_to_dict(st, "uri_card")
        ua_card = rows_to_dict(st, "ua_card")
        host_card = rows_to_dict(st, "host_card")
        method_card = rows_to_dict(st, "method_card")

        status = {}
        for cls in ("2*", "3*", "4*", "5*"):
            status[cls] = rows_to_dict(
                pivot(client, [{"type": "count", "id": "cnt"}], ["client_ip"],
                      query="status:%s" % cls, limit=600), "cnt")
        get_count = rows_to_dict(
            pivot(client, [{"type": "count", "id": "cnt"}], ["client_ip"],
                  query="method:GET", limit=600), "cnt")
        LOGIN_QUERY = ("uri:(\\\\/login* OR \\\\/passport* OR \\\\/captcha* "
                       "OR \\\\/jsxy\\\\/api\\\\/login*)")
        login_count = rows_to_dict(
            pivot(client, [{"type": "count", "id": "cnt"}], ["client_ip"],
                  query=LOGIN_QUERY, limit=600), "cnt")
        hourly = hourly_matrix(pivot(
            client, [{"type": "count", "id": "cnt"}], ["client_ip"],
            column_groups=[{"type": "time", "field": "timestamp", "interval": "1h"}],
            limit=600))
        four_xx = rows_to_dict(
            pivot(client, [{"type": "count", "id": "cnt"}], ["client_ip"],
                  query="status:4*", limit=800), "cnt")

        def build(ip, count, uris, uas, hosts, methods, status_map, get_map,
                  login_map, hourly_map):
            hour = bucket_stats(hourly_map.get(ip, []))
            sum_status = sum(status_map[cls].get(ip, 0) for cls in status_map)
            feat = empty_feat(ip)
            feat.update({
                "total": int(count),
                "uri_card": int(uris.get(ip, 0)),
                "ua_card": int(uas.get(ip, 0)),
                "host_card": int(hosts.get(ip, 0)),
                "method_card": int(methods.get(ip, 0)),
                "get_ratio": get_map.get(ip, 0) / count * 100 if count else 0.0,
                "login_ratio": login_map.get(ip, 0) / count * 100 if count else 0.0,
                "pct2": status_map["2*"].get(ip, 0) / max(1, sum_status) * 100,
                "pct3": status_map["3*"].get(ip, 0) / max(1, sum_status) * 100,
                "pct4": status_map["4*"].get(ip, 0) / max(1, sum_status) * 100,
                "pct5": status_map["5*"].get(ip, 0) / max(1, sum_status) * 100,
                "status_known": (sum_status / count * 100) if count else 0.0,
                "coverage": hour["coverage"], "cv": hour["cv"],
                "peak_share": hour["peak_share"],
                "four_xx": status_map["4*"].get(ip, 0),
            })
            return feat

        base = {}
        for ip, count in total.items():
            if count <= 0:
                continue
            base[ip] = build(ip, count, uri_card, ua_card, host_card,
                             method_card, status, get_count, login_count, hourly)

        extra = sorted(set(four_xx) - set(total), key=lambda ip: -four_xx[ip])
        extra = [ip for ip in extra if not ip.startswith(INTERNAL)][:600]
        for offset in range(0, len(extra), 80):
            chunk = extra[offset:offset + 80]
            chunk_query = " OR ".join('client_ip:"%s"' % ip for ip in chunk)
            st = pivot(client,
                       [{"type": "count", "id": "cnt"},
                        {"type": "card", "field": "uri", "id": "uri_card"},
                        {"type": "card", "field": "ua", "id": "ua_card"},
                        {"type": "card", "field": "host", "id": "host_card"},
                        {"type": "card", "field": "method", "id": "method_card"}],
                       ["client_ip"], query=chunk_query, limit=200)
            totals = rows_to_dict(st, "cnt")
            uris = rows_to_dict(st, "uri_card")
            uas = rows_to_dict(st, "ua_card")
            hosts = rows_to_dict(st, "host_card")
            methods = rows_to_dict(st, "method_card")
            chunk_status = {}
            for cls in ("2*", "3*", "4*", "5*"):
                chunk_status[cls] = rows_to_dict(
                    pivot(client, [{"type": "count", "id": "cnt"}], ["client_ip"],
                          query='status:%s AND (%s)' % (cls, chunk_query),
                          limit=200), "cnt")
            chunk_get = rows_to_dict(
                pivot(client, [{"type": "count", "id": "cnt"}], ["client_ip"],
                      query='method:GET AND (%s)' % chunk_query, limit=200), "cnt")
            chunk_login = rows_to_dict(
                pivot(client, [{"type": "count", "id": "cnt"}], ["client_ip"],
                      query='%s AND (%s)' % (LOGIN_QUERY, chunk_query),
                      limit=200), "cnt")
            chunk_hour = hourly_matrix(pivot(
                client, [{"type": "count", "id": "cnt"}], ["client_ip"],
                column_groups=[{"type": "time", "field": "timestamp",
                                "interval": "1h"}], query=chunk_query, limit=200))
            for cls in chunk_status:
                status[cls].update(chunk_status[cls])
            for ip in chunk:
                if ip not in totals or totals[ip] <= 0:
                    continue
                get_count[ip] = chunk_get.get(ip, 0)
                login_count[ip] = chunk_login.get(ip, 0)
                base[ip] = build(ip, totals[ip], uris, uas, hosts, methods,
                                 status, get_count, login_count, chunk_hour)

        shortlist = [ip for ip, feat in base.items()
                     if not ip.startswith(INTERNAL) and feat["total"] >= 10
                     and (feat["pct4"] >= 20 or feat["four_xx"] >= 20
                          or feat["uri_card"] >= 12)]
        shortlist.sort(key=lambda ip: -base[ip]["four_xx"])
        shortlist = shortlist[:260]
        details = self._details(shortlist)

        items = []
        for ip, feat in base.items():
            if ip.startswith(INTERNAL):
                continue
            detail = details.get(ip) or {}
            five = bucket_stats(detail.get("buckets5m") or [])
            uris = detail.get("uri") or []
            uas = detail.get("ua") or []
            feat["uri_templates"] = len({uri_template(uri) for uri, _ in uris})
            feat["top_uri"] = uris[0][0] if uris else ""
            feat["top_ua"] = uas[0][0] if uas else ""
            feat["ua_family"] = ua_family(feat["top_ua"])
            feat["ua_garbage"] = ua_garbage(feat["ua_card"], feat["top_ua"])
            feat["coverage5m"] = five["coverage"]
            score, signals = behavior_score(feat)
            if score < 60 or len(signals) < 2:
                continue
            feat["score"] = score
            feat["signals"] = signals
            feat["behavior_type"] = behavior_type(feat, feat["top_ua"])
            info = (self.asn.lookup(ip) if self.asn is not None else None) or {
                "asn": "", "org": "未知", "brand": "未知",
                "type": "unknown", "icon": "❓"}
            shared = TYPE_TO_SHARED.get(info["type"], "unknown")
            multi = feat["ua_card"] >= 5 or feat["host_card"] >= 10
            if shared in ("mobile", "education"):
                verdict = "仅告警（共享出口）"
                note = ("出口属于%s（%s），大概率有大量用户共享。"
                        % (info["brand"], SHARE_LABEL[shared]))
            elif shared == "broadband" and multi:
                verdict = "仅告警（疑似多用户）"
                note = ("出口属于%s（宽带 ISP），但该 IP 出现 %d 种 UA / %d 个站点，"
                        "疑似多用户共享。" % (info["brand"], feat["ua_card"],
                                              feat["host_card"]))
            elif shared == "broadband":
                verdict = "建议封禁" if score >= 75 else "建议人工确认"
                note = ("出口属于%s（宽带 ISP），但只有 %d 种 UA，更像单台设备。"
                        % (info["brand"], feat["ua_card"]))
            elif shared == "hosting":
                verdict = "建议封禁"
                note = "出口属于%s（云主机/机房）。" % info["brand"]
            else:
                verdict = "建议人工确认"
                note = "出口属于%s，归属不明确。" % info["brand"]
            geo = None
            if self.geo is not None:
                geo = self.geo.lookup(ip)
            feat.update({
                "country_code": (geo or {}).get("country_code", ""),
                "country_name": (geo or {}).get("country_name", ""),
                "region": (geo or {}).get("region", ""),
                "city": (geo or {}).get("city", ""),
                "verdict": verdict, "share_note": note,
                "reason_text": humanize(feat),
                "asn": info["asn"], "asn_org": info["org"],
                "asn_brand": info["brand"], "asn_type": info["type"],
                "asn_icon": info["icon"], "shared_type": shared,
            })
            items.append(feat)
        items.sort(key=lambda item: (-item["score"], -item["total"]))
        return items

    def run_once(self):
        started = time.time()
        try:
            items = self.analyze()
            self.store.replace_behavior_profiles(items)
            self.store.sync_fused()
            self.last_run = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
            self.last_error = None
            self.runs += 1
            print("[%s] 行为分析: %d 个违规 IP，%d 个建议封禁，耗时 %.1fs"
                  % (self.last_run, len(items),
                     sum(1 for item in items if item["verdict"] == "建议封禁"),
                     time.time() - started), flush=True)
            if self.notifier is not None:
                notify_fused(self.store, self.notifier, self.root)
            return items
        except Exception as exc:  # keep the service alive
            self.last_error = str(exc)
            print("[行为分析失败] %s" % exc, flush=True)
            return []

    def run_forever(self):
        time.sleep(8)
        while True:
            self.run_once()
            time.sleep(self.interval)


TYPE_LABEL = {"cloud": "云服务商", "idc": "IDC/机房", "mobile": "移动网络",
              "broadband": "宽带", "education": "教育网",
              "enterprise": "企业/政府", "unknown": "未知"}


def _local_time(value):
    import datetime as _dt
    try:
        stamp = _dt.datetime.fromisoformat(str(value).replace("Z", "+00:00"))
        return stamp.astimezone().strftime("%m-%d %H:%M")
    except (TypeError, ValueError):
        return str(value or "-")


_PUSH_ERROR_AT = 0


def notify_fused(store, notifier, root=None):
    """Push a single merged message for IPs due for a reminder (24h)."""
    due = store.due_pushes(active_hours=24, cooldown_hours=24)
    if not due:
        return 0
    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    lines = ["**建议封禁 · 融合检测**", ""]
    for row in due[:20]:
        geo = " · ".join(part for part in (row.get("country_name"),
                                           row.get("region")) if part)
        lines.append("**%s**%s" % (row["ip"], ("（%s）" % geo) if geo else ""))
        source = ("双来源" if row.get("sources") == "both"
                  else ("行为检测" if row.get("sources") == "behavior"
                        else "特征检测"))
        brand = row.get("asn_brand") or ""
        if brand or row.get("asn"):
            lines.append("归属：%s %s%s"
                         % (row.get("asn_icon") or "", brand or "未知",
                            (" · AS" + str(row["asn"])) if row.get("asn") else ""))
        score = max(float(row.get("feature_score") or 0),
                    float(row.get("behavior_score") or 0))
        lines.append("来源：%s · 发现：%s"
                     % (source, _local_time(row.get("first_banned_at"))))
        lines.append("评分 %.0f · 请求 %s 次"
                     % (score, format(int(row.get("total") or 0), ",")))
        if int(row.get("push_count") or 0) > 0:
            lines.append("⏰ 该 IP 仍在活动（第 %d 次提醒）"
                         % (int(row["push_count"]) + 1))
        if row.get("fused_reason"):
            lines.append("依据：" + row["fused_reason"])
        link = notifier.link_for(row["ip"])
        if link:
            lines.append("[查看详情](%s)" % link)
        lines.append("")
    try:
        notifier.post("\n".join(lines).strip())
    except Exception as exc:
        # 失败时不刷屏：每 5 分钟最多提示一次；下一轮会自动重试
        global _PUSH_ERROR_AT
        now = time.time()
        if now - _PUSH_ERROR_AT > 300:
            print("推送失败（每 5 分钟最多提示一次）: %s" % exc, flush=True)
            _PUSH_ERROR_AT = now
        return 0
    store.mark_pushed([row["ip"] for row in due[:20]], now)
    return len(due[:20])