refactor: 抽出统一入库管道,合并定时与手动两条重复路径
问题:app.py:auto_monitor_job(定时)和 routes/api.py:run_monitor(手动) 各自维护了一份几乎相同但细节不一致的写入逻辑,同一条数据经不同入口落库后 latest_time / source / offset / file_count 可能不同。 - 新增 services/time_utils.py:calculate_offset 下沉到无依赖模块,避免 db_ingest 与 routes.api 互相 import 形成循环。routes/api.py 里 re-export 一次,保证 app.py 原有的 from routes.api import calculate_offset 不失效。 - 新增 services/db_ingest.py:ingest_device_data(),承载全部入库细节,只 add/flush 不 commit,事务边界交给调用方。 - routes/api.py:run_monitor 瘦身为「触发爬虫 -> 调管道 -> 提交」。 - models.py:to_dict() 的 offset 改为读取时用 calculate_offset(latest_time) 实时计算,不再读 offset 列。该列是写入时刻的快照,采集一停摆就整体失真 (库里停在 2026-02-06,offset 却仍显示“当天”)。列保留但已废弃, 不执行 ALTER TABLE DROP COLUMN。 入库语义(db_ingest): - 只有爬虫拿到真实业务时间才更新主表 latest_time,拿不到就保留上一个有效值 (冻结),不再回退到 current_time。 - DeviceHistory 单独用 history_time:主表回答“数据到什么时候”,历史回答 “什么时候采过”。 - 历史表 json_data 改存本次增量切片,不再复制主表那份越滚越大的累积 JSON。 - 主表 json_data 改为覆盖式更新以切断无限膨胀,但保留 APP_OWNED_KEYS (bound_iccid / is_whitelist)—— 这两个键由 /bind_device_card 和 /toggle_whitelist 写入,按原方案直接覆盖会清空所有设备-流量卡绑定。
This commit is contained in:
@ -15,6 +15,11 @@ try:
|
||||
except ImportError:
|
||||
execute_monitor_task = None
|
||||
|
||||
try:
|
||||
from services.db_ingest import ingest_device_data
|
||||
except ImportError:
|
||||
ingest_device_data = None
|
||||
|
||||
try:
|
||||
from services.iot_api import sync_iot_data_service
|
||||
except ImportError:
|
||||
@ -27,21 +32,11 @@ api_bp = Blueprint('api', __name__, url_prefix='/api')
|
||||
# 0. 核心算法区:数据质量分析与辅助函数
|
||||
# =========================================================
|
||||
|
||||
def calculate_offset(latest_time_str):
|
||||
"""
|
||||
计算时间滞后天数
|
||||
用于前端展示设备数据是否过时
|
||||
"""
|
||||
if not latest_time_str or latest_time_str == "N/A":
|
||||
return "从未同步"
|
||||
try:
|
||||
# 兼容处理 2026_01_13 和 2026-01-13 格式
|
||||
clean = str(latest_time_str).split()[0].replace('_', '-')
|
||||
target = datetime.strptime(clean, "%Y-%m-%d").date()
|
||||
diff = (datetime.now().date() - target).days
|
||||
return "当天" if diff == 0 else f"滞后 {diff} 天"
|
||||
except:
|
||||
return "时间解析失败"
|
||||
# calculate_offset 已下沉到 services/time_utils.py。
|
||||
# 原因:services.db_ingest 需要计算 offset,而本模块又要 import db_ingest,
|
||||
# 留在本模块会形成循环导入。这里 re-export 一次,保证
|
||||
# `from routes.api import calculate_offset`(app.py 在用)继续可用。
|
||||
from services.time_utils import calculate_offset # noqa: E402
|
||||
|
||||
|
||||
def check_data_quality(content_data, source_type, data_time_str=None):
|
||||
@ -383,77 +378,14 @@ def run_monitor():
|
||||
|
||||
try:
|
||||
# --- A. 执行爬虫并入库 ---
|
||||
if execute_monitor_task:
|
||||
# 入库细节统一走 services.db_ingest,与定时任务 auto_monitor_job 保持一致,
|
||||
# 避免同一条数据经手动/自动两个入口落库后字段不一致。
|
||||
if execute_monitor_task and ingest_device_data:
|
||||
task_result = execute_monitor_task()
|
||||
if task_result:
|
||||
scraped_list = task_result.get('device_list', [])
|
||||
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
count_crawler = 0
|
||||
for item in scraped_list:
|
||||
d_name = item.get('name')
|
||||
if not d_name: continue
|
||||
|
||||
d_raw = item.get('raw_json', {})
|
||||
source = item.get('source', '')
|
||||
target_time = item.get('target_time')
|
||||
|
||||
if '106' in str(source):
|
||||
try:
|
||||
path_str = d_raw.get('path', '')
|
||||
match = re.search(r'/Data/(\d{4}_\d{2}_\d{2})/\w+_(\d{2}_\d{2}_\d{2})\.csv', path_str)
|
||||
if match:
|
||||
date_part = match.group(1).replace('_', '-')
|
||||
time_part = match.group(2).replace('_', ':')
|
||||
target_time = f"{date_part} {time_part}"
|
||||
except:
|
||||
pass
|
||||
|
||||
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()
|
||||
|
||||
if device.source == 'iot_card':
|
||||
device.source = source
|
||||
|
||||
device.status = item.get('status')
|
||||
device.current_value = item.get('value')
|
||||
device.latest_time = target_time
|
||||
device.check_time = current_time
|
||||
|
||||
# ✅ [核心修改] 获取爬虫返回的文件数量并保存
|
||||
f_count = item.get('num_files', 0)
|
||||
device.file_count = f_count
|
||||
|
||||
old_json = {}
|
||||
try:
|
||||
if device.json_data:
|
||||
old_json = json.loads(device.json_data)
|
||||
except:
|
||||
old_json = {}
|
||||
|
||||
new_json = d_raw if isinstance(d_raw, dict) else item.get('raw_json', {})
|
||||
if isinstance(new_json, dict):
|
||||
old_json.update(new_json)
|
||||
|
||||
device.json_data = json.dumps(old_json, ensure_ascii=False)
|
||||
device.offset = calculate_offset(device.latest_time)
|
||||
|
||||
# ✅ [核心修改] 写入历史记录时包含 file_count
|
||||
new_history = DeviceHistory(
|
||||
device_id=device.id,
|
||||
status=item.get('status'),
|
||||
result_data=item.get('value'),
|
||||
data_time=target_time,
|
||||
json_data=device.json_data,
|
||||
file_count=f_count # 确保历史数据也记录文件数
|
||||
)
|
||||
db.session.add(new_history)
|
||||
count_crawler += 1
|
||||
|
||||
msg_list.append(f"爬虫更新: {count_crawler}")
|
||||
updated, _ = ingest_device_data(scraped_list)
|
||||
msg_list.append(f"爬虫更新: {updated}")
|
||||
else:
|
||||
msg_list.append("爬虫无数据")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user