feat:增加批量插入和查询日志告警功能

service/log_alert_service.py

@@ -1,31 +1,70 @@
-import requests
-from requests.auth import HTTPBasicAuth
+from collections import defaultdict
 
-from schemas.log_alert import LogAlertTrigger, OpenobserveQuery
+from sqlalchemy import insert, select
+from sqlalchemy.orm import Session
+
+from models.log_alert import LogAlert
+from schemas.log_alert import LogAlertTrigger
 from service import openobserve_service
+from service.message_service import send_wechat_message, MessageEnum
+from utils.time import get_timestamp_range, format_timestamp
 
 
-def trigger_alert(log_alter: LogAlertTrigger):
-    #print(log_alter)
-    # openobserve_service.get_log()
+def trigger_alert(log_alter: LogAlertTrigger, db: Session) -> bool:
+    start_time, end_time = get_timestamp_range(log_alter.timestamp, 60)
+    result = openobserve_service.get_log(start_time, end_time)
+    hits = result.get("hits")
 
-    url = 'http://192.168.1.7:5080/api/default/_search'
-    username = 'njcxx0822@163.com'
-    password = '19940822Cxx'
+    # 判断日志是否为空
+    if len(hits) == 0:
+        return False
 
-    response = requests.post(
-        url,
-        headers={
-            'accept': 'application/json',
-            'Content-Type': 'application/json'
-        },
-        json={
-            "query": OpenobserveQuery(
-                start_time=1760773184953420,
-                end_time=1760773189953431
-            ).model_dump()
-        },
-        auth=HTTPBasicAuth(username, password)
+    # 判断日志告警是否已经存在
+    log_db_alter = db.execute(
+        select(LogAlert)
+        .where(LogAlert.alter_timestamp == log_alter.timestamp)
+    ).first()
+
+    if log_db_alter is not None:
+        return False
+
+    # 企业微信通知
+    alter_message = f"🔔 **告警服务**：{log_alter.alter_name} \n 🕒 **告警时间**：{format_timestamp(log_alter.timestamp)}"
+    send_wechat_message(MessageEnum.MARKDOWN2, alter_message)
+
+    # 批量插入
+    return batch_insert(log_alter, hits, db)
+
+
+def batch_insert(log_alter: LogAlertTrigger, hits: [], db: Session) -> bool:
+    stmt = insert(LogAlert).values(
+        [
+            {
+                "alter_name": log_alter.alter_name,
+                "alter_timestamp": log_alter.timestamp,
+                "timestamp": hit["_timestamp"],
+                "message": hit["message"],
+                "status": "pending"
+            }
+            for hit in hits
+        ]
     )
 
-    return response.json().get("hits")
+    db.execute(stmt)
+    db.commit()
+
+    return True
+
+
+def query_alert(db: Session):
+    alerts = db.execute(select(LogAlert)).scalars().all()
+
+    # 按照name和timestamp分类
+    classified = defaultdict(lambda: defaultdict(list))
+    for alert in alerts:
+        classified[alert.alter_name][alert.alter_timestamp].append({
+            "timestamp": format_timestamp(alert.timestamp),
+            "message": alert.message
+        })
+
+    return dict(classified)