Files
zydi-web/main.py
T
2025-01-12 03:01:51 +00:00

705 lines
27 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from redis import Redis
from openai import OpenAI
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import asyncio
app = FastAPI()
# 配置CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 在生产环境中应该设置具体的域名
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 定义摄像头和数据库的映射关系
CAMERA_DB_MAPPING = {
"camera001": 207,
"camera002": 208,
"camera003": 209,
"A01": 210,
"B02": 211,
"C03": 212,
"report": 213 # 分析报告使用213数据库
}
# 创建Redis连接池
redis_connections = {}
for camera_id, db in CAMERA_DB_MAPPING.items():
redis_connections[camera_id] = Redis(
host="222.186.10.253",
port=6379,
password="Obscura@2024",
db=db,
decode_responses=True
)
@app.get("/web/face/{camera_id}/data")
async def get_camera_data(camera_id: str, date: Optional[str] = None):
"""
获取摄像头某天的所有数据
"""
try:
if camera_id not in CAMERA_DB_MAPPING:
raise HTTPException(status_code=400, detail="Invalid camera ID")
# 如果没有指定日期,使用当前日期
if date is None:
date = datetime.now().strftime("%Y%m%d")
redis_client = redis_connections[camera_id]
# 使用新的键格式进行模式匹配
pattern = f"face_{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:
# 直接解析JSON数据,无需decode
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))
@app.get("/web/{camera_id}/data")
async def get_camera_data(camera_id: str, date: Optional[str] = None):
"""
获取摄像头某天的所有数据
"""
try:
if camera_id not in CAMERA_DB_MAPPING:
raise HTTPException(status_code=400, detail="Invalid camera ID")
# 如果没有指定日期,使用当前日期
if date is None:
date = datetime.now().strftime("%Y%m%d")
redis_client = redis_connections[camera_id]
# 使用新的键格式进行模式匹配
pattern = f"{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:
# 直接解析JSON数据,无需decode
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))
# 添加新路由支持日期查询
@app.get("/web/report/{date}")
async def get_report_by_date(date: str):
"""
获取指定日期的分析报告
:param date: 日期,格式为YYYY-MM-DD
"""
try:
print(f"\n=== 开始处理日期: {date} ===")
# 验证日期格式并转换为无连字符格式
try:
parsed_date = datetime.strptime(date, "%Y-%m-%d")
date_no_hyphen = parsed_date.strftime("%Y%m%d")
print(f"转换后的日期格式: {date_no_hyphen}")
except ValueError:
raise HTTPException(
status_code=400,
detail="Invalid date format. Please use YYYY-MM-DD"
)
# 使用report数据库存储报告
report_redis = redis_connections["report"]
report_key = f"report_{date_no_hyphen}"
print(f"缓存键: {report_key}")
# 尝试获取缓存的报告
cached_report = report_redis.get(report_key)
print(f"是否有缓存: {'' if cached_report else ''}")
if cached_report:
print("返回缓存数据")
return {
"message": "success",
"data": json.loads(cached_report),
"source": "cache"
}
# 如果缓存中没有,生成新报告
print("开始生成新报告...")
report = await generate_daily_report(date_no_hyphen)
# 检查是否返回"暂无数据"
if report.get("message") == "no_data":
print("生成报告结果: 暂无数据")
return report
print("生成报告成功,准备缓存...")
# 缓存报告到report数据库(设置30天过期)
report_redis.setex(
report_key,
timedelta(days=30),
json.dumps(report)
)
return {
"message": "success",
"data": report,
"source": "new_generation"
}
except HTTPException as he:
print(f"HTTP Exception: {str(he)}")
raise he
except Exception as e:
print(f"Error processing report request: {str(e)}")
import traceback
print(f"详细错误信息: {traceback.format_exc()}")
# 为了更好的调试,先打印错误信息再抛出异常
raise HTTPException(
status_code=500,
detail=f"处理报告时发生错误: {str(e)}\n错误类型: {type(e)}"
)
# 生成每日分析报告的函数
async def generate_daily_report(date: str) -> Dict:
print(f"\n=== 开始生成日报 {date} ===")
# 定义异常行为列表
ABNORMAL_BEHAVIORS = [
'打架',
'斗殴',
'摔倒',
'晕倒',
'昏倒',
'跌倒',
'滑倒',
'',
'',
'受伤',
'暴力',
'攻击',
'威胁',
'破坏',
'偷窃',
'抢夺',
'游荡',
'徘徊',
'尾随',
'骚扰'
]
# 定义行为类别
BEHAVIOR_CATEGORIES = {
"基础动作": [
"", "站立", "站着",
"", "走路", "散步", "行走", "徒步",
"", "奔跑", "慢跑",
"", "坐下", "坐着",
"", "蹲下", "蹲着",
"", "转身", "转头", "回头", "旋转", "转向", "转弯",
"", "", "", ""
],
"日常生活": [
"", "食用", "吃饭", "吃零食", "吃东西", "用餐", "咀嚼", "",
"喝水", "喝牛奶", "喝茶", "饮用", "喝咖啡", "", "饮水",
"穿衣服", "穿裤子", "穿鞋", "戴帽子", "戴口罩", "戴围巾",
"", "", "睡觉", "休息", "打哈欠",
"洗澡", "刷牙", "洗手", "洗涤", "清洁", "擦洗",
"吃药", "喝药", "服药"
],
"社交活动": [
"说话", "交流", "演讲", "谈话", "聊天", "采访", "社交",
"打麻将", "打牌", "玩手机", "玩电脑", "玩游戏", "赌博",
"", "大笑", "微笑", "哭泣", "咯咯笑", "皱眉"
],
"工作学习": [
"读书", "阅读", "看书",
"写作", "写字", "",
"工作", "学习", "使用电脑", "使用笔记本电脑", "使用手机", "开会", "打字",
"画画", "绘画", "摄影", "素描"
],
"运动娱乐": [
"", "跳跃", "跳舞", "游泳", "运动", "健身", "锻炼"
],
"异常行为": [
'打架',
'斗殴',
'摔倒',
'晕倒',
'昏倒',
'跌倒',
'滑倒',
'',
'',
'受伤',
'暴力',
'攻击',
'威胁',
'破坏',
'偷窃',
'抢夺',
'游荡',
'徘徊',
'尾随',
'骚扰'
],
"其他": ["其他"]
}
# 初始化数据收集结构
data_collection = {
"date": date,
"total_events": 0,
"abnormal_events": 0,
"camera_num": set(),
"activity_areas": {}, # 活动区域统计
"behavior_distribution": {}, # 行为分布
"hourly_stats": {}, # 每小时统计
"category_stats": { # 各类别行为统计
"基础动作": {
"count": 0,
"behaviors": {} # 改为以行为为key的统计
},
"日常生活": {
"count": 0,
"behaviors": {}
},
"社交活动": {
"count": 0,
"behaviors": {}
},
"工作学习": {
"count": 0,
"behaviors": {}
},
"运动娱乐": {
"count": 0,
"behaviors": {}
},
"异常行为": {
"count": 0,
"behaviors": {}
},
"其他": {
"count": 0,
"behaviors": {}
}
},
"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:
# 如果 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
else:
# 如果是字符串,直接处理
if 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": {} # 改为使用字典,键为"camera_time"组合
}
# 使用camera_id和hour_str组合作为唯一键
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)
print(f"\n=== 数据收集完成 ===")
print(f"has_any_data: {has_any_data}")
print(f"total_events: {data_collection['total_events']}")
print(f"活跃摄像头: {list(data_collection['camera_num'])}")
# 如果没有数据,提前返回
if len(data_collection["camera_num"]) == 0:
print("判定为无数据,返回")
return {
"message": "no_data",
"data": None,
"detail": "暂无数据"
}
print("\n开始生成报告...")
# 将摄像头集合转换为列表
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]]
# 准备发送给AI分析的数据
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分析
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
# SiliconFlow API Configuration
client = OpenAI(
base_url="https://api.siliconflow.cn/v1",
api_key="sk-ytxabphvgxrjbvnqiwercjyrabvlukwddqsmvnqnvwuazamd"
)
# 修改函数定义为异步函数
async def analyze_experiment_data(report_info):
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 = client.chat.completions.create(
model="deepseek-ai/DeepSeek-V2.5",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
response_format={'type': 'json_object'},
max_tokens=4096,
temperature=0.7
)
# 解析AI响应
ai_response = json.loads(response.choices[0].message.content)
return ai_response
except Exception as e:
raise HTTPException(
status_code=500,
detail="AI分析服务暂时不可用,请稍后重试"
)
@app.get("/web/report/download/{date}")
async def download_report(date: str):
"""
下载指定日期的分析报告
:param date: 日期,格式为YYYY-MM-DD
"""
try:
# 验证日期格式并转换为无连字符格式
try:
parsed_date = datetime.strptime(date, "%Y-%m-%d")
date_no_hyphen = parsed_date.strftime("%Y%m%d")
except ValueError:
error_msg = "日期格式必须为YYYY-MM-DD(例如:2024-12-31"
print(f"日期格式错误: {error_msg}")
raise HTTPException(
status_code=400,
detail=error_msg
)
report_key = f"report_{date_no_hyphen}"
# 使用report数据库
report_redis = redis_connections["report"]
# 获取报告数据
report_data = report_redis.get(report_key)
if not report_data:
error_msg = f"未找到 {date} 的报告数据,请先生成报告"
print(error_msg)
raise HTTPException(
status_code=404,
detail=error_msg
)
# 解析数据并移除 hourly_distribution 字段
try:
data = json.loads(report_data)
if "hourly_distribution" in data:
del data["hourly_distribution"]
print("成功获取报告数据")
return {
"message": "success",
"data": data
}
except json.JSONDecodeError as je:
error_msg = f"报告数据格式错误: {str(je)}"
print(error_msg)
raise HTTPException(
status_code=500,
detail=error_msg
)
except HTTPException as he:
raise he
except Exception as e:
error_msg = f"处理请求时发生错误: {str(e)}"
print(error_msg)
raise HTTPException(
status_code=500,
detail=error_msg
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=6005)