184 lines
7.3 KiB
Python
184 lines
7.3 KiB
Python
import json
|
||
from datetime import datetime
|
||
from typing import Any, Dict, Optional
|
||
|
||
import aiohttp
|
||
from loguru import logger
|
||
|
||
from src.core.config import get_config
|
||
from src.storage.database import Database
|
||
|
||
|
||
class CrontabAlertManager:
|
||
def __init__(self, db: Optional[Database] = None):
|
||
self._cfg = get_config().alert
|
||
self._db = db or Database()
|
||
|
||
@staticmethod
|
||
def build_dedupe_key(crontab_id: int, alert_type: str) -> str:
|
||
return f"crontab:{int(crontab_id)}:{alert_type}"
|
||
|
||
@staticmethod
|
||
def _pick_http_status(severe_http_statuses: list[int]) -> int:
|
||
statuses = [int(item) for item in severe_http_statuses if item is not None]
|
||
if not statuses:
|
||
return 0
|
||
return sorted(statuses)[0]
|
||
|
||
async def record_crawler_result(
|
||
self,
|
||
action: str,
|
||
crontab_id: int,
|
||
crontab_log_id: int,
|
||
task_result: Dict[str, Any],
|
||
output: str,
|
||
):
|
||
await self._db.ensure_alert_tables()
|
||
if action == "error_report":
|
||
return
|
||
|
||
policy = task_result.get("policy", "saved_required")
|
||
candidate_count = int(task_result.get("candidate_count") or 0)
|
||
saved_count = int(task_result.get("saved_count") or 0)
|
||
http_error_count = int(task_result.get("http_error_count") or 0)
|
||
severe_http_statuses = list(task_result.get("severe_http_statuses") or [])
|
||
execution_success = bool(task_result.get("execution_success", True))
|
||
success = bool(task_result.get("success"))
|
||
error_message = str(task_result.get("error_message") or "")
|
||
summary_text = str(task_result.get("summary_text") or "")
|
||
|
||
alert_type = ""
|
||
http_status = 0
|
||
if http_error_count > 0:
|
||
alert_type = "http_error"
|
||
http_status = self._pick_http_status(severe_http_statuses)
|
||
elif not execution_success or (not success and error_message):
|
||
alert_type = "task_exception"
|
||
elif policy == "saved_required" and saved_count <= 0:
|
||
alert_type = "no_data"
|
||
elif policy == "saved_if_candidates" and candidate_count > 0 and saved_count <= 0:
|
||
alert_type = "no_data"
|
||
|
||
if not alert_type:
|
||
await self._db.resolve_crontab_alerts(crontab_id)
|
||
return
|
||
|
||
detail = {
|
||
"summary_text": summary_text[:1000],
|
||
"candidate_count": candidate_count,
|
||
"saved_count": saved_count,
|
||
"http_error_count": http_error_count,
|
||
"severe_http_statuses": severe_http_statuses,
|
||
"error_message": error_message[:1000],
|
||
"output_excerpt": output[:2000],
|
||
}
|
||
await self._db.upsert_crontab_alert(
|
||
crontab_id=crontab_id,
|
||
crontab_log_id=crontab_log_id,
|
||
task_name=str(task_result.get("task_name") or action),
|
||
command=str(task_result.get("command") or "crawler"),
|
||
params=str(task_result.get("params") or action),
|
||
alert_type=alert_type,
|
||
http_status=http_status,
|
||
detail=detail,
|
||
dedupe_key=self.build_dedupe_key(crontab_id, alert_type),
|
||
)
|
||
|
||
async def dispatch_open_alerts(self, limit: int = 50) -> Dict[str, Any]:
|
||
await self._db.ensure_alert_tables()
|
||
alerts = await self._db.get_dispatchable_crontab_alerts(self._cfg.cooldown_seconds, limit=limit)
|
||
if not alerts:
|
||
return {
|
||
"success": True,
|
||
"count": 0,
|
||
"candidate_count": 0,
|
||
"saved_count": 0,
|
||
"summary_text": "无待发送告警",
|
||
}
|
||
|
||
if not self._cfg.dispatch_enabled or not self._cfg.wecom_webhook:
|
||
logger.warning(f"[alert] 企业微信告警未启用或 webhook 未配置，当前待发送 {len(alerts)} 条")
|
||
return {
|
||
"success": True,
|
||
"count": 0,
|
||
"candidate_count": len(alerts),
|
||
"saved_count": 0,
|
||
"summary_text": f"企业微信告警未启用或 webhook 未配置，待发送 {len(alerts)} 条",
|
||
}
|
||
|
||
sent = 0
|
||
failed = 0
|
||
for alert in alerts:
|
||
ok = await self._send_wecom_alert(alert)
|
||
if ok:
|
||
await self._db.mark_crontab_alert_notified(int(alert["id"]))
|
||
sent += 1
|
||
else:
|
||
failed += 1
|
||
|
||
return {
|
||
"success": failed == 0,
|
||
"count": sent,
|
||
"candidate_count": len(alerts),
|
||
"saved_count": sent,
|
||
"failed": failed,
|
||
"summary_text": f"企业微信告警发送完成：成功 {sent} 条，失败 {failed} 条，待发送 {len(alerts)} 条",
|
||
}
|
||
|
||
async def _send_wecom_alert(self, alert: Dict[str, Any]) -> bool:
|
||
message = self._build_markdown(alert)
|
||
payload = {
|
||
"msgtype": "markdown",
|
||
"markdown": {"content": message},
|
||
}
|
||
timeout = aiohttp.ClientTimeout(total=max(3, int(self._cfg.request_timeout)))
|
||
try:
|
||
async with aiohttp.ClientSession(timeout=timeout) as session:
|
||
async with session.post(self._cfg.wecom_webhook, json=payload) as resp:
|
||
body = await resp.text()
|
||
if resp.status != 200:
|
||
logger.warning(f"[alert] 企业微信发送失败 HTTP {resp.status}: {body[:300]}")
|
||
return False
|
||
data = json.loads(body)
|
||
if int(data.get("errcode", -1)) != 0:
|
||
logger.warning(f"[alert] 企业微信返回失败: {body[:300]}")
|
||
return False
|
||
return True
|
||
except Exception as e:
|
||
logger.warning(f"[alert] 企业微信发送异常: {e}")
|
||
return False
|
||
|
||
@staticmethod
|
||
def _build_markdown(alert: Dict[str, Any]) -> str:
|
||
detail_raw = alert.get("detail") or ""
|
||
try:
|
||
detail = json.loads(detail_raw) if isinstance(detail_raw, str) else (detail_raw or {})
|
||
except Exception:
|
||
detail = {"summary_text": str(detail_raw)}
|
||
|
||
def fmt_ts(value: Any) -> str:
|
||
try:
|
||
ts = int(value or 0)
|
||
if ts <= 0:
|
||
return "-"
|
||
return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")
|
||
except Exception:
|
||
return "-"
|
||
|
||
severe = detail.get("severe_http_statuses") or []
|
||
http_text = "/".join(str(item) for item in severe) if severe else (str(alert.get("http_status") or "-"))
|
||
summary = str(detail.get("summary_text") or detail.get("error_message") or "无摘要").replace("\n", " ")
|
||
|
||
return "\n".join([
|
||
"### Sport Era 定时任务告警",
|
||
f"> 任务：`{alert.get('task_name', '')}` / #{alert.get('crontab_id', 0)}",
|
||
f"> 命令：`{alert.get('command', '')} {alert.get('params', '')}`",
|
||
f"> 规则：`{alert.get('expression') or '-'}`",
|
||
f"> 类型：`{alert.get('alert_type', '')}`",
|
||
f"> HTTP：`{http_text}`",
|
||
f"> 保存：`{detail.get('saved_count', 0)}` / 候选：`{detail.get('candidate_count', 0)}` / 请求错误：`{detail.get('http_error_count', 0)}`",
|
||
f"> 首次发生：`{fmt_ts(alert.get('first_seen_at'))}`",
|
||
f"> 最近发生：`{fmt_ts(alert.get('last_seen_at'))}`",
|
||
f"> 摘要：{summary[:1000]}",
|
||
])
|