cxc/hik-push
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
quantulr
2024-01-26 11:31:36 +08:00
parent 6b67995704
commit 077076f234
50 changed files with 140 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
hik_push/read_event.py
View File

@ -5,7 +5,9 @@ from io import BytesIO
import requests
from custom_exception import FeishuAuthException
from event_map_mapping import mapping
from mapping.event_map_mapping import mapping
from mapping.device_map import device_map
from mapping.user_map import user_map
from redis import asyncio as aioredis
@ -27,7 +29,7 @@ def get_access_token(api_id, secret):
# 获取事件的图片
def get_image_url(event):
analysis_key = mapping.get(event['eventType'])['analyze_key']
analysis_key = mapping.get(event['eventType'])['detection_key']
if analysis_key is None:
return None
images = []
@ -65,7 +67,6 @@ def upload_image(url):
# 推送到飞书
def push_to_feishu(data):
push_url = "https://open.feishu.cn/open-apis/message/v4/batch_send/"
# push_url = "http://192.168.20.115:8000/ipaasuat/engine_company/anycross/trigger/callback/MmRhZTE4OTRiYjVkZDQ5YWNmOGRmZDI0NjQ1MTBlODUw/1.0.0"
data = json.dumps(data)
r = requests.post(push_url, data=data, headers=headers).content.decode('utf-8')
data = json.loads(r)
@ -76,92 +77,34 @@ def push_to_feishu(data):
async def read_event():
access_token = get_access_token(app_id, app_secret)
headers['Authorization'] = f"Bearer {access_token}"
redis_client = await aioredis.from_url("redis://localhost")
redis_client = await aioredis.Redis(host="127.0.0.1", port=6379)
while True:
data = await redis_client.brpop("event")
event_json = json.loads(data[1].decode('utf-8'))
data = await redis_client.brpop("hik-sub-event")
sub_json = json.loads(data[1].decode('utf-8'))
try:
events = event_json["params"]["events"]
events = sub_json["params"]["events"]
for event in events:
event_type = mapping.get(event['eventType'], {"type": "未知事件类型"})
print(event_type)
# TODO: 上传图片到飞书链接
images = get_image_url(event)
feishu_image_key = None
if images is not None:
feishu_image_key = upload_image(images[0])
print(feishu_image_key)
# TODO: 根据ip_address获取对应userid
#
# 将eventType ID 替换为中文字符串
event_type = event['eventType']
event_type_str = mapping.get(event_type, "未知事件类型")
event['eventType'] = event_type_str
# ip address
ip_address = event["data"]["ipAddress"]
# channel_name = get_value_by_nested_key(event, ("data", "channelName"))
# 根据 ip_address 获取对应设备名称
device_name = device_map.get(ip_address, "未知设备")
event['deviceName'] = device_name
# print(event['data'][event['data']['eventType']])
# TODO: 根据 ip_address 获取对应 userid
user_ids = user_map.get(ip_address, [])
event['userIds'] = user_ids
# TODO: 请求推送 api
#
print(event, device_name)
# TODO: 向飞书推送消息
push_data = {
"user_ids": [
"8dg79c18"
],
"msg_type": "interactive",
"card": {
"config": {
"wide_screen_mode": True
},
"elements": [
{
"tag": "markdown",
"content": "这里是卡片文本，支持使用markdown标签设置文本格式。例如：\n*斜体* 、**粗体**、~~删除线~~、[文字链接]("
"https://www.feishu.cn)、<at id=all></at>、<font color='red'> 彩色文本 </font>"
},
{
"alt": {
"content": "",
"tag": "plain_text"
},
"img_key": feishu_image_key,
"tag": "img",
"mode": "fit_horizontal",
"compact_width": False
},
{
"tag": "action",
"actions": [
{
"tag": "button",
"text": {
"tag": "plain_text",
"content": "这是跳转按钮"
},
"type": "primary",
"url": "https://feishu.cn"
}
]
}
],
"header": {
"template": "blue",
"title": {
"content": event_type["type"],
"tag": "plain_text"
}
}
}
}
push_to_feishu(push_data)
print(event_type["type"], ip_address)
except FeishuAuthException:
print("飞书登录失败")
# TODO: 重新获取 access token
access_token = get_access_token(app_id, app_secret)
headers['Authorization'] = f"Bearer {access_token}"
await redis_client.lpush("event", json.dumps(event_json))
continue
except Exception as e:
print("error: ", e)
continue
await asyncio.sleep(0.5)
if __name__ == "__main__":