当前位置：首页 > news >正文

DAMO-YOLO手机检测结果结构化解析：JSON输出格式与数据库存储设计

news 2026/4/20 6:03:25

DAMO-YOLO手机检测结果结构化解析：JSON输出格式与数据库存储设计

1. 引言：从检测框到结构化数据

当你运行一个手机检测模型，看到屏幕上出现一个个红色的方框时，你可能在想：这些检测结果怎么用起来？怎么保存下来？怎么分析统计？

这就是我们今天要解决的问题。基于阿里巴巴的DAMO-YOLO手机检测模型，我们已经有了一个高性能的检测服务——AP@0.5达到88.8%，推理速度仅需3.83毫秒。但检测框只是开始，真正的价值在于如何把这些检测结果变成结构化的数据，方便后续的分析、统计和业务应用。

想象一下这些场景：

商场里统计不同区域的人流密度和手机使用情况
工厂生产线上的手机质量检测和计数
公共场所的手机使用行为分析
零售店铺的顾客手机品牌识别

所有这些应用都需要一个关键环节：把检测结果从"图片上的框"变成"数据库里的记录"。本文将带你一步步实现这个转换过程，从JSON输出格式设计到数据库存储方案，让你真正把检测结果用起来。

2. 理解DAMO-YOLO的检测输出

2.1 原始检测结果长什么样

首先，我们来看看DAMO-YOLO模型直接输出的结果是什么样子。当你调用模型的推理接口时，得到的是一个Python字典结构的数据。

# 这是模型直接输出的原始结果示例 raw_result = { 'scores': [0.95, 0.87, 0.76, 0.65], # 置信度列表 'labels': ['phone', 'phone', 'phone', 'phone'], # 标签列表 'boxes': [ # 边界框坐标列表，格式为[x1, y1, x2, y2] [120, 80, 250, 320], [350, 90, 480, 310], [600, 120, 720, 280], [800, 150, 920, 270] ] }

这个结构很简单，但有几个问题：

缺少时间戳信息
没有图片的元数据
坐标是绝对像素值，没有归一化
缺少检测结果的唯一标识

2.2 我们需要什么样的结构化数据

在实际应用中，我们需要更丰富的信息。一个好的结构化数据应该包含：

检测结果本身：位置、置信度、类别
上下文信息：什么时候检测的、哪张图片、在什么环境下
元数据：图片尺寸、检测模型版本、处理参数
业务信息：检测场景、设备ID、用户ID等

3. 设计JSON输出格式

3.1 基础JSON结构设计

基于实际需求，我设计了一个完整的JSON输出格式。这个格式既包含了检测结果，也包含了丰富的上下文信息。

{ "detection_id": "det_20250206_143025_001", "timestamp": "2025-02-06T14:30:25.123456Z", "model_info": { "model_id": "damo/cv_tinynas_object-detection_damoyolo_phone", "version": "1.0.0", "framework": "PyTorch 2.9.1", "performance": { "ap_50": 0.888, "inference_speed_ms": 3.83 } }, "image_info": { "image_id": "img_001", "filename": "shopping_mall_001.jpg", "width": 1920, "height": 1080, "format": "JPEG", "size_bytes": 2456789 }, "detection_results": { "total_count": 4, "phones": [ { "id": "phone_001", "bbox": { "x1": 120, "y1": 80, "x2": 250, "y2": 320, "width": 130, "height": 240, "area": 31200, "center_x": 185, "center_y": 200 }, "bbox_normalized": { "x1": 0.0625, "y1": 0.0741, "x2": 0.1302, "y2": 0.2963, "width": 0.0677, "height": 0.2222 }, "confidence": 0.95, "label": "phone", "attributes": { "size_category": "large", "aspect_ratio": 0.5417, "position_category": "center_left" } } ] }, "processing_info": { "processing_time_ms": 15.2, "device": "GPU_T4", "batch_size": 1, "threshold": 0.5 }, "scene_info": { "location": "shopping_mall_entrance", "camera_id": "cam_001", "environment": "indoor", "lighting": "normal" } }

3.2 JSON字段详解

让我解释一下这个JSON结构中的关键字段：

检测结果核心字段：

bbox：边界框的绝对坐标（像素值）
bbox_normalized：归一化坐标（0-1之间），方便不同分辨率图片的比较
confidence：置信度，表示模型对这个检测结果的把握程度
attributes：额外的属性信息，比如根据宽高比判断手机大小

上下文信息字段：

detection_id：唯一标识符，方便追踪和去重
timestamp：精确到微秒的时间戳
image_info：图片的详细信息
scene_info：检测场景的信息

模型信息字段：

model_info：使用的模型版本和性能指标
processing_info：处理过程的参数和耗时

3.3 生成结构化JSON的代码实现

现在，让我们看看如何从原始检测结果生成这个结构化的JSON。

import json import uuid from datetime import datetime from typing import Dict, List, Any import numpy as np class PhoneDetectionFormatter: """手机检测结果格式化器""" def __init__(self, model_id: str = "damo/cv_tinynas_object-detection_damoyolo_phone"): self.model_id = model_id self.model_version = "1.0.0" def format_detection_result( self, raw_result: Dict, image_info: Dict, scene_info: Dict = None, threshold: float = 0.5 ) -> Dict: """ 格式化检测结果为结构化JSON 参数: raw_result: 模型原始输出 image_info: 图片信息字典 scene_info: 场景信息字典 threshold: 置信度阈值 返回: 结构化的检测结果字典 """ # 生成唯一ID和时间戳 detection_id = f"det_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}" timestamp = datetime.now().isoformat() # 过滤低置信度的检测结果 filtered_results = self._filter_results(raw_result, threshold) # 构建结构化结果 structured_result = { "detection_id": detection_id, "timestamp": timestamp, "model_info": self._get_model_info(), "image_info": image_info, "detection_results": self._format_detections(filtered_results, image_info), "processing_info": { "processing_time_ms": raw_result.get('processing_time', 0), "device": raw_result.get('device', 'unknown'), "batch_size": 1, "threshold": threshold }, "scene_info": scene_info or {} } return structured_result def _filter_results(self, raw_result: Dict, threshold: float) -> Dict: """过滤低置信度的检测结果""" scores = np.array(raw_result['scores']) boxes = np.array(raw_result['boxes']) labels = np.array(raw_result['labels']) # 保留置信度大于阈值的检测 mask = scores >= threshold return { 'scores': scores[mask].tolist(), 'boxes': boxes[mask].tolist(), 'labels': labels[mask].tolist() } def _get_model_info(self) -> Dict: """获取模型信息""" return { "model_id": self.model_id, "version": self.model_version, "framework": "PyTorch 2.9.1", "performance": { "ap_50": 0.888, "inference_speed_ms": 3.83 } } def _format_detections(self, filtered_result: Dict, image_info: Dict) -> Dict: """格式化检测结果""" phones = [] img_width = image_info.get('width', 1) img_height = image_info.get('height', 1) for i, (score, box, label) in enumerate(zip( filtered_result['scores'], filtered_result['boxes'], filtered_result['labels'] )): x1, y1, x2, y2 = box width = x2 - x1 height = y2 - y1 area = width * height # 计算归一化坐标 x1_norm = x1 / img_width y1_norm = y1 / img_height x2_norm = x2 / img_width y2_norm = y2 / img_height # 判断手机大小类别 size_category = self._get_size_category(width, height, img_width, img_height) # 计算宽高比 aspect_ratio = width / height if height > 0 else 0 # 判断位置 position_category = self._get_position_category(x1, y1, x2, y2, img_width, img_height) phone_info = { "id": f"phone_{i+1:03d}", "bbox": { "x1": int(x1), "y1": int(y1), "x2": int(x2), "y2": int(y2), "width": int(width), "height": int(height), "area": int(area), "center_x": int((x1 + x2) / 2), "center_y": int((y1 + y2) / 2) }, "bbox_normalized": { "x1": round(x1_norm, 4), "y1": round(y1_norm, 4), "x2": round(x2_norm, 4), "y2": round(y2_norm, 4), "width": round(width / img_width, 4), "height": round(height / img_height, 4) }, "confidence": round(float(score), 4), "label": label, "attributes": { "size_category": size_category, "aspect_ratio": round(aspect_ratio, 4), "position_category": position_category } } phones.append(phone_info) return { "total_count": len(phones), "phones": phones } def _get_size_category(self, width: int, height: int, img_width: int, img_height: int) -> str: """根据检测框大小分类""" box_area = width * height img_area = img_width * img_height ratio = box_area / img_area if ratio > 0.1: return "large" elif ratio > 0.05: return "medium" elif ratio > 0.01: return "small" else: return "tiny" def _get_position_category(self, x1: int, y1: int, x2: int, y2: int, img_width: int, img_height: int) -> str: """根据检测框位置分类""" center_x = (x1 + x2) / 2 center_y = (y1 + y2) / 2 # 划分九宫格 col = "left" if center_x < img_width / 3 else ("center" if center_x < 2 * img_width / 3 else "right") row = "top" if center_y < img_height / 3 else ("center" if center_y < 2 * img_height / 3 else "bottom") if col == "center" and row == "center": return "center" else: return f"{row}_{col}" def save_to_json(self, structured_result: Dict, filepath: str): """保存结构化结果到JSON文件""" with open(filepath, 'w', encoding='utf-8') as f: json.dump(structured_result, f, ensure_ascii=False, indent=2) print(f"检测结果已保存到: {filepath}")

4. 数据库存储设计

4.1 为什么需要数据库存储

你可能在想：JSON文件不是挺好的吗？为什么还要存数据库？

让我给你几个理由：

查询效率：从几千条JSON文件里找特定数据很慢，数据库索引可以秒级查询
统计分析：数据库可以方便地做聚合统计，比如"今天检测了多少手机"
数据关联：可以把检测结果和其他业务数据关联起来
数据安全：数据库有事务、备份、恢复机制
并发访问：多个用户可以同时查询数据

4.2 数据库表结构设计

我设计了两个主要的数据表：一个存储检测任务的基本信息，一个存储具体的检测结果。

-- 检测任务表：存储每次检测的基本信息 CREATE TABLE detection_tasks ( task_id VARCHAR(50) PRIMARY KEY, detection_id VARCHAR(50) UNIQUE NOT NULL, timestamp TIMESTAMP NOT NULL, model_id VARCHAR(100) NOT NULL, model_version VARCHAR(20), image_filename VARCHAR(255), image_width INT, image_height INT, total_phones INT DEFAULT 0, avg_confidence DECIMAL(5,4), processing_time_ms INT, device VARCHAR(50), threshold DECIMAL(3,2), location VARCHAR(100), camera_id VARCHAR(50), environment VARCHAR(50), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_timestamp (timestamp), INDEX idx_location (location), INDEX idx_camera_id (camera_id) ); -- 手机检测结果表：存储每个检测到的手机的详细信息 CREATE TABLE phone_detections ( detection_record_id BIGINT AUTO_INCREMENT PRIMARY KEY, task_id VARCHAR(50) NOT NULL, phone_id VARCHAR(20) NOT NULL, bbox_x1 INT NOT NULL, bbox_y1 INT NOT NULL, bbox_x2 INT NOT NULL, bbox_y2 INT NOT NULL, bbox_width INT NOT NULL, bbox_height INT NOT NULL, bbox_area INT NOT NULL, bbox_center_x INT NOT NULL, bbox_center_y INT NOT NULL, bbox_norm_x1 DECIMAL(6,4), bbox_norm_y1 DECIMAL(6,4), bbox_norm_x2 DECIMAL(6,4), bbox_norm_y2 DECIMAL(6,4), bbox_norm_width DECIMAL(6,4), bbox_norm_height DECIMAL(6,4), confidence DECIMAL(5,4) NOT NULL, label VARCHAR(20) DEFAULT 'phone', size_category VARCHAR(10), aspect_ratio DECIMAL(6,4), position_category VARCHAR(20), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (task_id) REFERENCES detection_tasks(task_id), INDEX idx_task_id (task_id), INDEX idx_confidence (confidence), INDEX idx_size_category (size_category), INDEX idx_created_at (created_at) ); -- 统计信息表：用于快速查询常用统计 CREATE TABLE detection_statistics ( stat_date DATE PRIMARY KEY, total_tasks INT DEFAULT 0, total_phones INT DEFAULT 0, avg_phones_per_task DECIMAL(10,2), avg_confidence DECIMAL(5,4), most_common_size VARCHAR(10), most_common_position VARCHAR(20), peak_hour INT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP );

4.3 表结构设计思路

让我解释一下这个设计背后的思考：

检测任务表（detection_tasks）：

存储每次检测的整体信息
包含时间、地点、设备等上下文信息
建立索引加速按时间、地点查询

手机检测结果表（phone_detections）：

存储每个检测到的手机的详细信息
包含坐标、置信度、大小分类等
外键关联到检测任务表
建立索引加速按置信度、大小等查询

统计信息表（detection_statistics）：

预计算的统计信息，避免每次查询都做复杂计算
按天聚合，方便生成日报
自动更新机制

4.4 Python数据库操作实现

现在，让我们看看如何用Python把JSON数据存到数据库。

import mysql.connector from mysql.connector import Error from datetime import datetime from typing import Dict, List, Optional import json class PhoneDetectionDB: """手机检测结果数据库操作类""" def __init__(self, host: str, database: str, user: str, password: str): self.host = host self.database = database self.user = user self.password = password self.connection = None def connect(self): """连接到数据库""" try: self.connection = mysql.connector.connect( host=self.host, database=self.database, user=self.user, password=self.password ) print("数据库连接成功") return True except Error as e: print(f"数据库连接失败: {e}") return False def save_detection_task(self, structured_result: Dict) -> Optional[str]: """保存检测任务到数据库""" if not self.connection: print("数据库未连接") return None try: cursor = self.connection.cursor() # 计算平均置信度 phones = structured_result['detection_results']['phones'] if phones: avg_confidence = sum(p['confidence'] for p in phones) / len(phones) else: avg_confidence = 0 # 插入检测任务 task_query = """ INSERT INTO detection_tasks ( task_id, detection_id, timestamp, model_id, model_version, image_filename, image_width, image_height, total_phones, avg_confidence, processing_time_ms, device, threshold, location, camera_id, environment ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ task_data = ( f"task_{datetime.now().strftime('%Y%m%d%H%M%S')}", structured_result['detection_id'], structured_result['timestamp'], structured_result['model_info']['model_id'], structured_result['model_info']['version'], structured_result['image_info'].get('filename'), structured_result['image_info'].get('width'), structured_result['image_info'].get('height'), structured_result['detection_results']['total_count'], avg_confidence, structured_result['processing_info'].get('processing_time_ms', 0), structured_result['processing_info'].get('device', 'unknown'), structured_result['processing_info'].get('threshold', 0.5), structured_result['scene_info'].get('location'), structured_result['scene_info'].get('camera_id'), structured_result['scene_info'].get('environment') ) cursor.execute(task_query, task_data) task_id = cursor.lastrowid # 插入手机检测结果 if phones: self._save_phone_detections(cursor, task_id, phones) self.connection.commit() cursor.close() print(f"检测任务保存成功，任务ID: {task_id}") return task_id except Error as e: print(f"保存检测任务失败: {e}") self.connection.rollback() return None def _save_phone_detections(self, cursor, task_id: str, phones: List[Dict]): """保存手机检测结果""" detection_query = """ INSERT INTO phone_detections ( task_id, phone_id, bbox_x1, bbox_y1, bbox_x2, bbox_y2, bbox_width, bbox_height, bbox_area, bbox_center_x, bbox_center_y, bbox_norm_x1, bbox_norm_y1, bbox_norm_x2, bbox_norm_y2, bbox_norm_width, bbox_norm_height, confidence, label, size_category, aspect_ratio, position_category ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ detection_data = [] for phone in phones: bbox = phone['bbox'] bbox_norm = phone['bbox_normalized'] attributes = phone['attributes'] detection_data.append(( task_id, phone['id'], bbox['x1'], bbox['y1'], bbox['x2'], bbox['y2'], bbox['width'], bbox['height'], bbox['area'], bbox['center_x'], bbox['center_y'], bbox_norm['x1'], bbox_norm['y1'], bbox_norm['x2'], bbox_norm['y2'], bbox_norm['width'], bbox_norm['height'], phone['confidence'], phone['label'], attributes['size_category'], attributes['aspect_ratio'], attributes['position_category'] )) cursor.executemany(detection_query, detection_data) print(f"保存了 {len(phones)} 个手机检测结果") def query_detections_by_date(self, date_str: str) -> List[Dict]: """按日期查询检测任务""" query = """ SELECT task_id, COUNT(*) as phone_count, AVG(confidence) as avg_confidence, MIN(timestamp) as first_detection, MAX(timestamp) as last_detection FROM detection_tasks dt JOIN phone_detections pd ON dt.task_id = pd.task_id WHERE DATE(timestamp) = %s GROUP BY dt.task_id ORDER BY timestamp DESC """ cursor = self.connection.cursor(dictionary=True) cursor.execute(query, (date_str,)) results = cursor.fetchall() cursor.close() return results def get_statistics_by_location(self, location: str, days: int = 7) -> Dict: """获取指定位置的统计信息""" query = """ SELECT DATE(timestamp) as stat_date, COUNT(DISTINCT dt.task_id) as task_count, COUNT(pd.detection_record_id) as phone_count, AVG(pd.confidence) as avg_confidence, MAX(pd.confidence) as max_confidence, MIN(pd.confidence) as min_confidence, MODE() WITHIN GROUP (ORDER BY pd.size_category) as most_common_size, MODE() WITHIN GROUP (ORDER BY pd.position_category) as most_common_position FROM detection_tasks dt JOIN phone_detections pd ON dt.task_id = pd.task_id WHERE dt.location = %s AND timestamp >= DATE_SUB(CURDATE(), INTERVAL %s DAY) GROUP BY DATE(timestamp) ORDER BY stat_date DESC """ cursor = self.connection.cursor(dictionary=True) cursor.execute(query, (location, days)) results = cursor.fetchall() cursor.close() return { 'location': location, 'period_days': days, 'statistics': results } def close(self): """关闭数据库连接""" if self.connection: self.connection.close() print("数据库连接已关闭")

5. 完整工作流程示例

5.1 从检测到存储的完整流程

现在，让我们把所有的部分组合起来，看看一个完整的工作流程是什么样的。

import time from modelscope.pipelines import pipeline from modelscope.utils.constant import Tasks class PhoneDetectionPipeline: """手机检测完整流程""" def __init__(self, db_config: Dict = None): # 初始化检测模型 self.detector = pipeline( Tasks.domain_specific_object_detection, model='damo/cv_tinynas_object-detection_damoyolo_phone', cache_dir='/root/ai-models', trust_remote_code=True ) # 初始化格式化器 self.formatter = PhoneDetectionFormatter() # 初始化数据库连接 if db_config: self.db = PhoneDetectionDB(**db_config) self.db.connect() else: self.db = None def process_image(self, image_path: str, scene_info: Dict = None) -> Dict: """处理单张图片的完整流程""" start_time = time.time() # 1. 执行检测 print(f"开始检测图片: {image_path}") raw_result = self.detector(image_path) # 2. 获取图片信息 import cv2 img = cv2.imread(image_path) if img is None: raise ValueError(f"无法读取图片: {image_path}") height, width = img.shape[:2] image_info = { 'filename': image_path.split('/')[-1], 'width': width, 'height': height, 'format': image_path.split('.')[-1].upper(), 'size_bytes': os.path.getsize(image_path) } # 3. 添加处理时间信息 processing_time = (time.time() - start_time) * 1000 # 转换为毫秒 raw_result['processing_time'] = processing_time raw_result['device'] = 'GPU_T4' # 假设使用T4 GPU # 4. 格式化结果 structured_result = self.formatter.format_detection_result( raw_result=raw_result, image_info=image_info, scene_info=scene_info, threshold=0.5 ) # 5. 保存到JSON文件 json_filename = f"detection_{structured_result['detection_id']}.json" self.formatter.save_to_json(structured_result, json_filename) # 6. 保存到数据库 if self.db: task_id = self.db.save_detection_task(structured_result) if task_id: print(f"数据已保存到数据库，任务ID: {task_id}") # 7. 打印统计信息 self._print_statistics(structured_result) return structured_result def process_batch(self, image_paths: List[str], scene_info: Dict = None) -> List[Dict]: """批量处理图片""" results = [] for i, image_path in enumerate(image_paths, 1): print(f"处理第 {i}/{len(image_paths)} 张图片: {image_path}") try: result = self.process_image(image_path, scene_info) results.append(result) except Exception as e: print(f"处理图片失败 {image_path}: {e}") continue print(f"批量处理完成，成功处理 {len(results)}/{len(image_paths)} 张图片") return results def _print_statistics(self, result: Dict): """打印检测统计信息""" detection_results = result['detection_results'] print("\n" + "="*50) print("检测统计信息:") print(f"检测ID: {result['detection_id']}") print(f"检测时间: {result['timestamp']}") print(f"图片: {result['image_info']['filename']}") print(f"图片尺寸: {result['image_info']['width']}x{result['image_info']['height']}") print(f"检测到手机数量: {detection_results['total_count']}") if detection_results['phones']: confidences = [p['confidence'] for p in detection_results['phones']] sizes = [p['attributes']['size_category'] for p in detection_results['phones']] positions = [p['attributes']['position_category'] for p in detection_results['phones']] print(f"平均置信度: {sum(confidences)/len(confidences):.3f}") print(f"最高置信度: {max(confidences):.3f}") print(f"最低置信度: {min(confidences):.3f}") print(f"大小分布: {', '.join(f'{s}: {sizes.count(s)}' for s in set(sizes))}") print(f"位置分布: {', '.join(f'{p}: {positions.count(p)}' for p in set(positions))}") print(f"处理耗时: {result['processing_info']['processing_time_ms']:.1f}ms") print("="*50 + "\n") def generate_daily_report(self, date_str: str = None): """生成日报""" if not self.db: print("数据库未配置，无法生成报告") return if date_str is None: date_str = datetime.now().strftime('%Y-%m-%d') print(f"\n生成日报 - {date_str}") print("="*60) # 查询当天的检测任务 tasks = self.db.query_detections_by_date(date_str) if not tasks: print(f"{date_str} 没有检测记录") return total_tasks = len(tasks) total_phones = sum(task['phone_count'] for task in tasks) avg_phones = total_phones / total_tasks if total_tasks > 0 else 0 print(f"检测任务总数: {total_tasks}") print(f"检测到手机总数: {total_phones}") print(f"平均每任务检测数: {avg_phones:.1f}") print(f"最早检测时间: {min(task['first_detection'] for task in tasks)}") print(f"最晚检测时间: {max(task['last_detection'] for task in tasks)}") # 按小时统计 hour_stats = {} for task in tasks: hour = task['first_detection'].hour hour_stats[hour] = hour_stats.get(hour, 0) + task['phone_count'] if hour_stats: peak_hour = max(hour_stats.items(), key=lambda x: x[1])[0] print(f"检测高峰时段: {peak_hour}:00-{peak_hour+1}:00") print(f"高峰时段检测数: {hour_stats[peak_hour]}") print("="*60)

5.2 实际使用示例

让我们看看这个完整流程在实际中怎么用。

# 配置数据库连接 db_config = { 'host': 'localhost', 'database': 'phone_detection', 'user': 'your_username', 'password': 'your_password' } # 初始化检测流程 pipeline = PhoneDetectionPipeline(db_config) # 场景信息 scene_info = { 'location': 'shopping_mall_entrance', 'camera_id': 'cam_001', 'environment': 'indoor', 'lighting': 'normal' } # 处理单张图片 result = pipeline.process_image( image_path='path/to/shopping_mall_001.jpg', scene_info=scene_info ) # 批量处理图片 image_paths = [ 'path/to/image1.jpg', 'path/to/image2.jpg', 'path/to/image3.jpg' ] results = pipeline.process_batch(image_paths, scene_info) # 生成日报 pipeline.generate_daily_report('2025-02-06') # 查询位置统计 if pipeline.db: stats = pipeline.db.get_statistics_by_location('shopping_mall_entrance', days=7) print(f"过去7天商场入口统计:") for stat in stats['statistics']: print(f"日期: {stat['stat_date']}, 任务数: {stat['task_count']}, " f"手机数: {stat['phone_count']}, 平均置信度: {stat['avg_confidence']:.3f}") # 关闭数据库连接 if pipeline.db: pipeline.db.close()