diff --git a/.env b/.env
new file mode 100644
index 0000000..262d663
--- /dev/null
+++ b/.env
@@ -0,0 +1,11 @@
+log_level=DEBUG
+
+# sqlite连接方式：sqlite:///
+database_url=sqlite:///./logs.db
+
+wechat_webhook_url=https://qyapi.weixin.qq.com/cgi-bin/webhook/send
+wechat_webhook_key=f09eb098-f709-4fc6-83b5-4a08a8431b8f
+
+openobserve_url=http://172.29.100.41:5080/api/default/_search
+openobserve_username=developer@estun.com
+openobserve_password=developer123
diff --git a/.idea/.gitignore b/.idea/.gitignore
new file mode 100644
index 0000000..13566b8
--- /dev/null
+++ b/.idea/.gitignore
@@ -0,0 +1,8 @@
+# Default ignored files
+/shelf/
+/workspace.xml
+# Editor-based HTTP Client requests
+/httpRequests/
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml
diff --git a/.idea/inspectionProfiles/Project_Default.xml b/.idea/inspectionProfiles/Project_Default.xml
new file mode 100644
index 0000000..3dce9c6
--- /dev/null
+++ b/.idea/inspectionProfiles/Project_Default.xml
@@ -0,0 +1,12 @@
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.idea/inspectionProfiles/profiles_settings.xml b/.idea/inspectionProfiles/profiles_settings.xml
new file mode 100644
index 0000000..105ce2d
--- /dev/null
+++ b/.idea/inspectionProfiles/profiles_settings.xml
@@ -0,0 +1,6 @@
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.idea/log-alert-service.iml b/.idea/log-alert-service.iml
new file mode 100644
index 0000000..b9dcfb8
--- /dev/null
+++ b/.idea/log-alert-service.iml
@@ -0,0 +1,14 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.idea/misc.xml b/.idea/misc.xml
new file mode 100644
index 0000000..d5b6360
--- /dev/null
+++ b/.idea/misc.xml
@@ -0,0 +1,7 @@
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.idea/modules.xml b/.idea/modules.xml
new file mode 100644
index 0000000..d88adc6
--- /dev/null
+++ b/.idea/modules.xml
@@ -0,0 +1,8 @@
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.idea/vcs.xml b/.idea/vcs.xml
new file mode 100644
index 0000000..35eb1dd
--- /dev/null
+++ b/.idea/vcs.xml
@@ -0,0 +1,6 @@
+
+
+
+
+
+
\ No newline at end of file
diff --git a/__pycache__/main.cpython-310.pyc b/__pycache__/main.cpython-310.pyc
new file mode 100644
index 0000000..fd24eb6
Binary files /dev/null and b/__pycache__/main.cpython-310.pyc differ
diff --git a/apis/__pycache__/__init__.cpython-310.pyc b/apis/__pycache__/__init__.cpython-310.pyc
new file mode 100644
index 0000000..aa96731
Binary files /dev/null and b/apis/__pycache__/__init__.cpython-310.pyc differ
diff --git a/apis/__pycache__/log.cpython-310.pyc b/apis/__pycache__/log.cpython-310.pyc
new file mode 100644
index 0000000..656dd7c
Binary files /dev/null and b/apis/__pycache__/log.cpython-310.pyc differ
diff --git a/apis/log.py b/apis/log.py
index 1bfcc1d..7acd199 100644
--- a/apis/log.py
+++ b/apis/log.py
@@ -1,5 +1,7 @@
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+from config.database import get_db
from schemas.log_alert import LogAlertTrigger
from service import log_alert_service
@@ -10,6 +12,11 @@ router = APIRouter(
)
+@router.get("", summary="查询日志告警")
+def query_alert(db: Session = Depends(get_db)):
+ return log_alert_service.query_alert(db)
+
+
@router.post("", summary="触发日志告警")
-def trigger_log_alter(log_alter: LogAlertTrigger):
- return log_alert_service.trigger_alert(log_alter)
+def trigger_log_alter(log_alter: LogAlertTrigger, db: Session = Depends(get_db)) -> bool:
+ return log_alert_service.trigger_alert(log_alter, db)
diff --git a/config/__pycache__/database.cpython-310.pyc b/config/__pycache__/database.cpython-310.pyc
new file mode 100644
index 0000000..0c1b3f5
Binary files /dev/null and b/config/__pycache__/database.cpython-310.pyc differ
diff --git a/config/__pycache__/setting.cpython-310.pyc b/config/__pycache__/setting.cpython-310.pyc
new file mode 100644
index 0000000..f918920
Binary files /dev/null and b/config/__pycache__/setting.cpython-310.pyc differ
diff --git a/config/setting.py b/config/setting.py
index 1ee8784..63c5c1a 100644
--- a/config/setting.py
+++ b/config/setting.py
@@ -2,17 +2,16 @@ from pydantic_settings import BaseSettings
class Settings(BaseSettings):
- # sqlite连接方式：sqlite:///
- database_url: str = "sqlite:///./logs.db"
- wechat_webhook: str = ""
- smtp_server: str = ""
- smtp_port: int = 587
- smtp_user: str = ""
- smtp_password: str = ""
- smtp_sender: str = "log-alert@example.com"
+ log_level: str
+ database_url: str
+ wechat_webhook_url: str
+ wechat_webhook_key: str
+ openobserve_url: str
+ openobserve_username: str
+ openobserve_password: str
class Config:
env_file = ".env"
-settings = Settings()
\ No newline at end of file
+settings = Settings()
diff --git a/id_generator/__init__.py b/id_generator/__init__.py
deleted file mode 100644
index e69de29..0000000
diff --git a/id_generator/__pycache__/__init__.cpython-312.pyc b/id_generator/__pycache__/__init__.cpython-312.pyc
deleted file mode 100644
index 934c604..0000000
Binary files a/id_generator/__pycache__/__init__.cpython-312.pyc and /dev/null differ
diff --git a/id_generator/__pycache__/generator.cpython-312.pyc b/id_generator/__pycache__/generator.cpython-312.pyc
deleted file mode 100644
index 3ef7f83..0000000
Binary files a/id_generator/__pycache__/generator.cpython-312.pyc and /dev/null differ
diff --git a/id_generator/__pycache__/options.cpython-312.pyc b/id_generator/__pycache__/options.cpython-312.pyc
deleted file mode 100644
index 781eaf9..0000000
Binary files a/id_generator/__pycache__/options.cpython-312.pyc and /dev/null differ
diff --git a/id_generator/__pycache__/snowflake.cpython-312.pyc b/id_generator/__pycache__/snowflake.cpython-312.pyc
deleted file mode 100644
index 6fca833..0000000
Binary files a/id_generator/__pycache__/snowflake.cpython-312.pyc and /dev/null differ
diff --git a/id_generator/__pycache__/snowflake_m1.cpython-312.pyc b/id_generator/__pycache__/snowflake_m1.cpython-312.pyc
deleted file mode 100644
index 9e662f0..0000000
Binary files a/id_generator/__pycache__/snowflake_m1.cpython-312.pyc and /dev/null differ
diff --git a/id_generator/generator.py b/id_generator/generator.py
deleted file mode 100644
index fc995d7..0000000
--- a/id_generator/generator.py
+++ /dev/null
@@ -1,38 +0,0 @@
-"""
-雪花算法生成器IdGenerator
-"""
-
-# !/usr/bin/python
-# coding=UTF-8
-
-
-from . import options
-from . import snowflake_m1
-
-
-class DefaultIdGenerator:
- """
- ID生成器
- """
-
- def __init__(self):
- self.snowflake = None
-
- def set_id_generator(self, option: options.IdGeneratorOptions):
- """
- 设置id生成规则信息
- """
-
- if option.base_time < 100000:
- raise ValueError("base time error.")
-
- self.snowflake = snowflake_m1.SnowFlakeM1(option)
-
- def next_id(self) -> int:
- """
- 获取新的UUID
- """
-
- if self.snowflake is None:
- raise ValueError("please set id generator at first.")
- return self.snowflake.next_id()
diff --git a/id_generator/idregister.py b/id_generator/idregister.py
deleted file mode 100644
index 787faac..0000000
--- a/id_generator/idregister.py
+++ /dev/null
@@ -1,134 +0,0 @@
-"""
-worker id generator
-"""
-
-# !/usr/bin/python
-# coding=UTF-8
-
-
-from threading import Thread
-import time
-import logging
-import redis
-
-
-class Register:
- """
- redis封装
- - host 代表redis ip
- - port 代表redis端口
- - max_worker_id worker_id的最大值, 默认为100
- - password redis的密码, 默认为空
- """
-
- def __init__(self, host, port, max_worker_id=100, password=None):
- self.redis_impl = redis.StrictRedis(host=host, port=port, db=0, password=password)
- self.loop_count = 0
- self.max_loop_count = 10
- self.worker_id_expire_time = 15
- self.max_worker_id = max_worker_id
- self.worker_id = -1
- self.is_stop = False
-
- def get_lock(self, key):
- """
- 获取分布式全局锁,并设置过期时间为30秒
- """
-
- if self.redis_impl.setnx(key, 1):
- self.redis_impl.expire(key, 30)
- return True
- if self.redis_impl.ttl(key) < 0:
- self.redis_impl.expire(key, 30)
- return False
-
- def stop(self):
- """
- 退出注册器的线程
- """
-
- self.is_stop = True
-
- def get_worker_id(self):
- """
- 获取全局唯一worker_id, 会创建一个线程给worker id续期
- 失败返回-1
- """
-
- self.loop_count = 0
-
- def extern_life(my_id):
- while 1:
- time.sleep(self.worker_id_expire_time / 3)
- # 是否关闭了
- if self.is_stop:
- return
- # 更新生命周期
- if self.worker_id != my_id:
- break
- try:
- self.redis_impl.expire(
- f"IdGen:WorkerId:Value:{my_id}",
- self.worker_id_expire_time)
- except Exception as exe:
- logging.error(exe)
- continue
-
- self.worker_id = self.__get_next_worker_id()
- if self.worker_id > -1:
- Thread(target=extern_life, args=[self.worker_id]).start()
- return self.worker_id
-
- def __get_next_worker_id(self):
- """
- 获取全局唯一worker id内部实现
- """
-
- cur = self.redis_impl.incrby("IdGen:WorkerId:Index", 1)
-
- def can_reset():
- try:
- reset_value = self.redis_impl.incr("IdGen:WorkerId:Value:Edit")
- return reset_value != 1
- except Exception as ept:
- logging.error(ept)
- return False
-
- def end_reset():
- try:
- self.redis_impl.set("IdGen:WorkerId:Value:Edit", 0)
- except Exception as ept:
- logging.error(ept)
-
- def is_available(worker_id: int):
- try:
- rst = self.redis_impl.get(f"IdGen:WorkerId:Value:{worker_id}")
- return rst != "Y"
- except Exception as ept:
- logging.error(ept)
- return False
-
- if cur > self.max_worker_id:
- if can_reset():
- self.redis_impl.set("IdGen:WorkerId:Index", -1)
- end_reset()
- self.loop_count += 1
-
- if self.loop_count > self.max_loop_count:
- self.loop_count = 0
- return -1
-
- time.sleep(0.2 * self.loop_count)
- return self.__get_next_worker_id()
- time.sleep(0.2)
- return self.__get_next_worker_id()
- if is_available(cur):
- self.redis_impl.setex(
- f"IdGen:WorkerId:Value:{cur}",
- self.worker_id_expire_time,
- "Y"
- )
- self.loop_count = 0
- return cur
-
- return self.__get_next_worker_id()
diff --git a/id_generator/options.py b/id_generator/options.py
deleted file mode 100644
index 8558381..0000000
--- a/id_generator/options.py
+++ /dev/null
@@ -1,43 +0,0 @@
-"""
-生成器IdGenerator配置选项
-"""
-
-# !/usr/bin/python
-# coding=UTF-8
-
-
-class IdGeneratorOptions:
- """
- ID生成器配置
- - worker_id 全局唯一id, 区分不同uuid生成器实例
- - worker_id_bit_length 生成的uuid中worker_id占用的位数
- - seq_bit_length 生成的uuid中序列号占用的位数
- """
-
- def __init__(self, worker_id=0, worker_id_bit_length=6, seq_bit_length=6):
-
- # 雪花计算方法,（1-漂移算法|2-传统算法）, 默认1。目前只实现了1。
- self.method = 1
-
- # 基础时间（ms单位）, 不能超过当前系统时间
- self.base_time = 1582136402000
-
- # 机器码, 必须由外部设定, 最大值 2^worker_id_bit_length-1
- self.worker_id = worker_id
-
- # 机器码位长, 默认值6, 取值范围 [1, 15]（要求：序列数位长+机器码位长不超过22）
- self.worker_id_bit_length = worker_id_bit_length
-
- # 序列数位长, 默认值6, 取值范围 [3, 21]（要求：序列数位长+机器码位长不超过22）
- self.seq_bit_length = seq_bit_length
-
- # 最大序列数（含）, 设置范围 [max_seq_number, 2^seq_bit_length-1]
- # 默认值0, 表示最大序列数取最大值（2^seq_bit_length-1]）
- self.max_seq_number = 0
-
- # 最小序列数（含）, 默认值5, 取值范围 [5, max_seq_number], 每毫秒的前5个序列数对应编号0-4是保留位
- # 其中1-4是时间回拨相应预留位, 0是手工新值预留位
- self.min_seq_number = 5
-
- # 最大漂移次数（含）, 默认2000, 推荐范围500-10000（与计算能力有关）
- self.top_over_cost_count = 2000
diff --git a/id_generator/snowflake.py b/id_generator/snowflake.py
deleted file mode 100644
index ea36cfa..0000000
--- a/id_generator/snowflake.py
+++ /dev/null
@@ -1,20 +0,0 @@
-
-"""
-雪花算法生成器接口声明
-"""
-
-# !/usr/bin/python
-# coding=UTF-8
-
-
-class SnowFlake():
-
- def __init__(self, options):
- self.options = options
-
- def next_id(self) -> int:
- """
- 获取新的UUID
- """
-
- return 0
diff --git a/id_generator/snowflake_m1.py b/id_generator/snowflake_m1.py
deleted file mode 100644
index 596b76d..0000000
--- a/id_generator/snowflake_m1.py
+++ /dev/null
@@ -1,147 +0,0 @@
-"""
-M1生成器
-"""
-
-# !/usr/bin/python
-# coding=UTF-8
-
-import threading
-import time
-from .snowflake import SnowFlake
-from .options import IdGeneratorOptions
-
-
-class SnowFlakeM1(SnowFlake):
- """
- M1规则ID生成器配置
- """
-
- def __init__(self, options: IdGeneratorOptions):
- # 1.base_time
- self.base_time = 1582136402000
- if options.base_time != 0:
- self.base_time = int(options.base_time)
-
- # 2.worker_id_bit_length
- self.worker_id_bit_length = 6
- if options.worker_id_bit_length != 0:
- self.worker_id_bit_length = int(options.worker_id_bit_length)
-
- # 3.worker_id
- self.worker_id = options.worker_id
-
- # 4.seq_bit_length
- self.seq_bit_length = 6
- if options.seq_bit_length != 0:
- self.seq_bit_length = int(options.seq_bit_length)
-
- # 5.max_seq_number
- self.max_seq_number = int(options.max_seq_number)
- if options.max_seq_number <= 0:
- self.max_seq_number = (1 << self.seq_bit_length) - 1
-
- # 6.min_seq_number
- self.min_seq_number = int(options.min_seq_number)
-
- # 7.top_over_cost_count
- self.top_over_cost_count = int(options.top_over_cost_count)
-
- # 8.Others
- self.__timestamp_shift = self.worker_id_bit_length + self.seq_bit_length
- self.__current_seq_number = self.min_seq_number
- self.__last_time_tick: int = 0
- self.__turn_back_time_tick: int = 0
- self.__turn_back_index: int = 0
- self.__is_over_cost = False
- self.___over_cost_count_in_one_term: int = 0
- self.__id_lock = threading.Lock()
-
- def __next_over_cost_id(self) -> int:
- current_time_tick = self.__get_current_time_tick()
- if current_time_tick > self.__last_time_tick:
- self.__last_time_tick = current_time_tick
- self.__current_seq_number = self.min_seq_number
- self.__is_over_cost = False
- self.___over_cost_count_in_one_term = 0
- return self.__calc_id(self.__last_time_tick)
-
- if self.___over_cost_count_in_one_term >= self.top_over_cost_count:
- self.__last_time_tick = self.__get_next_time_tick()
- self.__current_seq_number = self.min_seq_number
- self.__is_over_cost = False
- self.___over_cost_count_in_one_term = 0
- return self.__calc_id(self.__last_time_tick)
-
- if self.__current_seq_number > self.max_seq_number:
- self.__last_time_tick += 1
- self.__current_seq_number = self.min_seq_number
- self.__is_over_cost = True
- self.___over_cost_count_in_one_term += 1
- return self.__calc_id(self.__last_time_tick)
-
- return self.__calc_id(self.__last_time_tick)
-
- def __next_normal_id(self) -> int:
- current_time_tick = self.__get_current_time_tick()
- if current_time_tick < self.__last_time_tick:
- if self.__turn_back_time_tick < 1:
- self.__turn_back_time_tick = self.__last_time_tick - 1
- self.__turn_back_index += 1
- # 每毫秒序列数的前5位是预留位, 0用于手工新值, 1-4是时间回拨次序
- # 支持4次回拨次序（避免回拨重叠导致ID重复）, 可无限次回拨（次序循环使用）。
- if self.__turn_back_index > 4:
- self.__turn_back_index = 1
-
- return self.__calc_turn_back_id(self.__turn_back_time_tick)
-
- # 时间追平时, _TurnBackTimeTick清零
- self.__turn_back_time_tick = min(self.__turn_back_time_tick, 0)
-
- if current_time_tick > self.__last_time_tick:
- self.__last_time_tick = current_time_tick
- self.__current_seq_number = self.min_seq_number
- return self.__calc_id(self.__last_time_tick)
-
- if self.__current_seq_number > self.max_seq_number:
- self.__last_time_tick += 1
- self.__current_seq_number = self.min_seq_number
- self.__is_over_cost = True
- self.___over_cost_count_in_one_term = 1
- return self.__calc_id(self.__last_time_tick)
-
- return self.__calc_id(self.__last_time_tick)
-
- def __calc_id(self, use_time_tick) -> int:
- self.__current_seq_number += 1
- return (
- (use_time_tick << self.__timestamp_shift) +
- (self.worker_id << self.seq_bit_length) +
- self.__current_seq_number
- ) % int(1e64)
-
- def __calc_turn_back_id(self, use_time_tick) -> int:
- self.__turn_back_time_tick -= 1
- return (
- (use_time_tick << self.__timestamp_shift) +
- (self.worker_id << self.seq_bit_length) +
- self.__turn_back_index
- ) % int(1e64)
-
- def __get_current_time_tick(self) -> int:
- return int((time.time_ns() / 1e6) - self.base_time)
-
- def __get_next_time_tick(self) -> int:
- temp_time_ticker = self.__get_current_time_tick()
- while temp_time_ticker <= self.__last_time_tick:
- # 0.001 = 1 mili sec
- time.sleep(0.001)
- temp_time_ticker = self.__get_current_time_tick()
- return temp_time_ticker
-
- def next_id(self) -> int:
- with self.__id_lock:
- if self.__is_over_cost:
- nextid = self.__next_over_cost_id()
- else:
- nextid = self.__next_normal_id()
- return nextid
diff --git a/logs.db b/logs.db
index 763fad3..6c380f9 100644
Binary files a/logs.db and b/logs.db differ
diff --git a/models/__pycache__/base.cpython-312.pyc b/models/__pycache__/base.cpython-312.pyc
deleted file mode 100644
index bb5d27c..0000000
Binary files a/models/__pycache__/base.cpython-312.pyc and /dev/null differ
diff --git a/models/__pycache__/log_alert.cpython-310.pyc b/models/__pycache__/log_alert.cpython-310.pyc
new file mode 100644
index 0000000..a77da78
Binary files /dev/null and b/models/__pycache__/log_alert.cpython-310.pyc differ
diff --git a/models/base.py b/models/base.py
deleted file mode 100644
index d35671e..0000000
--- a/models/base.py
+++ /dev/null
@@ -1,40 +0,0 @@
-from sqlalchemy import Column, BigInteger, DateTime, event
-from sqlalchemy.ext.declarative import declared_attr, declarative_base
-
-from datetime import datetime
-
-from config.database import Base
-from utils.common import camel_to_snake
-from id_generator import options, generator
-
-# https://github.com/yitter/IdGenerator/tree/master/Python
-options = options.IdGeneratorOptions(worker_id=23)
-idgen = generator.DefaultIdGenerator()
-idgen.set_id_generator(options)
-
-
-# 第二层基类：包含ID
-class IdBase(Base):
- __abstract__ = True
-
- id = Column(BigInteger, primary_key=True, index=True)
-
- @declared_attr
- def __tablename__(cls):
- # 自动把数据库实体类名驼峰转为数据库表名下划线
- return camel_to_snake(cls.__name__)
-
-
-# 自动填充id
-@event.listens_for(IdBase, 'before_insert', propagate=True)
-def before_insert_listener(mapper, connection, target):
- if target.id is None:
- target.id = idgen.next_id()
-
-
-# 第二层基类：包含ID和审计字段
-class AuditBase(IdBase):
- __abstract__ = True
-
- create_time = Column(DateTime, nullable=True, default=datetime.now)
- update_time = Column(DateTime, nullable=True, default=datetime.now, onupdate=datetime.now)
diff --git a/models/log_alert.py b/models/log_alert.py
index 6d8f97e..791d8ad 100644
--- a/models/log_alert.py
+++ b/models/log_alert.py
@@ -1,7 +1,8 @@
-from sqlalchemy import Column, Integer, String, DateTime, Text
-from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
+from sqlalchemy import Column, Integer, String, Text, DateTime
+from sqlalchemy.ext.declarative import declarative_base
+
Base = declarative_base()
@@ -9,9 +10,12 @@ class LogAlert(Base):
__tablename__ = "log_alerts"
id = Column(Integer, primary_key=True, index=True)
- timestamp = Column(DateTime)
- log_level = Column(String(20))
+ alter_name = Column(String(50))
+ alter_timestamp = Column(Integer)
+
+ timestamp = Column(Integer)
message = Column(Text)
- source = Column(String(100))
- context = Column(Text) # 存储JSON格式的上下文信息
status = Column(String(20), default="pending") # pending, notified, resolved
+
+ create_time = Column(DateTime, nullable=True, default=datetime.now)
+ update_time = Column(DateTime, nullable=True, default=datetime.now, onupdate=datetime.now)
diff --git a/requirements.txt b/requirements.txt
index 36adcf2..ad92be6 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1 +1 @@
-fastapi~=0.119.0
pydantic~=2.12.2
SQLAlchemy~=2.0.44
requests~=2.32.5
\ No newline at end of file
+fastapi~=0.119.0
pydantic~=2.12.2
SQLAlchemy~=2.0.44
requests~=2.32.5
pydantic-settings~=2.11.0
uvicorn~=0.38.0
\ No newline at end of file
diff --git a/schemas/__pycache__/log_alert.cpython-310.pyc b/schemas/__pycache__/log_alert.cpython-310.pyc
new file mode 100644
index 0000000..2c2bf8a
Binary files /dev/null and b/schemas/__pycache__/log_alert.cpython-310.pyc differ
diff --git a/service/__pycache__/log_alert_service.cpython-310.pyc b/service/__pycache__/log_alert_service.cpython-310.pyc
new file mode 100644
index 0000000..e6c1bb9
Binary files /dev/null and b/service/__pycache__/log_alert_service.cpython-310.pyc differ
diff --git a/service/__pycache__/message_service.cpython-310.pyc b/service/__pycache__/message_service.cpython-310.pyc
new file mode 100644
index 0000000..4fabce2
Binary files /dev/null and b/service/__pycache__/message_service.cpython-310.pyc differ
diff --git a/service/__pycache__/openobserve_service.cpython-310.pyc b/service/__pycache__/openobserve_service.cpython-310.pyc
new file mode 100644
index 0000000..b8a368f
Binary files /dev/null and b/service/__pycache__/openobserve_service.cpython-310.pyc differ
diff --git a/service/log_alert_service.py b/service/log_alert_service.py
index 2174280..1c91d37 100644
--- a/service/log_alert_service.py
+++ b/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)
diff --git a/service/message_service.py b/service/message_service.py
new file mode 100644
index 0000000..c10d057
--- /dev/null
+++ b/service/message_service.py
@@ -0,0 +1,32 @@
+import json
+from enum import Enum
+
+import requests
+
+from config.setting import settings
+
+
+class MessageEnum(Enum):
+ TEXT = 'text'
+ MARKDOWN = 'markdown'
+ MARKDOWN2 = 'markdown2'
+
+
+def send_wechat_message(message_type: MessageEnum, message: str):
+ wechat_message = {}
+ match message_type:
+ case MessageEnum.TEXT:
+ wechat_message = {"msgtype": "text", "text": {"content": message}}
+ case MessageEnum.MARKDOWN:
+ wechat_message = {"msgtype": "markdown", "markdown": {"content": message}}
+ case MessageEnum.MARKDOWN2:
+ wechat_message = {"msgtype": "markdown_v2", "markdown_v2": {"content": message}}
+
+ requests.post(
+ settings.wechat_webhook_url,
+ headers={"Content-Type": "application/json"},
+ params={'key': settings.wechat_webhook_key},
+ data=json.dumps(wechat_message, ensure_ascii=False).encode('utf-8')
+ )
+
+ return True
diff --git a/service/openobserve_service.py b/service/openobserve_service.py
index 961d813..14ef81b 100644
--- a/service/openobserve_service.py
+++ b/service/openobserve_service.py
@@ -1,27 +1,19 @@
import requests
from requests.auth import HTTPBasicAuth
+from config.setting import settings
from schemas.log_alert import OpenobserveQuery
-def get_log():
- url = 'http://192.168.1.7:5080/api/default/_search'
- username = 'njcxx0822@163.com'
- password = '19940822Cxx'
-
+def get_log(start_time: int, end_time: int) -> str:
response = requests.post(
- url,
+ settings.openobserve_url,
headers={
'accept': 'application/json',
'Content-Type': 'application/json'
},
- json={
- "query": OpenobserveQuery(
- start_time=1760773184953420,
- end_time=1760773189953431
- ).model_dump()
- },
- auth=HTTPBasicAuth(username, password)
+ json={"query": OpenobserveQuery(start_time=start_time, end_time=end_time).model_dump()},
+ auth=HTTPBasicAuth(settings.openobserve_username, settings.openobserve_password)
)
- print(response)
+ return response.json()
diff --git a/utils/__pycache__/time.cpython-310.pyc b/utils/__pycache__/time.cpython-310.pyc
new file mode 100644
index 0000000..ef68244
Binary files /dev/null and b/utils/__pycache__/time.cpython-310.pyc differ
diff --git a/utils/time.py b/utils/time.py
new file mode 100644
index 0000000..0aac705
--- /dev/null
+++ b/utils/time.py
@@ -0,0 +1,25 @@
+from datetime import datetime
+from typing import Tuple
+
+
+def get_timestamp_range(ts: int, delta_seconds: int = 5, unit: str = "microseconds") -> Tuple[int, int]:
+ """
+ 返回时间戳前后delta_seconds秒的范围（单位与输入一致）
+ """
+ if ts <= 0:
+ return 0, 0
+
+ scale = {
+ "seconds": 1,
+ "milliseconds": 1000,
+ "microseconds": 1_000_000,
+ "nanoseconds": 1_000_000_000
+
+ }.get(unit, 1_000_000) # 默认微秒
+
+ delta = delta_seconds * scale
+ return ts - delta, ts + delta
+
+
+def format_timestamp(timestamp: int) -> str:
+ return datetime.fromtimestamp(timestamp / 1_000_000).strftime("%Y-%m-%d %H:%M:%S")