1

.env.example

@@ -64,6 +64,8 @@ ALPHAX_ETHERSCAN_ENABLED=0
 ALPHAX_ETHERSCAN_API_KEY=
 ALPHAX_HELIUS_ENABLED=0
 ALPHAX_HELIUS_API_KEY=
+ALPHAX_SYSTEM_ERROR_FEISHU_ENABLED=0
+ALPHAX_SYSTEM_ERROR_FEISHU_WEBHOOK=
 
 # 邮箱验证码 SMTP 配置。没有配置时，注册验证码只会生成，不会发邮件。
 ASTOCK_SMTP_HOST=

app/db/system_logs.py

@@ -10,6 +10,7 @@ import traceback
 from datetime import datetime, timedelta
 
 from app.db.schema import get_conn
+from app.integrations.system_error_push import push_system_error_alert
 
 
 def _now() -> str:
@@ -91,6 +92,20 @@ def record_system_error(
             )
             log_id = row.fetchone()["id"]
             conn.commit()
+            push_system_error_alert({
+                "id": int(log_id),
+                "created_at": _now(),
+                "level": _truncate(level or "error", 20),
+                "source": _truncate(source or "app", 80),
+                "error_type": _truncate(error_type, 160),
+                "message": _truncate(message, 2000),
+                "stack_trace": _truncate(stack_trace, 60000),
+                "request_method": _truncate(request_method, 16),
+                "request_path": _truncate(request_path, 500),
+                "user_email": _truncate(user_email, 255),
+                "status_code": int(status_code or 0),
+                "fingerprint": fingerprint,
+            })
             return int(log_id)
         finally:
             conn.close()

app/integrations/system_error_push.py

@@ -0,0 +1,71 @@
+"""Feishu webhook transport for system error alerts."""
+
+from __future__ import annotations
+
+import os
+
+import requests
+
+
+def _webhook_url() -> str:
+    return os.getenv("ALPHAX_SYSTEM_ERROR_FEISHU_WEBHOOK", "").strip()
+
+
+def _runtime_env() -> str:
+    return str(os.getenv("ALPHAX_ENV") or "dev").strip().lower() or "dev"
+
+
+def _enabled() -> bool:
+    raw = os.getenv("ALPHAX_SYSTEM_ERROR_FEISHU_ENABLED", "1")
+    return str(raw).strip().lower() in ("1", "true", "yes", "on")
+
+
+def push_system_error_alert(item: dict) -> tuple[bool, object]:
+    """Send one system error alert. This function must never raise."""
+    try:
+        if not _enabled():
+            return False, "system error alert disabled"
+        webhook = _webhook_url()
+        if not webhook:
+            return False, "ALPHAX_SYSTEM_ERROR_FEISHU_WEBHOOK not configured"
+        title = f"AlphaX 系统错误 #{item.get('id') or '--'}"
+        env = _runtime_env()
+        if env not in ("prod", "production"):
+            title = f"[{env.upper()}] {title}"
+        message = str(item.get("message") or "--")
+        stack = str(item.get("stack_trace") or "")
+        if len(stack) > 900:
+            stack = stack[:900] + "\n..."
+        fields = [
+            ("级别", item.get("level") or "--"),
+            ("来源", item.get("source") or "--"),
+            ("类型", item.get("error_type") or "--"),
+            ("状态", item.get("status_code") or "--"),
+            ("路径", item.get("request_path") or "--"),
+            ("时间", item.get("created_at") or "--"),
+            ("指纹", item.get("fingerprint") or "--"),
+        ]
+        content = "\n".join(f"**{label}**: {value}" for label, value in fields)
+        card = {
+            "config": {"wide_screen_mode": True},
+            "header": {
+                "template": "red" if str(item.get("level") or "") == "error" else "yellow",
+                "title": {"tag": "plain_text", "content": title},
+            },
+            "elements": [
+                {"tag": "div", "text": {"tag": "lark_md", "content": content}},
+                {"tag": "div", "text": {"tag": "lark_md", "content": f"**消息**: {message[:900]}"}},
+                {"tag": "note", "elements": [{"tag": "plain_text", "content": "请到日志中心 /logs 查看完整上下文与堆栈。"}]},
+            ],
+        }
+        if stack:
+            card["elements"].append({"tag": "div", "text": {"tag": "lark_md", "content": f"```text\n{stack}\n```"}})
+        resp = requests.post(webhook, json={"msg_type": "interactive", "card": card}, timeout=10)
+        result = resp.json()
+        ok = resp.status_code == 200 and result.get("StatusCode") == 0
+        return ok, result
+    except Exception as exc:
+        return False, str(exc)
+
+
+__all__ = ["push_system_error_alert"]

tests/test_system_error_logs.py

@@ -33,6 +33,24 @@ def test_record_and_query_system_error():
     assert rows["items"][0]["id"] == log_id
 
 
+def test_record_system_error_sends_feishu_alert(monkeypatch):
+    pushed = []
+    monkeypatch.setattr("app.db.system_logs.push_system_error_alert", lambda item: pushed.append(item) or (True, {"StatusCode": 0}))
+
+    log_id = record_system_error(
+        source="web",
+        error_type="RuntimeError",
+        message="alert me",
+        stack_trace="Traceback\nRuntimeError: alert me",
+        request_path="/api/alert",
+    )
+
+    assert log_id > 0
+    assert pushed
+    assert pushed[0]["id"] == log_id
+    assert pushed[0]["message"] == "alert me"
+
+
 def test_admin_system_error_api_uses_local_admin():
     log_id = record_system_error(
         source="scheduler",