feat: 统计图表功能 - 人数时间线曲线图、事件类型分布、历史趋势

This commit is contained in:
2026-04-17 11:34:10 +08:00
parent 7dea6fa38f
commit a7e005bc65
4 changed files with 415 additions and 0 deletions

View File

@@ -194,6 +194,127 @@ async def rename_person(person_id: str, name: str):
raise HTTPException(status_code=404, detail="人员不存在")
@app.get("/api/stats/daily")
async def get_daily_stats(date: str = None):
"""获取每日统计数据"""
from datetime import datetime, timedelta
# 默认昨天的数据
if date is None:
yesterday = datetime.now() - timedelta(days=1)
date = yesterday.strftime('%Y-%m-%d')
conn = db._get_conn()
conn.row_factory = sqlite3.Row
# 获取当天所有图片
cursor = conn.execute(
"SELECT id, timestamp, analyzed FROM images WHERE date_folder = ? ORDER BY timestamp",
(date,)
)
images = [dict(row) for row in cursor.fetchall()]
# 获取每个时间点的人数
timeline = []
for img in images:
# 获取该图片的事件
events = db.get_events_by_image(img['id'])
# 计算人数
person_count = 0
for event in events:
if '人物活动' in event['event_type'] or '人员进出' in event['event_type']:
# 从描述中提取人数
desc = event['description']
if '' in desc and '' in desc:
try:
count_str = desc.split('')[1].split('')[0]
person_count = int(count_str)
except:
person_count = 1
elif '#1' in desc or '#2' in desc or '#3' in desc:
person_count = max(person_count, 1)
timeline.append({
'time': img['timestamp'],
'image_id': img['id'],
'person_count': person_count,
'analyzed': img['analyzed']
})
# 统计汇总
total_images = len(images)
total_events = conn.execute(
"SELECT COUNT(*) FROM events WHERE image_id IN (SELECT id FROM images WHERE date_folder = ?)",
(date,)
).fetchone()[0]
# 事件类型统计
cursor = conn.execute(
"""SELECT event_type, COUNT(*) as count
FROM events WHERE image_id IN
(SELECT id FROM images WHERE date_folder = ?)
GROUP BY event_type""",
(date,)
)
event_types = [{'type': row[0], 'count': row[1]} for row in cursor.fetchall()]
conn.close()
return {
'date': date,
'timeline': timeline,
'total_images': total_images,
'total_events': total_events,
'event_types': event_types
}
@app.get("/api/stats/history")
async def get_history_stats(days: int = 7):
"""获取历史统计最近N天"""
from datetime import datetime, timedelta
conn = db._get_conn()
history = []
for i in range(days):
date = (datetime.now() - timedelta(days=i)).strftime('%Y-%m-%d')
# 统计当天的数据
total_images = conn.execute(
"SELECT COUNT(*) FROM images WHERE date_folder = ?",
(date,)
).fetchone()[0]
total_events = conn.execute(
"SELECT COUNT(*) FROM events WHERE image_id IN (SELECT id FROM images WHERE date_folder = ?)",
(date,)
).fetchone()[0]
# 人数变化统计
cursor = conn.execute(
"""SELECT e.event_type, COUNT(*) as count
FROM events e JOIN images i ON e.image_id = i.id
WHERE i.date_folder = ? AND e.event_type LIKE '%人员进出%'
GROUP BY e.event_type""",
(date,)
)
person_changes = sum(row[1] for row in cursor.fetchall())
history.append({
'date': date,
'total_images': total_images,
'total_events': total_events,
'person_changes': person_changes
})
conn.close()
return {'history': history, 'days': days}
# ============== 图片 API ==============
@app.get("/api/images")