115 lines
3.7 KiB
Python
115 lines
3.7 KiB
Python
import base64
|
||
from datetime import datetime, timezone
|
||
import math
|
||
from zoneinfo import ZoneInfo
|
||
from flask import current_app, json
|
||
import numpy as np
|
||
from werkzeug.security import generate_password_hash, check_password_hash
|
||
import threading # 用于线程安全
|
||
from Crypto.Cipher import AES
|
||
from Crypto.Util.Padding import unpad
|
||
|
||
|
||
# --- 全局状态管理 ---
|
||
# 使用一个简单的标志和锁来确保线程安全
|
||
cluster_ensured_running = False
|
||
cluster_check_lock = threading.Lock()
|
||
|
||
|
||
def encrypt_password(raw_password: str) -> str:
|
||
"""加密密码（使用pbkdf2:sha256算法）"""
|
||
return generate_password_hash(raw_password, method="pbkdf2:sha256", salt_length=16)
|
||
|
||
|
||
def verify_password(encrypted_password: str, raw_password: str) -> bool:
|
||
"""验证密码"""
|
||
return check_password_hash(encrypted_password, raw_password)
|
||
|
||
|
||
def convert_incident_time_format(data_list):
|
||
for item in data_list:
|
||
# 原始数据是Timestamp类型，直接获取
|
||
original_ts = item["incident_time"]
|
||
print('original_ts',type(original_ts))
|
||
# 1. 为Timestamp添加原时区（假设原时间是Asia/Shanghai时区，且原始Timestamp无时区）
|
||
# 注意：如果Timestamp已有时区，直接跳过localize，用tz_convert即可
|
||
if original_ts.tz is None: # 检查是否已有时区
|
||
original_tz_ts = original_ts.tz_localize(ZoneInfo("Asia/Shanghai"))
|
||
else:
|
||
original_tz_ts = original_ts # 已有时区则直接使用
|
||
|
||
# 2. 转换为UTC时区
|
||
utc_ts = original_tz_ts.tz_convert(ZoneInfo("UTC"))
|
||
|
||
# 3. 格式化输出目标格式（添加T、毫秒和时区标识）
|
||
target_str = utc_ts.strftime("%Y-%m-%dT%H:%M:%S") + ".000+0000"
|
||
|
||
# 更新字段
|
||
item["incident_time"] = target_str
|
||
|
||
|
||
# 处理NaN的值
|
||
# processed_data = {}
|
||
for key, value in item.items():
|
||
# 检查当前值是否为NaN（仅针对数值类型）
|
||
if isinstance(value, (int, float)): # 确保是数值类型
|
||
# 判断是否为NaN（同时兼容math.nan和np.nan）
|
||
if math.isnan(value) or np.isnan(value):
|
||
item[key] = None
|
||
else:
|
||
item[key] = value
|
||
else:
|
||
# 非数值类型（如字符串、None等）直接保留
|
||
item[key] = value
|
||
|
||
return data_list
|
||
|
||
# gen5的车码查询前替换为真实的车码
|
||
def mask_vin(real_vin):
|
||
id_str = real_vin[-6:]
|
||
if not id_str.isdigit():
|
||
print('The last six digits of vin:%s are not numbers!' % real_vin)
|
||
return real_vin
|
||
masked_id = ''
|
||
add = 0
|
||
total = 0
|
||
mapping = {
|
||
73: 88,
|
||
79: 89,
|
||
81: 90,
|
||
}
|
||
for i in id_str:
|
||
total += int(i)
|
||
for i in id_str:
|
||
add += total + 2
|
||
c = int(i) + 65 + add % 13
|
||
if c in [73, 79, 81]:
|
||
c = mapping[c]
|
||
masked_id += chr(c)
|
||
|
||
masked_vin = real_vin[:-6] + masked_id[::-1]
|
||
|
||
return masked_vin
|
||
|
||
|
||
def decrypt_code(encrypted_data):
|
||
try:
|
||
raw_key=current_app.config['HEADERS_SECRET_KEY']
|
||
key_bytes=raw_key.encode('utf-8')[:16]
|
||
if len(key_bytes)<16:
|
||
key_bytes=key_bytes.ljust(16,b"0")
|
||
|
||
combined=base64.b64decode(encrypted_data)
|
||
iv_byte=combined[:16]
|
||
ct_byte=combined[16:]
|
||
|
||
ciper=AES.new(key_bytes,AES.MODE_CBC,iv=iv_byte)
|
||
des_padd=ciper.decrypt(ct_byte)
|
||
|
||
desrypt_res=unpad(des_padd,AES.block_size)
|
||
|
||
return desrypt_res.decode('utf-8')
|
||
|
||
except Exception as e:
|
||
print('解析失败')
|
||
return None |