129 lines
3.8 KiB
Python
129 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
违禁品查获排行榜 - Web展示应用
|
||
支持日榜、月榜、总榜展示，带动态筛选和可视化效果
|
||
"""
|
||
|
||
from flask import Flask, render_template, jsonify
|
||
import json
|
||
import os
|
||
from datetime import datetime, date
|
||
|
||
app = Flask(__name__)
|
||
|
||
# 数据文件路径 (支持从环境变量覆盖)
|
||
DATA_FILE = os.environ.get('CONTRABAND_DATA_FILE',
|
||
os.path.join(os.path.dirname(__file__), 'security_data_v2.json'))
|
||
|
||
|
||
def load_and_process_data():
|
||
"""
|
||
加载并处理违禁品数据
|
||
返回格式: {维度: {榜单列表}}
|
||
维度: 'month', 'all'
|
||
- month: 本月统计（有记录人员）
|
||
- all: 总榜（所有在职人员，包括0记录）
|
||
"""
|
||
# 检查数据文件
|
||
if not os.path.exists(DATA_FILE):
|
||
app.logger.warning(f"数据文件不存在: {DATA_FILE}")
|
||
return {'month': [], 'all': []}
|
||
|
||
try:
|
||
with open(DATA_FILE, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
except (json.JSONDecodeError, UnicodeDecodeError) as e:
|
||
app.logger.error(f"数据文件解析失败: {e}")
|
||
return {'month': [], 'all': []}
|
||
|
||
members = data.get("members", {})
|
||
logs = data.get("logs", [])
|
||
|
||
# 获取业务日期（处理跨夜班的情况）
|
||
today_date = date.today()
|
||
today_str = today_date.strftime("%Y-%m-%d")
|
||
current_month = today_date.strftime("%Y-%m")
|
||
|
||
# 初始化三个维度的容器: {维度: {姓名: [明细列表]}}
|
||
categories = {
|
||
"month": {},
|
||
"all": {}
|
||
}
|
||
|
||
# 总榜初始化：所有在职人员（包括0记录）
|
||
for name, info in members.items():
|
||
if info.get('status') == 'active':
|
||
categories["all"][name] = []
|
||
|
||
# 遍历日志进行分类
|
||
for log in logs:
|
||
# 跳过已删除的记录
|
||
if log.get('is_deleted'):
|
||
continue
|
||
|
||
name = log.get('name')
|
||
biz_date = log.get('biz_date', '')
|
||
biz_month = log.get('biz_month', '')
|
||
|
||
# 构建明细
|
||
detail = {
|
||
"time": log.get('actual_time', ''),
|
||
"content": log.get('content', '')
|
||
}
|
||
|
||
# 归档至"总榜"（只统计在职人员）
|
||
if name in categories["all"]:
|
||
categories["all"][name].append(detail)
|
||
|
||
# 归档至"月榜"
|
||
if biz_month == current_month:
|
||
categories["month"].setdefault(name, []).append(detail)
|
||
|
||
# 格式化前端所需的榜单数据
|
||
final_data = {}
|
||
for cat, user_map in categories.items():
|
||
sorted_rank = []
|
||
for name, details in user_map.items():
|
||
member_info = members.get(name, {})
|
||
sorted_rank.append({
|
||
"name": name,
|
||
"score": len(details),
|
||
"shift": member_info.get("shift", "未知"),
|
||
"status": member_info.get("status", "unknown"),
|
||
"details": details
|
||
})
|
||
|
||
# 按分数降序排序
|
||
sorted_rank.sort(key=lambda x: x['score'], reverse=True)
|
||
final_data[cat] = sorted_rank
|
||
|
||
return final_data
|
||
|
||
|
||
@app.route('/')
|
||
def index():
|
||
"""主页 - 加载所有榜单数据"""
|
||
all_data = load_and_process_data()
|
||
return render_template('index.html',
|
||
data_json=json.dumps(all_data, ensure_ascii=False))
|
||
|
||
|
||
@app.route('/api/rankings')
|
||
def api_rankings():
|
||
"""API接口 - 返回JSON格式的榜单数据"""
|
||
all_data = load_and_process_data()
|
||
return jsonify(all_data)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# 开发环境配置
|
||
debug_mode = os.environ.get('FLASK_DEBUG', 'False').lower() in ('true', '1', 't')
|
||
host = '127.0.0.1'
|
||
port = 9517
|
||
|
||
print(f"🚀 违禁品排行榜启动中...")
|
||
print(f"📁 数据文件: {DATA_FILE}")
|
||
print(f"🌐 访问地址: http://{host}:{port}")
|
||
|
||
app.run(host=host, port=port, debug=debug_mode)
|