feat: 优化补录逻辑和班次时间

修复:
- 修复 send_notification 函数中 now.strftime() 调用错误

新功能:
- 补录命令支持指定时间: 补录 日期 姓名 内容 [时间]
- 不指定时间则使用班次下班前2分钟
- 班次统一为 18:00-02:00(次日),默认补录时间 01:58

代码优化:
- 添加文件头注释
- 函数添加 docstring
- 移除调试代码
This commit is contained in:
OpenClaw Agent
2026-02-07 18:09:16 +00:00
parent 92c62b9da5
commit b4b7f8ff1e

View File

@@ -1,3 +1,9 @@
#!/usr/bin/env python3
"""
违禁品查获排行榜系统 - 数据管理脚本
支持业务日期自动校准、WebDAV备份、企业微信通知
"""
import json import json
import os import os
import re import re
@@ -75,8 +81,9 @@ def format_name(name):
def send_notification(action_text, data): def send_notification(action_text, data):
"""发送企业微信通知 - v2.3.8 对齐增强与删除功能 + 新业务日期""" """发送企业微信通知 - v2.3.8 对齐增强与删除功能 + 新业务日期"""
if not WEBHOOK_URL: return if not WEBHOOK_URL: return
now = get_now().strftime("%Y-%m-%d %H:%M") now = get_now()
biz_month = get_now().strftime("%Y-%m") now_str = now.strftime("%Y-%m-%d %H:%M")
biz_month = now.strftime("%Y-%m")
# 1. 统计当月排名 (过滤0票) # 1. 统计当月排名 (过滤0票)
stats = {} stats = {}
@@ -136,28 +143,53 @@ def remove_member(name):
save_data(data, action_text) save_data(data, action_text)
return f"✅ 已去除人员:{name} (已标记为离职,不参与排名)" return f"✅ 已去除人员:{name} (已标记为离职,不参与排名)"
def record_entry(name, content, biz_date_manual=None): def record_entry(name, content, biz_date_manual=None, biz_time=None):
"""录入查获记录
Args:
name: 人员姓名
content: 查获内容
biz_date_manual: 补录的业务日期 (MM-DD格式)None表示普通录入
biz_time: 补录的指定时间 (HH:MM格式)可选默认为班次下班前2分钟
"""
data = load_data() data = load_data()
if name not in data["members"] or data["members"][name]["status"] == "offboarded": if name not in data["members"] or data["members"][name]["status"] == "offboarded":
return f"错误:未找到活跃人员 '{name}' 或该人员已离职。" return f"错误:未找到活跃人员 '{name}' 或该人员已离职。"
now = get_now() now = get_now()
is_manual = False
if biz_date_manual: if biz_date_manual:
# 补录模式
year = now.year year = now.year
dt = datetime.strptime(f"{year}{biz_date_manual.replace('-','')}", "%Y%m%d").replace(tzinfo=BEIJING_TZ) dt = datetime.strptime(f"{year}{biz_date_manual.replace('-','')}", "%Y%m%d").replace(tzinfo=BEIJING_TZ)
info = {"biz_date": dt.strftime("%Y-%m-%d"), "biz_month": dt.strftime("%Y-%m"), "is_manual": True} info = {"biz_date": dt.strftime("%Y-%m-%d"), "biz_month": dt.strftime("%Y-%m"), "is_manual": True}
is_manual = True
# 计算 actual_time
if biz_time:
# 使用指定时间
hour, minute = map(int, biz_time.split(':'))
actual_dt = dt.replace(hour=hour, minute=minute, second=0, microsecond=0)
else:
# 使用班次下班前2分钟
shift = data["members"][name].get("shift", "")
# 班次统一为 18:00-02:00(次日)默认下班前2分钟
actual_dt = dt.replace(hour=1, minute=58, second=0, microsecond=0)
actual_time = actual_dt.isoformat()
else: else:
# 修正业务日期判定逻辑00:00-05:00 归属到前一个日历日 # 普通录入模式
# 重点:判断当前小时是否在 0 到 5 之间(包含 0 和 5
biz_dt = now - timedelta(days=1) if 0 <= now.hour <= 5 else now biz_dt = now - timedelta(days=1) if 0 <= now.hour <= 5 else now
info = {"biz_date": biz_dt.strftime("%Y-%m-%d"), "biz_month": biz_dt.strftime("%Y-%m"), "is_manual": False} info = {"biz_date": biz_dt.strftime("%Y-%m-%d"), "biz_month": biz_dt.strftime("%Y-%m"), "is_manual": False}
actual_time = now.isoformat()
data["last_id"] += 1 data["last_id"] += 1
new_id = data["last_id"] new_id = data["last_id"]
log = {"id": new_id, "name": name, "content": content, "actual_time": now.isoformat(), "biz_date": info["biz_date"], "biz_month": info["biz_month"]} log = {"id": new_id, "name": name, "content": content, "actual_time": actual_time, "biz_date": info["biz_date"], "biz_month": info["biz_month"]}
data["logs"].append(log) data["logs"].append(log)
tag = "[补录] " if info.get("is_manual") else "" tag = "[补录] " if is_manual else ""
action_text = f"{tag}新增 {format_name(name)} {content}" action_text = f"{tag}新增 {format_name(name)} {content}"
save_data(data, action_text) save_data(data, action_text)
@@ -169,7 +201,7 @@ def delete_entry(log_id):
found = False found = False
for log in data["logs"]: for log in data["logs"]:
if log['id'] == log_id: if log['id'] == log_id:
log['is_deleted'] = True # 标记为删除,而不是真正移除 log['is_deleted'] = True
found = True found = True
break break
if found: if found:
@@ -212,7 +244,6 @@ def get_ranking(target_month=None):
if __name__ == "__main__": if __name__ == "__main__":
args = sys.argv[1:] args = sys.argv[1:]
# print(f"DEBUG: Raw args received: {args}") # 再次调试打印
if not args: print(f"安检系统 {VERSION}"); sys.exit(0) if not args: print(f"安检系统 {VERSION}"); sys.exit(0)
@@ -222,12 +253,28 @@ if __name__ == "__main__":
# 优先检查新增成员命令 # 优先检查新增成员命令
elif len(args) >= 2 and args[0] == "新增": elif len(args) >= 2 and args[0] == "新增":
print(add_member(" ".join(args[1:]))) print(add_member(" ".join(args[1:])))
# 优先检查删除命令,直接从 args 列表解析 # 优先检查删除命令
elif len(args) >= 2 and args[0] == "删除" and args[1].startswith("#"): elif len(args) >= 2 and args[0] == "删除" and args[1].startswith("#"):
try: try:
log_id = int(args[1][1:].strip()) log_id = int(args[1][1:].strip())
print(delete_entry(log_id)) print(delete_entry(log_id))
except ValueError: print("错误:删除指令格式为 '删除 #索引号'") except ValueError: print("错误:删除指令格式为 '删除 #索引号'")
# 补录命令:补录 日期 姓名 内容 [时间 HH:MM]
elif len(args) >= 4 and args[0] == "补录":
biz_date = args[1]
# 验证日期格式
if not re.match(r'^\d{2}-?\d{2}$', biz_date):
print(f"错误:日期格式不正确,应为 MM-DD 或 MMDD如 02-06")
sys.exit(1)
name = args[2]
# 检查是否有时间参数(格式 HH:MM在倒数第二个位置
if len(args) >= 5 and re.match(r'^\d{2}:\d{2}$', args[-1]):
content = " ".join(args[3:-1])
biz_time = args[-1]
else:
content = " ".join(args[3:])
biz_time = None # 默认使用班次下班前2分钟
print(record_entry(name, content, biz_date, biz_time))
# 检查排行榜命令 # 检查排行榜命令
elif "排行榜" in " ".join(args): elif "排行榜" in " ".join(args):
m = re.search(r"排行榜\s*(.*)", " ".join(args)) m = re.search(r"排行榜\s*(.*)", " ".join(args))