from datetime import datetime import json import asyncio from typing import Dict from fastapi import HTTPException from openai import OpenAI from .config import ( CAMERA_DB_MAPPING, BEHAVIOR_CATEGORIES, ABNORMAL_BEHAVIORS, REDIS_CACHE_CONFIG, ai_client ) async def get_camera_data_by_date(camera_id: str, date: str, redis_client, is_face: bool = False): """获取摄像头某天的所有数据""" try: if camera_id not in CAMERA_DB_MAPPING: raise HTTPException(status_code=400, detail="Invalid camera ID") # 使用新的键格式进行模式匹配 pattern = f"{'face_' if is_face else ''}{camera_id}_{date}_*" all_keys = redis_client.keys(pattern) if not all_keys: return {"message": "No data found", "data": None} # 获取所有键的数据并解析 all_data = {} for key in all_keys: data = redis_client.get(key) if data: all_data[key] = json.loads(data) return { "message": "success", "data": all_data, "total_records": len(all_data) } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) def background_generate_report(date_no_hyphen: str, redis_connections): """后台生成报告的函数""" try: print(f"\n=== 后台任务开始生成报告 {date_no_hyphen} ===") loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) report_redis = redis_connections["report"] task_key = f"task_status_{date_no_hyphen}" report_redis.setex( task_key, REDIS_CACHE_CONFIG["task_status_expiry"], json.dumps({"status": "running"}) ) print("开始调用 generate_daily_report...") report = loop.run_until_complete(generate_daily_report(date_no_hyphen, redis_connections)) print(f"报告生成结果: {report.get('message', 'unknown')}") if report.get("message") != "no_data": print("报告生成成功，准备保存到Redis...") report_key = f"report_{date_no_hyphen}" report_redis.setex( report_key, REDIS_CACHE_CONFIG["report_expiry"], json.dumps(report) ) print(f"报告已保存到Redis，键名: {report_key}") report_redis.setex( task_key, REDIS_CACHE_CONFIG["task_status_expiry"], json.dumps({"status": "completed"}) ) else: report_redis.setex( task_key, REDIS_CACHE_CONFIG["task_status_expiry"], json.dumps({"status": "completed"}) ) print("任务状态已更新为completed") except Exception as e: print(f"报告生成失败: {str(e)}") report_redis.setex( task_key, REDIS_CACHE_CONFIG["task_status_expiry"], json.dumps({ "status": "failed", "error": str(e) }) ) print("任务状态已更新为failed") finally: loop.close() print(f"=== 后台任务结束 {date_no_hyphen} ===\n") async def generate_daily_report(date: str, redis_connections) -> Dict: """生成每日分析报告""" print(f"\n=== 开始生成日报 {date} ===") # 初始化数据收集结构 data_collection = { "date": date, "total_events": 0, "abnormal_events": 0, "camera_num": set(), "activity_areas": {}, # 活动区域统计 "behavior_distribution": {}, # 行为分布 "hourly_stats": {}, # 每小时统计 "category_stats": { # 各类别行为统计 category: { "count": 0, "behaviors": {} } for category in BEHAVIOR_CATEGORIES.keys() }, "abnormal_stats": { "behaviors": [], "times": [], "locations": [] } } # 初始化摄像头小时统计 camera_hourly_counts = { camera_id: {f"{hour:02d}": 0 for hour in range(24)} for camera_id in CAMERA_DB_MAPPING.keys() if camera_id != "report" } # 遍历所有摄像头数据 has_any_data = False total_cameras = len([cam for cam in CAMERA_DB_MAPPING.keys() if cam != "report"]) processed_cameras = 0 print(f"开始处理 {total_cameras} 个摄像头的数据") # 数据收集和预处理 for camera_id, redis_client in redis_connections.items(): if camera_id == "report": continue processed_cameras += 1 print(f"\n处理摄像头 {camera_id} ({processed_cameras}/{total_cameras})") camera_event_count = 0 for hour in range(24): hour_str = f"{hour:02d}" pattern = f"{camera_id}_{date}_{hour_str}*" hour_keys = redis_client.keys(pattern) for key in hour_keys: hour_data = redis_client.get(key) if hour_data: has_any_data = True hour_json = json.loads(hour_data) for video_file, video_data in hour_json.items(): if "video_analysis" in video_data: analysis = video_data["video_analysis"]["qwen-7B"]["extracted_info"] behaviors = analysis.get("actions", []) camera_event_count += len(behaviors) data_collection["total_events"] += len(behaviors) environment = analysis.get("environment", "") if environment: if isinstance(environment, list): for env in environment: if isinstance(env, str): data_collection["activity_areas"][env] = \ data_collection["activity_areas"].get(env, 0) + 1 elif isinstance(environment, str): data_collection["activity_areas"][environment] = \ data_collection["activity_areas"].get(environment, 0) + 1 if hour_str not in data_collection["hourly_stats"]: data_collection["hourly_stats"][hour_str] = { "event_count": 0, "categories": {cat: 0 for cat in BEHAVIOR_CATEGORIES.keys()} } data_collection["hourly_stats"][hour_str]["event_count"] += len(behaviors) camera_hourly_counts[camera_id][hour_str] += len(behaviors) for behavior in behaviors: data_collection["behavior_distribution"][behavior] = \ data_collection["behavior_distribution"].get(behavior, 0) + 1 behavior_categorized = False for category, keywords in BEHAVIOR_CATEGORIES.items(): if any(keyword in behavior for keyword in keywords): data_collection["category_stats"][category]["count"] += 1 if behavior not in data_collection["category_stats"][category]["behaviors"]: data_collection["category_stats"][category]["behaviors"][behavior] = { "count": 0, "occurrences": {} } occurrence_key = f"{camera_id}_{hour_str}" if occurrence_key not in data_collection["category_stats"][category]["behaviors"][behavior]["occurrences"]: data_collection["category_stats"][category]["behaviors"][behavior]["count"] += 1 data_collection["category_stats"][category]["behaviors"][behavior]["occurrences"][occurrence_key] = { "time": f"{hour_str}:00", "camera": camera_id } data_collection["hourly_stats"][hour_str]["categories"][category] += 1 behavior_categorized = True break if not behavior_categorized: data_collection["category_stats"]["其他"]["count"] += 1 if behavior not in data_collection["category_stats"]["其他"]["behaviors"]: data_collection["category_stats"]["其他"]["behaviors"][behavior] = { "count": 0, "occurrences": {} } occurrence_key = f"{camera_id}_{hour_str}" if occurrence_key not in data_collection["category_stats"]["其他"]["behaviors"][behavior]["occurrences"]: data_collection["category_stats"]["其他"]["behaviors"][behavior]["count"] += 1 data_collection["category_stats"]["其他"]["behaviors"][behavior]["occurrences"][occurrence_key] = { "time": f"{hour_str}:00", "camera": camera_id } data_collection["hourly_stats"][hour_str]["categories"]["其他"] += 1 if any(abnormal in behavior for abnormal in ABNORMAL_BEHAVIORS): occurrence_key = f"{camera_id}_{hour_str}" abnormal_key = f"{behavior}_{occurrence_key}" if abnormal_key not in data_collection["abnormal_stats"]["behaviors"]: data_collection["abnormal_events"] += 1 data_collection["abnormal_stats"]["behaviors"].append({ "behavior": behavior, "time": f"{hour_str}:00", "camera": camera_id }) else: print(f" - {hour_str}时 无数据") print(f"摄像头 {camera_id} 总计: {camera_event_count} 个事件") if camera_event_count > 0: data_collection["camera_num"].add(camera_id) if len(data_collection["camera_num"]) == 0: print("判定为无数据，返回") return { "message": "no_data", "data": None, "detail": "暂无数据" } data_collection["camera_num"] = list(data_collection["camera_num"]) sorted_hours = sorted( data_collection["hourly_stats"].items(), key=lambda x: x[1]["event_count"], reverse=True ) data_collection["peak_hours"] = [hour for hour, _ in sorted_hours[:3]] preprocessed_data = { "日期": data_collection["date"], "摄像头数量": len(data_collection["camera_num"]), "行为总数": data_collection["total_events"], "异常行为数": data_collection["abnormal_events"], "行为高峰时段": data_collection["peak_hours"], "主要活动区域": data_collection["activity_areas"], "行为类别统计": data_collection["category_stats"], "异常行为统计": data_collection["abnormal_stats"], "每小时行为统计": data_collection["hourly_stats"] } ai_analysis = await analyze_experiment_data(preprocessed_data) final_report = { "整体活动趋势": ai_analysis["整体活动趋势"], "高峰时段分析": ai_analysis["高峰时段分析"], "异常行为分析": ai_analysis["异常行为分析"], "行为分析": ai_analysis["行为分析"], "建议": ai_analysis["建议"] } final_report["hourly_distribution"] = [] for camera_id in camera_hourly_counts: total_events = sum(camera_hourly_counts[camera_id].values()) if total_events > 0: camera_data = { "camera_id": camera_id, "data": [] } for hour in range(24): hour_str = f"{hour:02d}" hour_data = { "hour": f"{hour_str}:00", "count": camera_hourly_counts[camera_id][hour_str], "categories": data_collection["hourly_stats"].get(hour_str, {}).get("categories", {}) } camera_data["data"].append(hour_data) final_report["hourly_distribution"].append(camera_data) return final_report async def analyze_experiment_data(report_info): """使用AI分析实验数据""" system_prompt = """ You are an AI assistant tasked with analyzing data. Generate a comprehensive analysis report in JSON format. The JSON structure must strictly follow the provided template. """ user_prompt = f"""Analyze the preprocessed data based on the following information: Preprocessed data: {json.dumps(report_info, ensure_ascii=False)} Generate a JSON response with the following structure: {{ "整体活动趋势": {{ "日期": "{report_info['日期']}", "摄像头数量": {report_info['摄像头数量']}, "行为总数": {report_info['行为总数']}, "异常行为数": {report_info['异常行为数']}, "行为高峰时段": {json.dumps(report_info['行为高峰时段'], ensure_ascii=False)}, "主要活动区域": {json.dumps(report_info['主要活动区域'], ensure_ascii=False)} }}, "高峰时段分析": {{ "高峰时段": "分析行为高峰时段", "高峰时段行为": "分析高峰时段主要行为", "活动规律": "分析活动规律" }}, "异常行为分析": {{ "异常行为": "分析异常行为类型", "异常行为次数": "分析异常行为频率", "异常行为出现时间": "分析异常行为时间分布", "异常行为地点": "分析监测到异常行为的摄像头" }}, "行为分析": {{ "基础动作": {{ "站立行为": "分析站立相关行为", "行走行为": "分析行走相关行为", "坐卧行为": "分析坐卧相关行为", "其他基础动作": "分析其他基础动作" }}, "日常生活": {{ "饮食情况": "分析饮食相关行为", "休息情况": "分析休息相关行为", "医疗情况": "分析医疗相关行为" }}, "社交活动": {{ "交际情况": "分析交际相关行为", "娱乐情况": "分析娱乐相关行为", "情感表达": "分析情感表达相关行为" }}, "工作学习": {{ "学习情况": "分析学习相关行为", "工作情况": "分析学习相关行为", "创作活动": "分析创作相关行为" }}, "运动娱乐": {{ "运动情况": "分析运动相关行为", "运动时长": "分析运动持续时间", "运动强度": "分析运动强度" }}, "其他行为": {{ "出现时间":"分析其他行为出现时间", "出现次数":"分析其他行为出现次数" }} }}, "建议": {{ "生活作息": ["建议1", "建议2"], "活动安排": ["建议1", "建议2"], "安全防护": ["建议1", "建议2"], "健康建议": ["建议1", "建议2"] }} }} """ try: response = ai_client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], max_tokens=4096, temperature=0.7, response_format={'type': 'json_object'} ) return json.loads(response.choices[0].message.content) except Exception as e: raise HTTPException( status_code=500, detail="AI分析服务暂时不可用，请稍后重试" )