duxingchen/ZDXX
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

数据异常处理

Browse Source
This commit is contained in:
YueL1331
2026-01-12 15:57:34 +08:00
parent 43f049112f
commit e2333ea9b8
3 changed files with 399 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

202
2_1banben/routes/api.py
View File

@ -18,7 +18,131 @@ api_bp = Blueprint('api', __name__, url_prefix='/api')
# =======================
# 0. 认证接口
# 0. 核心算法：数据质量分析 (含夜间免打扰)
# =======================
def check_data_quality(content_data, source_type, data_time_str=None):
"""
在后端快速分析数据质量
:param content_data: 已解析的 JSON 对象 (Dict 或 String)
:param source_type: 设备类型源字符串 (区分 106 或 82)
:param data_time_str: 数据生成时间字符串 (用于判断是否为夜晚)
:return: 'ok' | 'warning' | 'error'
"""
if not content_data:
return 'ok'
# --- [夜间免打扰逻辑] ---
# 物理规律：晚上没有太阳，光谱仪数值低是正常的，不应报错。
# 逻辑：只有在 08:00 - 17:00 之间才检查数值。
if data_time_str and data_time_str != 'N/A':
try:
# 1. 格式清洗
clean_time = str(data_time_str).replace('_', '-')
# 2. 尝试解析时间
dt = None
try:
dt = datetime.strptime(clean_time, "%Y-%m-%d %H:%M:%S")
except ValueError:
try:
dt = datetime.strptime(clean_time, "%Y-%m-%d %H:%M")
except ValueError:
pass
# 3. 如果解析成功，判断小时数
if dt:
start_hour = 8 # 早上 8 点
end_hour = 17 # 下午 5 点
# 如果当前时间 小于8点 或者 大于等于17点，视为夜晚，直接返回正常
if dt.hour < start_hour or dt.hour >= end_hour:
return 'ok'
except Exception:
pass
# ---------------------------
status = 'ok'
source_str = str(source_type)
# === 106 设备逻辑 (CSV 格式) ===
if '106' in source_str:
try:
text_content = ""
if isinstance(content_data, dict):
text_content = content_data.get('content', str(content_data))
else:
text_content = str(content_data)
lines = text_content.split('\n')
for line in lines:
if 'OSIFBeta' not in line: continue
parts = line.split(',')
if len(parts) < 10: continue
try:
int_time = int(parts[2])
except:
continue
# 只有积分时间饱和 (>= 66534) 才检查数值
if int_time >= 66534:
data_points = []
for p in parts[3:]:
try:
data_points.append(float(p))
except:
pass
if not data_points: continue
# 规则1：红色报错 (存在 < 100 的点)
for val in data_points:
if val < 100:
return 'error'
# 规则2：黄色警告 (连续 5 个点在 100-500 之间)
consecutive_warning = 0
for val in data_points:
if 100 <= val <= 500:
consecutive_warning += 1
if consecutive_warning >= 5:
status = 'warning'
else:
consecutive_warning = 0
return status
except Exception:
return 'ok'
# === 82 设备逻辑 (JSON 格式) ===
else:
try:
if not isinstance(content_data, dict):
return 'ok'
specs = content_data.get('downspec', [])
if not specs:
specs = content_data.get('upspec', [])
if not specs: return 'ok'
consecutive_low = 0
for val in specs:
if not isinstance(val, (int, float)): continue
if val < 500:
consecutive_low += 1
if consecutive_low >= 2:
return 'error'
else:
consecutive_low = 0
return 'ok'
except Exception:
return 'ok'
# =======================
# 1. 认证接口
# =======================
@api_bp.route('/login', methods=['POST'])
@ -39,16 +163,32 @@ def login():
# =======================
# 1. 设备概览与详情接口
# 2. 设备概览与详情接口
# =======================
@api_bp.route('/devices_overview', methods=['GET'])
def devices_overview():
try:
devices = Device.query.all()
data_list = [d.to_dict() for d in devices]
data_list = []
for d in devices:
item = d.to_dict()
parsed_content = None
if d.json_data:
try:
parsed_content = json.loads(d.json_data)
except:
parsed_content = None
# 传入 d.latest_time 以启用夜间判断
quality_status = check_data_quality(parsed_content, d.source, d.latest_time)
item['data_quality'] = quality_status
data_list.append(item)
return jsonify({'code': 200, 'data': data_list})
except Exception as e:
print(f"Overview Error: {e}")
return jsonify({'code': 500, 'message': str(e)})
@ -86,7 +226,7 @@ def device_data_by_date():
# =======================
# 2. 维修日志接口
# 3. 维修日志接口
# =======================
@api_bp.route('/logs/list', methods=['GET'])
@ -166,7 +306,7 @@ def delete_log():
# =======================
# 3. 辅助与控制接口 (核心修复逻辑)
# 4. 辅助与控制接口
# =======================
def calculate_offset(latest_time_str):
@ -201,7 +341,6 @@ def run_monitor():
source = item.get('source', '')
target_time = item.get('target_time')
# 处理 106 路径时间
if '106' in str(source):
try:
path_str = d_raw.get('path', '')
@ -213,15 +352,12 @@ def run_monitor():
json_str = json.dumps(d_raw, ensure_ascii=False) if isinstance(d_raw, (dict, list)) else str(d_raw)
# --- 关键修改：先查询，后更新 ---
device = Device.query.filter_by(name=d_name).first()
if not device:
# 只有新设备才初始化静态字段
device = Device(name=d_name, source=source, install_site="")
db.session.add(device)
db.session.flush() # 获取 ID 供 History 使用
db.session.flush()
# 仅更新动态抓取的字段，保留手动填写的 install_site, is_maintaining, is_hidden
device.status = item.get('status')
device.current_value = item.get('value')
device.latest_time = target_time
@ -276,4 +412,48 @@ def toggle_hidden():
device.is_hidden = data.get('is_hidden')
db.session.commit()
return jsonify({'code': 200})
return jsonify({'code': 404}), 404
return jsonify({'code': 404}), 404
# =======================
# 5. 手动添加设备接口 (新增)
# =======================
@api_bp.route('/add_device', methods=['POST'])
def add_device():
data = request.get_json()
name = data.get('name')
site = data.get('site', '')
if not name:
return jsonify({'code': 400, 'message': '必须填写设备名称'}), 400
# 1. 检查是否已存在
existing = Device.query.filter_by(name=name).first()
if existing:
return jsonify({'code': 400, 'message': f'设备 {name} 已存在，无需重复添加'}), 400
try:
# 2. 创建新设备记录
# source 标记为 'manual'，方便以后区分
# status 默认为 'offline' (离线)
# latest_time 默认为 'N/A'
new_device = Device(
name=name,
install_site=site,
source='manual',
status='offline',
current_value='0',
latest_time='N/A',
check_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
json_data='{}',
is_hidden=0,
is_maintaining=0
)
db.session.add(new_device)
db.session.commit()
return jsonify({'code': 200, 'message': '设备添加成功'})
except Exception as e:
db.session.rollback()
return jsonify({'code': 500, 'message': str(e)})