110 lines
3.8 KiB
Python
Executable File
110 lines
3.8 KiB
Python
Executable File
from datetime import datetime, timedelta
|
|
from collections import defaultdict
|
|
import sys
|
|
import os
|
|
|
|
# 添加项目根目录到 Python 路径
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from scripts.models import db, Device, WorkOrder, WorkOrderHandling, ManualWorkOrder
|
|
from utils.lzmx import send_weekly_stats_message
|
|
from flask import Flask
|
|
|
|
# 创建一个简单的 Flask 应用
|
|
app = Flask(__name__)
|
|
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://4an:GtbJp6Ai5azntnTB@10.10.10.253/4an'
|
|
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
|
|
LZMX_WEBHOOK_URL = "http://imtwo.zdxlz.com/im-external/v1/webhook/send?key=ef1dc605827f425e96491ce5725b5274"
|
|
|
|
# 初始化数据库
|
|
db.init_app(app)
|
|
|
|
def get_reward_by_distance(project_type, distance):
|
|
if project_type == '校园安防"4+N"项目':
|
|
if distance <= 5: return 20
|
|
elif distance <= 30: return 40
|
|
elif distance <= 50: return 50
|
|
else: return 60
|
|
elif project_type == '综治视联网':
|
|
if distance <= 10: return 30
|
|
elif distance <= 30: return 40
|
|
elif distance <= 50: return 50
|
|
else: return 60
|
|
elif project_type == '教育城域网':
|
|
if distance <= 5: return 20
|
|
elif distance <= 30: return 30
|
|
elif distance <= 50: return 40
|
|
else: return 50
|
|
elif project_type == '其他':
|
|
return 30
|
|
return 0
|
|
|
|
def weekly_stats():
|
|
with app.app_context():
|
|
# 计算本周的开始和结束时间
|
|
now = datetime.now()
|
|
week_start = now - timedelta(days=now.weekday())
|
|
week_start = week_start.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
week_end = now
|
|
|
|
# 统计工单数据
|
|
total_orders = WorkOrder.query.filter(
|
|
WorkOrder.dispatch_time >= week_start,
|
|
WorkOrder.dispatch_time < week_end
|
|
).count()
|
|
|
|
completed_orders = WorkOrder.query.filter(
|
|
WorkOrder.dispatch_time >= week_start,
|
|
WorkOrder.dispatch_time < week_end,
|
|
WorkOrder.status == '已处理'
|
|
).count()
|
|
|
|
manual_orders = ManualWorkOrder.query.filter(
|
|
ManualWorkOrder.create_time >= week_start,
|
|
ManualWorkOrder.create_time < week_end
|
|
).count()
|
|
|
|
# 获取处理人统计
|
|
handler_stats = defaultdict(float)
|
|
|
|
# 统计普通工单奖励
|
|
handlings = WorkOrderHandling.query.filter(
|
|
WorkOrderHandling.restore_time >= week_start,
|
|
WorkOrderHandling.restore_time < week_end
|
|
).all()
|
|
|
|
for handling in handlings:
|
|
device = db.session.get(Device, handling.device_id) # 更新为新的方法
|
|
if device:
|
|
distance = float(handling.distance_from_branch)
|
|
reward = get_reward_by_distance(device.project_type, distance)
|
|
handler_stats[handling.handler] += reward
|
|
|
|
# 统计手工故障单奖励
|
|
manual_orders_list = ManualWorkOrder.query.filter(
|
|
ManualWorkOrder.create_time >= week_start,
|
|
ManualWorkOrder.create_time < week_end
|
|
).all()
|
|
|
|
for order in manual_orders_list:
|
|
distance = float(order.distance_from_branch)
|
|
reward = get_reward_by_distance(order.project_type, distance)
|
|
handler_stats[order.handler] += reward
|
|
|
|
# 获取前三名
|
|
top_handlers = sorted(handler_stats.items(), key=lambda x: x[1], reverse=True)[:3]
|
|
|
|
# 准备统计数据
|
|
stats = {
|
|
'total_orders': total_orders,
|
|
'completed_orders': completed_orders,
|
|
'pending_orders': total_orders - completed_orders,
|
|
'manual_orders': manual_orders,
|
|
'top_handlers': top_handlers
|
|
}
|
|
|
|
# 发送统计消息
|
|
send_weekly_stats_message(LZMX_WEBHOOK_URL, stats)
|
|
|
|
if __name__ == '__main__':
|
|
weekly_stats() |