dsh/dual_camera_system/third_party_pusher.py

"""
第三方平台推送模块
将批次信息推送到第三方平台接口
"""

import os
import time
import json
import logging
import threading
import queue
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 的字段名。

    老字段包括：
    - panorama.local_path / panorama.oss_url
    - total_persons
    - ptz_images_count
    - persons（含 person_index、bbox 字典、confidence 等）
    """
    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")

    # 优先取 marked OSS URL，兼容老代码只读取 panorama.oss_url 的行为
    oss_url = urls.get("marked") or urls.get("original") or None
    local_path = image_paths[0] if image_paths else None

    persons = []
    for i, det in enumerate(new_info.get("detections") or []):
        bbox = det.get("bbox", [0, 0, 0, 0])
        person = {
            "person_index": i,
            "bbox": {
                "x1": int(bbox[0]),
                "y1": int(bbox[1]),
                "x2": int(bbox[2]),
                "y2": int(bbox[3]),
            },
            "confidence": float(det.get("confidence", 0.0)),
            "camera_type": det.get("camera_type", camera_type),
        }
        ptz_position = new_info.get("ptz_position")
        if ptz_position:
            person["ptz_position"] = ptz_position
        persons.append(person)

    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(),
        "panorama": {
            "local_path": local_path,
            "oss_url": oss_url,
        },
        "total_persons": len(persons),
        "ptz_images_count": 1 if camera_type == "ptz" else 0,
        "persons": persons,
    }
    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)
        
        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
                        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}批次上报",
                'ext1': json.dumps([image_url]),
                'ext2': json.dumps({
                    'batchId': report.batch_id,
                    'deviceId': report.device_id,
                    'projectId': report.project_id,
                    '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
        
        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