""" 第三方平台推送模块 将批次信息推送到第三方平台接口 """ import os import time import json import logging import threading import queue import cv2 import requests from typing import Optional, Dict, Any, List, Callable from dataclasses import dataclass from datetime import datetime from pathlib import Path logger = logging.getLogger(__name__) def _normalize_timestamp(ts: float) -> float: """统一时间戳为秒。CaptureUploader 使用毫秒，其他可能使用秒。""" if ts > 1e12: return ts / 1000.0 return ts def _convert_to_legacy_batch_info(new_info: Dict[str, Any]) -> Dict[str, Any]: """ 把新版 CaptureUploader 生成的 batch_info 转成业务平台要求的 PairedImageSaver 老格式。 参考字段： - batch_id / device_id / project_id / timestamp / datetime - total_persons / ptz_images_count - panorama: local_path / local_path_original / oss_url / oss_url_original - persons: person_index / position(x,y 归一化) / bbox / confidence / ptz_position / ptz_bbox / ptz_image_saved / ptz_image_path / ptz_image_original_path / ptz_oss_url / ptz_oss_url_original - upload_status """ normalized_ts = _normalize_timestamp(new_info.get("timestamp", time.time())) urls = new_info.get("image_urls") or {} image_paths = new_info.get("image_paths") or [] camera_type = new_info.get("camera_type", "panorama") is_ptz = camera_type == "ptz" # image_paths 约定：[original, marked] original_path = image_paths[0] if len(image_paths) > 0 else None marked_path = image_paths[1] if len(image_paths) > 1 else original_path oss_url_original = urls.get("original") or None oss_url_marked = urls.get("marked") or oss_url_original # 读取原图尺寸，用于把人体中心坐标归一化到 0~1 img_w, img_h = 0, 0 if original_path and os.path.exists(original_path): try: img = cv2.imread(original_path) if img is not None: img_h, img_w = img.shape[:2] except Exception: pass ptz_position = new_info.get("ptz_position") or {} persons = [] for i, det in enumerate(new_info.get("detections") or []): bbox = det.get("bbox", [0, 0, 0, 0]) x1, y1, x2, y2 = int(bbox[0]), int(bbox[1]), int(bbox[2]), int(bbox[3]) cx = (x1 + x2) / 2.0 cy = (y1 + y2) / 2.0 position = { "x": round(cx / img_w, 4) if img_w else round(cx, 4), "y": round(cy / img_h, 4) if img_h else round(cy, 4), } person = { "person_index": i, "position": position, "bbox": {"x1": x1, "y1": y1, "x2": x2, "y2": y2}, "confidence": float(det.get("confidence", 0.0)), } # PTZ 流检测时，把同一张 PTZ 图作为每个人的特写图复用 # 始终包含 ptz_position 字段，第三方平台要求必须有 pan/tilt/zoom 数值 if is_ptz: # 使用实际检测时的 PTZ 位置（若无实际位置，用默认值 0/0/1） ptz_pan = ptz_position.get("pan") if isinstance(ptz_position, dict) else 0 ptz_tilt = ptz_position.get("tilt") if isinstance(ptz_position, dict) else 0 ptz_zoom = ptz_position.get("zoom") if isinstance(ptz_position, dict) else 1 person["ptz_position"] = { "pan": ptz_pan if ptz_pan is not None else 0, "tilt": ptz_tilt if ptz_tilt is not None else 0, "zoom": ptz_zoom if ptz_zoom is not None else 1, } person["ptz_bbox"] = {"x1": x1, "y1": y1, "x2": x2, "y2": y2} person["ptz_image_saved"] = bool(marked_path and os.path.exists(marked_path)) person["ptz_image_path"] = marked_path person["ptz_image_original_path"] = original_path person["ptz_oss_url"] = oss_url_marked person["ptz_oss_url_original"] = oss_url_original else: # 全景相机检测时，复用当前检测框作为 ptz_bbox person["ptz_position"] = { "pan": 0, "tilt": 0, "zoom": 1, } person["ptz_bbox"] = {"x1": x1, "y1": y1, "x2": x2, "y2": y2} person["ptz_image_saved"] = bool(marked_path and os.path.exists(marked_path)) person["ptz_image_path"] = marked_path person["ptz_image_original_path"] = original_path person["ptz_oss_url"] = oss_url_marked person["ptz_oss_url_original"] = oss_url_original persons.append(person) ptz_uploaded = is_ptz and bool(oss_url_marked) legacy = { "batch_id": new_info.get("batch_id", ""), "device_id": new_info.get("device_id", ""), "project_id": new_info.get("project_id", ""), "timestamp": normalized_ts, "datetime": datetime.fromtimestamp(normalized_ts).isoformat(), "install_address": new_info.get("install_address", ""), "total_persons": len(persons), "ptz_images_count": len(persons), "panorama": { "local_path": marked_path, "local_path_original": original_path, "oss_url": oss_url_marked, "oss_url_original": oss_url_original, }, "persons": persons, "upload_status": { "panorama_uploaded": bool(oss_url_marked or oss_url_original), "panorama_original_uploaded": bool(oss_url_original), "all_ptz_uploaded": ptz_uploaded, }, } return legacy @dataclass class BatchReport: """批次上报数据""" batch_id: str device_id: str project_id: str timestamp: float batch_info: Dict[str, Any] # batch_info.json 的完整内容 local_path: Optional[str] = None # batch_info.json 本地路径 class ThirdPartyPusher: """ 第三方平台推送器 负责将批次信息推送到配置的第三方平台接口 """ def __init__(self, config: Dict[str, Any] = None): """ 初始化第三方平台推送器 Args: config: 第三方平台配置字典 """ from config import THIRD_PARTY_CONFIG, DEVICE_CONFIG self.config = config or THIRD_PARTY_CONFIG self.device_config = DEVICE_CONFIG # 功能开关 self.enabled = self.config.get('enabled', False) # 平台配置 self.platform_type = self.config.get('platform_type', 'custom') self.base_url = self.config.get('base_url', '') self.api_version = self.config.get('api_version', 'v1') # 认证配置 self.auth_type = self.config.get('auth_type', 'none') self.api_key = self.config.get('api_key', '') self.api_secret = self.config.get('api_secret', '') self.oauth2_config = self.config.get('oauth2', {}) # 接口路径 self.endpoints = self.config.get('endpoints', {}) self.batch_report_url = self.endpoints.get('batch_report', '/api/batch/report') self.heartbeat_url = self.endpoints.get('heartbeat', '/api/device/heartbeat') # 推送控制 self.push_interval = self.config.get('push_interval', 1.0) self.retry_count = self.config.get('retry_count', 3) self.retry_delay = self.config.get('retry_delay', 2.0) self.timeout = self.config.get('timeout', 10) self.data_format = self.config.get('data_format', 'json') self.include_images = self.config.get('include_images', False) # OAuth2 Token self._access_token = None self._token_expires_at = 0 # 上报队列 self.report_queue = queue.Queue() # 工作线程 self.running = False self.worker_thread = None # 统计 self.stats = { 'total_reports': 0, 'success_reports': 0, 'failed_reports': 0, } self.stats_lock = threading.Lock() # 回调 self.on_report_success: Optional[Callable] = None self.on_report_failed: Optional[Callable] = None # 最后上报时间 self.last_report_time = 0 if self.enabled: logger.info(f"[第三方平台] 推送器初始化完成: {self.base_url}") def start(self): """启动推送器""" if not self.enabled: logger.info("[第三方平台] 推送器未启用") return if self.running: return self.running = True self.worker_thread = threading.Thread(target=self._worker_loop, daemon=True) self.worker_thread.start() logger.info("[第三方平台] 推送器已启动") def stop(self): """停止推送器""" self.running = False if self.worker_thread: self.worker_thread.join(timeout=5) logger.info("[第三方平台] 推送器已停止") def _worker_loop(self): """工作线程循环""" while self.running: try: report = self.report_queue.get(timeout=1.0) self._process_report(report) except queue.Empty: continue except Exception as e: logger.error(f"[第三方平台] 处理上报错误: {e}") def _get_auth_headers(self) -> Dict[str, str]: """获取认证请求头（当前第三方接口不需要自定义 header，返回空避免 422）""" return {} def _get_oauth2_token(self) -> Optional[str]: """获取 OAuth2 Token""" # 检查现有 token 是否有效 if self._access_token and time.time() < self._token_expires_at - 60: return self._access_token # 重新获取 token token_url = self.oauth2_config.get('token_url', '') client_id = self.oauth2_config.get('client_id', '') client_secret = self.oauth2_config.get('client_secret', '') scope = self.oauth2_config.get('scope', '') if not all([token_url, client_id, client_secret]): logger.error("[第三方平台] OAuth2 配置不完整") return None try: data = { 'grant_type': 'client_credentials', 'client_id': client_id, 'client_secret': client_secret, } if scope: data['scope'] = scope response = requests.post(token_url, data=data, timeout=self.timeout) if response.status_code == 200: result = response.json() self._access_token = result.get('access_token') expires_in = result.get('expires_in', 3600) self._token_expires_at = time.time() + expires_in logger.info("[第三方平台] OAuth2 Token 获取成功") return self._access_token else: logger.error(f"[第三方平台] OAuth2 Token 获取失败: {response.status_code}") return None except Exception as e: logger.error(f"[第三方平台] OAuth2 Token 请求异常: {e}") return None def _process_report(self, report: BatchReport): """处理单个上报任务""" # 检查推送间隔 current_time = time.time() time_since_last = current_time - self.last_report_time if time_since_last < self.push_interval: time.sleep(self.push_interval - time_since_last) success = self._send_batch_report(report) with self.stats_lock: self.stats['total_reports'] += 1 if success: self.stats['success_reports'] += 1 else: self.stats['failed_reports'] += 1 self.last_report_time = time.time() # 触发回调 if success and self.on_report_success: try: self.on_report_success(report) except Exception as e: logger.error(f"[第三方平台] 成功回调执行错误: {e}") elif not success and self.on_report_failed: try: self.on_report_failed(report) except Exception as e: logger.error(f"[第三方平台] 失败回调执行错误: {e}") def _send_batch_report(self, report: BatchReport) -> bool: """ 发送批次上报请求 Args: report: 批次上报数据 Returns: bool: 是否成功 """ if not self.base_url: logger.error("[第三方平台] 未配置 base_url") return False url = f"{self.base_url}{self.batch_report_url}" # 构建请求数据 payload = self._build_payload(report) try: logger.info("[第三方平台] 上报 payload: %s", json.dumps(payload, ensure_ascii=False)) except Exception: logger.info("[第三方平台] 上报 payload: %s", payload) headers = self._get_auth_headers() for attempt in range(self.retry_count): try: if self.data_format == 'json': response = requests.post( url, json=payload, headers=headers, timeout=self.timeout, verify=False ) else: response = requests.post( url, data=payload, headers=headers, timeout=self.timeout, verify=False ) if response.status_code == 200: result = response.json() status = result.get('status', '') message = result.get('message', '') if (result.get('code') == 200 or result.get('success') == True or status in ('pending', 'success', 'accepted') or '请求已接收' in message or message == 'accepted'): logger.info(f"[第三方平台] 批次上报成功: {report.batch_id}, task_id={result.get('task_id')}") return True else: logger.warning(f"[第三方平台] 批次上报失败: {result.get('msg', '未知错误')}") try: logger.warning(f"[第三方平台] 响应内容: {str(result)[:500]}") except Exception: pass else: logger.warning(f"[第三方平台] 批次上报失败: HTTP {response.status_code}") try: logger.warning(f"[第三方平台] 响应内容: {response.text[:500]}") except Exception: pass if attempt < self.retry_count - 1: time.sleep(self.retry_delay) except requests.exceptions.Timeout: logger.warning(f"[第三方平台] 请求超时 (尝试 {attempt + 1}/{self.retry_count})") if attempt < self.retry_count - 1: time.sleep(self.retry_delay) except Exception as e: logger.error(f"[第三方平台] 请求异常 (尝试 {attempt + 1}/{self.retry_count}): {e}") if attempt < self.retry_count - 1: time.sleep(self.retry_delay) logger.error(f"[第三方平台] 批次上报最终失败: {report.batch_id}") return False def _build_payload(self, report: BatchReport) -> Dict[str, Any]: """ 构建上报请求体 Args: report: 批次上报数据 Returns: Dict: 请求体字典 """ batch_info = report.batch_info # 根据平台类型调整格式 normalized_ts = _normalize_timestamp(report.timestamp) if self.platform_type == 'jtjai': # 优先取 OSS URL，否则用本地路径 urls = batch_info.get('image_urls') or {} image_url = urls.get('original') or urls.get('marked') or (batch_info.get('image_paths') or [None])[0] payload = { 'createTime': datetime.fromtimestamp(normalized_ts).strftime("%Y-%m-%d %H:%M:%S"), 'addr': f"设备{report.device_id}批次上报", 'installAddress': batch_info.get('install_address', ''), 'ext1': json.dumps([image_url]), 'ext2': json.dumps({ 'batchId': report.batch_id, 'deviceId': report.device_id, 'projectId': report.project_id, 'installAddress': batch_info.get('install_address', ''), 'totalPersons': len(batch_info.get('detections', [])), 'ptzImagesCount': 1 if batch_info.get('camera_type') == 'ptz' else 0, 'persons': batch_info.get('detections', []), 'imageUrls': urls, }) } else: # custom / 其他平台：把新版 batch_info 转回老字段名后上报， # 兼容原人体分析平台对 panorama / total_persons / persons 的解析。 payload = _convert_to_legacy_batch_info(batch_info) # 统一时间戳单位为秒，避免第三方解析错误 payload['timestamp'] = normalized_ts return payload def report_batch(self, batch_info: Dict[str, Any], local_path: Optional[str] = None): """ 上报批次信息 Args: batch_info: batch_info.json 的字典内容 local_path: batch_info.json 的本地文件路径（可选） """ if not self.enabled: return # 接受所有相机类型的检测上报（panorama 或 ptz） # 业务流程：检测到人 → 上传 OSS → 推送第三方平台 report = BatchReport( batch_id=batch_info.get('batch_id', ''), device_id=batch_info.get('device_id', ''), project_id=batch_info.get('project_id', ''), timestamp=batch_info.get('timestamp', time.time()), batch_info=batch_info, local_path=local_path ) self.report_queue.put(report) def report_batch_sync(self, batch_info: Dict[str, Any], local_path: Optional[str] = None) -> bool: """ 同步上报批次信息 Args: batch_info: batch_info.json 的字典内容 local_path: batch_info.json 的本地文件路径（可选） Returns: bool: 是否成功 """ if not self.enabled: return False report = BatchReport( batch_id=batch_info.get('batch_id', ''), device_id=batch_info.get('device_id', ''), project_id=batch_info.get('project_id', ''), timestamp=batch_info.get('timestamp', time.time()), batch_info=batch_info, local_path=local_path ) return self._send_batch_report(report) def send_heartbeat(self) -> bool: """ 发送心跳 Returns: bool: 是否成功 """ if not self.enabled or not self.heartbeat_url: return False url = f"{self.base_url}{self.heartbeat_url}" payload = { 'deviceId': self.device_config.get('device_id', ''), 'projectId': self.device_config.get('project_id', ''), 'timestamp': time.time(), 'status': 'online', } headers = self._get_auth_headers() try: response = requests.post( url, json=payload, headers=headers, timeout=self.timeout, verify=False ) if response.status_code == 200: logger.debug("[第三方平台] 心跳发送成功") return True else: logger.warning(f"[第三方平台] 心跳发送失败: HTTP {response.status_code}") return False except Exception as e: logger.error(f"[第三方平台] 心跳发送异常: {e}") return False def set_callbacks(self, on_success: Callable = None, on_failed: Callable = None): """ 设置回调函数 Args: on_success: 上报成功回调 on_failed: 上报失败回调 """ self.on_report_success = on_success self.on_report_failed = on_failed def get_stats(self) -> Dict[str, int]: """获取统计信息""" with self.stats_lock: return self.stats.copy() def is_enabled(self) -> bool: """检查是否启用""" return self.enabled # 全局单例 _third_party_pusher_instance: Optional[ThirdPartyPusher] = None _third_party_pusher_lock = threading.Lock() def get_third_party_pusher(config: Dict[str, Any] = None) -> ThirdPartyPusher: """ 获取第三方平台推送器实例（单例模式，线程安全） Args: config: 第三方平台配置 Returns: ThirdPartyPusher 实例 """ global _third_party_pusher_instance if _third_party_pusher_instance is None: with _third_party_pusher_lock: if _third_party_pusher_instance is None: _third_party_pusher_instance = ThirdPartyPusher(config) return _third_party_pusher_instance def reset_third_party_pusher(): """重置第三方平台推送器实例""" global _third_party_pusher_instance with _third_party_pusher_lock: if _third_party_pusher_instance is not None: _third_party_pusher_instance.stop() _third_party_pusher_instance = None