260 lines
11 KiB
Python
260 lines
11 KiB
Python
from __future__ import annotations
|
||
|
||
from collections import Counter
|
||
from typing import Any
|
||
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.api.v1.bgp import BGP_SOURCES
|
||
from app.models.bgp_anomaly import BGPAnomaly
|
||
from app.models.bgp_incident import BGPIncident
|
||
from app.models.bgp_observation import BGPObservation
|
||
from app.schemas.ai import SituationalAnalysisRequest
|
||
from app.services.bgp_collectors import build_bgp_collector_coverage
|
||
from app.services.bgp_enrichment import lookup_prefix_geography
|
||
|
||
|
||
def _format_counter(counter: dict[str, int], empty_text: str = "无") -> str:
|
||
if not counter:
|
||
return empty_text
|
||
return "，".join(f"{key} {value}" for key, value in counter.items())
|
||
|
||
|
||
def _severity_rank(value: str | None) -> int:
|
||
order = {
|
||
"critical": 0,
|
||
"high": 1,
|
||
"medium": 2,
|
||
"low": 3,
|
||
"info": 4,
|
||
}
|
||
return order.get((value or "").lower(), 99)
|
||
|
||
|
||
def _normalize_geo_key(country: str | None, city: str | None) -> str:
|
||
if city and country:
|
||
return f"{city}, {country}"
|
||
return city or country or "未知区域"
|
||
|
||
|
||
def _top_counter_items(counter: Counter[str], limit: int = 5) -> dict[str, int]:
|
||
return {name: count for name, count in counter.most_common(limit) if name}
|
||
|
||
|
||
def _collect_incident_regions(incidents: list[BGPIncident]) -> Counter[str]:
|
||
counter: Counter[str] = Counter()
|
||
for item in incidents:
|
||
for region in item.affected_regions or []:
|
||
if not isinstance(region, dict):
|
||
continue
|
||
counter[_normalize_geo_key(region.get("country"), region.get("city"))] += 1
|
||
return counter
|
||
|
||
|
||
def _collect_collector_regions(collectors: list[dict[str, Any]]) -> Counter[str]:
|
||
counter: Counter[str] = Counter()
|
||
for item in collectors:
|
||
counter[_normalize_geo_key(item.get("country"), item.get("city"))] += int(item.get("recent_24h_observation_count") or 0)
|
||
return counter
|
||
|
||
|
||
def _format_geo_evidence(prefix_geographies: dict[str, dict[str, Any]], limit: int = 6) -> str:
|
||
if not prefix_geographies:
|
||
return "没有命中 prefix geography 证据。"
|
||
|
||
rows = []
|
||
for prefix, item in list(prefix_geographies.items())[:limit]:
|
||
region = _normalize_geo_key(item.get("country"), item.get("city"))
|
||
source = item.get("source") or item.get("geography_mode") or "unknown"
|
||
as_hint = item.get("asn")
|
||
as_name = item.get("as_name")
|
||
as_text = ""
|
||
if as_hint:
|
||
as_text = f" / ASN AS{as_hint}"
|
||
if as_name:
|
||
as_text += f" ({as_name})"
|
||
rows.append(f"{prefix} -> {region} / 来源 {source}{as_text}")
|
||
return "；".join(rows)
|
||
|
||
|
||
async def build_bgp_brief_request(
|
||
db: AsyncSession,
|
||
*,
|
||
incident_limit: int = 5,
|
||
anomaly_limit: int = 6,
|
||
collector_limit: int = 5,
|
||
) -> tuple[SituationalAnalysisRequest, list[str], dict[str, int | str | dict[str, int]]]:
|
||
incidents_result = await db.execute(
|
||
select(BGPIncident)
|
||
.order_by(BGPIncident.created_at.desc(), BGPIncident.id.desc())
|
||
.limit(max(incident_limit, 1))
|
||
)
|
||
anomalies_result = await db.execute(
|
||
select(BGPAnomaly)
|
||
.order_by(BGPAnomaly.created_at.desc(), BGPAnomaly.id.desc())
|
||
.limit(max(anomaly_limit, 1))
|
||
)
|
||
observations_result = await db.execute(
|
||
select(BGPObservation).where(BGPObservation.source.in_(BGP_SOURCES))
|
||
)
|
||
incident_count_result = await db.execute(select(func.count(BGPIncident.id)))
|
||
anomaly_count_result = await db.execute(select(func.count(BGPAnomaly.id)))
|
||
|
||
incidents = incidents_result.scalars().all()
|
||
anomalies = anomalies_result.scalars().all()
|
||
observations = observations_result.scalars().all()
|
||
collectors = await build_bgp_collector_coverage(db, source_filter=BGP_SOURCES)
|
||
|
||
total_incidents = incident_count_result.scalar() or 0
|
||
total_anomalies = anomaly_count_result.scalar() or 0
|
||
total_observations = len(observations)
|
||
active_collectors = [item for item in collectors if item["observation_count"] > 0]
|
||
|
||
incident_status_counts = Counter((item.status or "unknown") for item in incidents)
|
||
incident_severity_counts = Counter((item.severity or "unknown") for item in incidents)
|
||
incident_type_counts = Counter((item.incident_type or "unknown") for item in incidents)
|
||
anomaly_type_counts = Counter((item.anomaly_type or "unknown") for item in anomalies)
|
||
event_type_counts = Counter((item.event_type or "unknown") for item in observations)
|
||
incident_region_counts = _collect_incident_regions(incidents)
|
||
|
||
top_collectors = sorted(
|
||
active_collectors,
|
||
key=lambda item: (
|
||
-int(item["recent_24h_observation_count"]),
|
||
-int(item["observation_count"]),
|
||
str(item["collector"]),
|
||
),
|
||
)[: max(collector_limit, 1)]
|
||
collector_region_counts = _collect_collector_regions(top_collectors)
|
||
|
||
prefix_candidates = sorted(
|
||
{
|
||
prefix
|
||
for item in incidents
|
||
for prefix in (item.affected_prefixes or [])
|
||
if prefix
|
||
}
|
||
| {item.prefix for item in anomalies if item.prefix}
|
||
)
|
||
prefix_geographies = await lookup_prefix_geography(db, prefix_candidates) if prefix_candidates else {}
|
||
geography_region_counts = Counter(
|
||
_normalize_geo_key(item.get("country"), item.get("city"))
|
||
for item in prefix_geographies.values()
|
||
if item.get("country") or item.get("city")
|
||
)
|
||
hotspot_region_counts = geography_region_counts + incident_region_counts
|
||
collector_bias_regions = [
|
||
region
|
||
for region, count in collector_region_counts.most_common(3)
|
||
if count > hotspot_region_counts.get(region, 0)
|
||
]
|
||
|
||
observations_lines: list[str] = [
|
||
f"当前共有 {total_incidents} 起 BGP incidents、{total_anomalies} 条 anomalies、{total_observations} 条原始观测事件。",
|
||
f"活跃观测站 {len(active_collectors)} 个；近 24 小时事件数合计 {sum(int(item['recent_24h_observation_count']) for item in active_collectors)}。",
|
||
f"最近 incidents 严重度分布：{_format_counter(dict(sorted(incident_severity_counts.items(), key=lambda item: _severity_rank(item[0]))))}。",
|
||
f"最近 incidents 状态分布：{_format_counter(dict(incident_status_counts))}。",
|
||
f"最近 incidents 类型分布：{_format_counter(dict(incident_type_counts.most_common(5)))}。",
|
||
f"最近 anomalies 类型分布：{_format_counter(dict(anomaly_type_counts.most_common(6)))}。",
|
||
f"观测事件类型分布：{_format_counter(dict(event_type_counts.most_common(6)))}。",
|
||
]
|
||
|
||
if hotspot_region_counts:
|
||
observations_lines.append(
|
||
"区域热点事实层："
|
||
+ _format_counter(_top_counter_items(hotspot_region_counts, limit=5), empty_text="无明显区域聚集")
|
||
+ "。"
|
||
)
|
||
|
||
if prefix_geographies:
|
||
observations_lines.append("Prefix geography 证据：" + _format_geo_evidence(prefix_geographies))
|
||
|
||
if collector_bias_regions:
|
||
observations_lines.append(
|
||
"观测偏差提示：重点观测站最近 24h 活跃度更集中在 "
|
||
+ "、".join(collector_bias_regions)
|
||
+ "，这些区域的事件升温结论需要结合 prefix geography 与 affected regions 交叉验证。"
|
||
)
|
||
elif top_collectors:
|
||
observations_lines.append(
|
||
"观测偏差提示：当前未发现明显高于区域热点事实层的单一观测站集中区域，但仍需区分 collector coverage 与真实区域风险。"
|
||
)
|
||
|
||
if incidents:
|
||
observations_lines.append(
|
||
"最近 incident 摘要：" + "；".join(
|
||
[
|
||
f"{item.incident_type} / {item.severity} / {item.status}"
|
||
f" / 前缀 {', '.join(item.affected_prefixes[:2]) if item.affected_prefixes else '-'}"
|
||
f" / 观测站 {len(item.affected_collectors or [])} 个"
|
||
for item in incidents
|
||
]
|
||
)
|
||
)
|
||
|
||
if anomalies:
|
||
observations_lines.append(
|
||
"最近 anomaly 摘要：" + "；".join(
|
||
[
|
||
f"{item.anomaly_type} / {item.severity}"
|
||
f" / 前缀 {item.prefix or '-'}"
|
||
f" / ASN {item.new_origin_asn or item.origin_asn or '-'}"
|
||
for item in anomalies
|
||
]
|
||
)
|
||
)
|
||
|
||
if top_collectors:
|
||
observations_lines.append(
|
||
"重点观测站：" + "；".join(
|
||
[
|
||
f"{item['collector']} ({', '.join([part for part in [item.get('city'), item.get('country')] if part]) or '未知位置'})"
|
||
f" / 近24h {item['recent_24h_observation_count']} 条"
|
||
f" / 前缀 {item['prefix_count']} 个"
|
||
for item in top_collectors
|
||
]
|
||
)
|
||
)
|
||
|
||
context = {
|
||
"source": "bgp-overview",
|
||
"incident_total": total_incidents,
|
||
"anomaly_total": total_anomalies,
|
||
"observation_total": total_observations,
|
||
"active_collectors": len(active_collectors),
|
||
"top_incident_types": dict(incident_type_counts.most_common(5)),
|
||
"top_anomaly_types": dict(anomaly_type_counts.most_common(6)),
|
||
"top_event_types": dict(event_type_counts.most_common(6)),
|
||
"region_hotspots": _top_counter_items(hotspot_region_counts, limit=6),
|
||
"incident_regions": _top_counter_items(incident_region_counts, limit=6),
|
||
"collector_bias_regions": collector_bias_regions,
|
||
"prefix_geography_sources": dict(
|
||
Counter(str(item.get("source") or "unknown") for item in prefix_geographies.values()).most_common(5)
|
||
),
|
||
"prefix_geography_sample": {
|
||
prefix: {
|
||
"country": item.get("country"),
|
||
"city": item.get("city"),
|
||
"source": item.get("source"),
|
||
"asn": item.get("asn"),
|
||
"as_name": item.get("as_name"),
|
||
}
|
||
for prefix, item in list(prefix_geographies.items())[:8]
|
||
},
|
||
}
|
||
|
||
return SituationalAnalysisRequest(
|
||
title="BGP 态势 AI 简报",
|
||
objective="基于当前 BGP incidents、anomalies、原始观测事件、观测站覆盖与 prefix geography 证据，生成一份面向操作员的简明态势简报，突出区域热点、观测偏差、当前风险、证据和优先动作。",
|
||
observations=observations_lines,
|
||
constraints=[
|
||
"明确区分事实、推断与建议。",
|
||
"优先指出需要立即关注的高严重度 incident 或异常模式。",
|
||
"需要单独指出哪些区域结论来自 prefix geography / affected regions，哪些可能受 collector coverage 偏差影响。",
|
||
"结论应服务值班排障，不要写成泛泛的模型演示文案。",
|
||
"如果证据不足，要明确指出缺失数据。",
|
||
],
|
||
context=context,
|
||
), observations_lines, context
|