290 lines
9.5 KiB
Python
290 lines
9.5 KiB
Python
"""
|
||
每日数据汇总 - 每天 23:59 自动发送到企业微信
|
||
"""
|
||
import os
|
||
import asyncio
|
||
import logging
|
||
from datetime import datetime, time as dtime
|
||
from collections import defaultdict
|
||
|
||
import httpx
|
||
from dotenv import load_dotenv
|
||
|
||
load_dotenv()
|
||
|
||
logger = logging.getLogger("cs_agent")
|
||
|
||
|
||
async def generate_daily_report() -> str:
|
||
"""生成今日数据报告"""
|
||
import pymysql
|
||
|
||
# 从环境变量读取数据库配置
|
||
MYSQL_HOST = os.getenv("MYSQL_HOST", "127.0.0.1")
|
||
MYSQL_PORT = int(os.getenv("MYSQL_PORT", "3306"))
|
||
MYSQL_USER = os.getenv("MYSQL_USER", "root")
|
||
MYSQL_PASSWORD = os.getenv("MYSQL_PASSWORD", "")
|
||
MYSQL_DATABASE = os.getenv("MYSQL_DATABASE", "ai_cs")
|
||
|
||
try:
|
||
conn = pymysql.connect(
|
||
host=MYSQL_HOST,
|
||
port=MYSQL_PORT,
|
||
user=MYSQL_USER,
|
||
password=MYSQL_PASSWORD,
|
||
database=MYSQL_DATABASE,
|
||
charset='utf8mb4',
|
||
cursorclass=pymysql.cursors.DictCursor
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"[DailySummary] 数据库连接失败: {e}")
|
||
return f"❌ 数据库连接失败: {e}"
|
||
|
||
try:
|
||
cur = conn.cursor()
|
||
today = datetime.now().strftime('%Y-%m-%d')
|
||
|
||
# 1. 基础统计
|
||
cur.execute(f"SELECT COUNT(DISTINCT customer_id) as cnt FROM chat_logs WHERE DATE(timestamp) = '{today}'")
|
||
total_customers = cur.fetchone()['cnt']
|
||
|
||
cur.execute(f"SELECT COUNT(*) as cnt FROM chat_logs WHERE DATE(timestamp) = '{today}'")
|
||
total_msgs = cur.fetchone()['cnt']
|
||
|
||
cur.execute(f"SELECT direction, COUNT(*) as cnt FROM chat_logs WHERE DATE(timestamp) = '{today}' GROUP BY direction")
|
||
dir_stats = {r['direction']: r['cnt'] for r in cur.fetchall()}
|
||
|
||
# 2. 获取所有对话
|
||
cur.execute(f'''
|
||
SELECT customer_id, customer_name, message, direction, timestamp
|
||
FROM chat_logs
|
||
WHERE DATE(timestamp) = '{today}'
|
||
ORDER BY customer_id, timestamp
|
||
''')
|
||
rows = cur.fetchall()
|
||
|
||
# 按客户分组
|
||
customers = defaultdict(list)
|
||
for r in rows:
|
||
cid = r['customer_id'] or 'unknown'
|
||
customers[cid].append(r)
|
||
|
||
# 统计
|
||
transferred = 0
|
||
paid = 0
|
||
shipped = 0
|
||
closed = 0
|
||
new_orders = 0
|
||
|
||
customer_details = []
|
||
for cid, msgs in customers.items():
|
||
cname = ''
|
||
has_transfer = False
|
||
has_paid = False
|
||
has_shipped = False
|
||
has_closed = False
|
||
has_new = False
|
||
need = ''
|
||
|
||
for m in msgs:
|
||
if m['customer_name']:
|
||
cname = m['customer_name']
|
||
msg = m['message'] or ''
|
||
|
||
if '[转移会话]' in msg:
|
||
has_transfer = True
|
||
if '买家已付款' in msg:
|
||
has_paid = True
|
||
if '卖家已发货' in msg:
|
||
has_shipped = True
|
||
if '订单关闭' in msg or '已取消' in msg:
|
||
has_closed = True
|
||
if '新订单' in msg:
|
||
has_new = True
|
||
|
||
# 识别需求
|
||
if m['direction'] == 'in' and not need:
|
||
if '找原图' in msg or '原图' in msg:
|
||
need = '找原图'
|
||
elif '修复' in msg or '高清' in msg:
|
||
need = '高清修复'
|
||
elif '找图' in msg:
|
||
need = '找图'
|
||
|
||
if has_transfer:
|
||
transferred += 1
|
||
if has_paid:
|
||
paid += 1
|
||
if has_shipped:
|
||
shipped += 1
|
||
if has_closed:
|
||
closed += 1
|
||
if has_new:
|
||
new_orders += 1
|
||
|
||
# 状态
|
||
if has_shipped:
|
||
status = '✅已发货'
|
||
elif has_paid:
|
||
status = '💰已付款'
|
||
elif has_closed:
|
||
status = '❌已关闭'
|
||
elif has_transfer:
|
||
status = '→已转接'
|
||
else:
|
||
status = '💬对话中'
|
||
|
||
display_name = cname if cname and cname != 'unknown' else cid[:12]
|
||
if display_name != 'unknown' and cid != 'unknown':
|
||
customer_details.append((display_name, need or '-', status))
|
||
|
||
# 构建报告
|
||
report = f'''📊 【{today} 客服数据日报】
|
||
|
||
📈 基础数据
|
||
• 独立客户数: {total_customers}
|
||
• 总消息数: {total_msgs}
|
||
- 客户发来: {dir_stats.get('in', 0)} 条
|
||
- AI/人工回复: {dir_stats.get('out', 0)} 条
|
||
|
||
💼 业务数据
|
||
• 新订单: {new_orders} 单
|
||
• 已付款: {paid} 单
|
||
• 已发货: {shipped} 单
|
||
• 订单关闭: {closed} 单
|
||
• 转接设计师: {transferred} 次
|
||
|
||
📝 客户明细 ({len(customer_details)} 条)
|
||
'''
|
||
|
||
# 按状态分组显示
|
||
shipped_list = [c for c in customer_details if '已发货' in c[2]]
|
||
paid_list = [c for c in customer_details if '已付款' in c[2]]
|
||
transfer_list = [c for c in customer_details if '已转接' in c[2]]
|
||
closed_list = [c for c in customer_details if '已关闭' in c[2]]
|
||
|
||
if shipped_list:
|
||
report += '\n✅ 已发货:\n'
|
||
for name, need, _ in shipped_list[:15]:
|
||
report += f' • {name} ({need})\n'
|
||
if len(shipped_list) > 15:
|
||
report += f' ... 等共 {len(shipped_list)} 单\n'
|
||
|
||
if paid_list:
|
||
report += '\n💰 已付款待发货:\n'
|
||
for name, need, _ in paid_list[:10]:
|
||
report += f' • {name} ({need})\n'
|
||
|
||
if transfer_list:
|
||
report += '\n→ 已转接设计师:\n'
|
||
for name, need, _ in transfer_list[:15]:
|
||
report += f' • {name} ({need})\n'
|
||
if len(transfer_list) > 15:
|
||
report += f' ... 等共 {len(transfer_list)} 人\n'
|
||
|
||
if closed_list:
|
||
report += '\n❌ 订单关闭:\n'
|
||
for name, need, _ in closed_list[:5]:
|
||
report += f' • {name}\n'
|
||
|
||
return report
|
||
|
||
except Exception as e:
|
||
logger.error(f"[DailySummary] 生成报告失败: {e}")
|
||
return f"❌ 生成报告失败: {e}"
|
||
finally:
|
||
if 'conn' in locals():
|
||
conn.close()
|
||
|
||
|
||
async def send_to_wechat(content: str) -> bool:
|
||
"""发送到企业微信"""
|
||
from config.config import WECHAT_WEBHOOK
|
||
|
||
if not WECHAT_WEBHOOK:
|
||
logger.warning("[DailySummary] 未配置企业微信 Webhook")
|
||
return False
|
||
|
||
try:
|
||
async with httpx.AsyncClient() as client:
|
||
resp = await client.post(
|
||
WECHAT_WEBHOOK,
|
||
json={'msgtype': 'text', 'text': {'content': content}},
|
||
timeout=10
|
||
)
|
||
if resp.status_code == 200:
|
||
logger.info("[DailySummary] 日报已发送到企业微信")
|
||
return True
|
||
else:
|
||
logger.error(f"[DailySummary] 发送失败: {resp.status_code}")
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"[DailySummary] 发送异常: {e}")
|
||
return False
|
||
|
||
|
||
async def run_daily_summary():
|
||
"""执行日报任务"""
|
||
logger.info("[DailySummary] 开始生成每日日报...")
|
||
report = await generate_daily_report()
|
||
await send_to_wechat(report)
|
||
|
||
|
||
def _seconds_until(target_hour: int, target_minute: int) -> float:
|
||
"""计算距离目标时间的秒数"""
|
||
now = datetime.now()
|
||
target = now.replace(hour=target_hour, minute=target_minute, second=0, microsecond=0)
|
||
|
||
# 如果今天的目标时间已过，则计算到明天
|
||
if now >= target:
|
||
target = target.replace(day=target.day + 1)
|
||
# 处理月末
|
||
try:
|
||
target = target
|
||
except ValueError:
|
||
# 下个月第一天
|
||
if target.month == 12:
|
||
target = target.replace(year=target.year + 1, month=1, day=1)
|
||
else:
|
||
target = target.replace(month=target.month + 1, day=1)
|
||
|
||
return (target - now).total_seconds()
|
||
|
||
|
||
async def scheduler():
|
||
"""
|
||
定时任务调度器 - 每天 23:59 执行日报
|
||
"""
|
||
logger.info("[DailySummary] 每日日报定时任务已启动，将在每天 23:59 发送")
|
||
|
||
while True:
|
||
try:
|
||
# 计算距离 23:59 的秒数
|
||
wait_seconds = _seconds_until(23, 59)
|
||
hours = int(wait_seconds // 3600)
|
||
minutes = int((wait_seconds % 3600) // 60)
|
||
logger.info(f"[DailySummary] 下次日报将在 {hours}小时{minutes}分钟 后发送")
|
||
|
||
# 等待到目标时间
|
||
await asyncio.sleep(wait_seconds)
|
||
|
||
# 执行日报
|
||
await run_daily_summary()
|
||
|
||
# 等待1分钟，避免重复执行
|
||
await asyncio.sleep(60)
|
||
|
||
except asyncio.CancelledError:
|
||
logger.info("[DailySummary] 定时任务已取消")
|
||
break
|
||
except Exception as e:
|
||
logger.error(f"[DailySummary] 定时任务异常: {e}")
|
||
# 出错后等待5分钟再重试
|
||
await asyncio.sleep(300)
|
||
|
||
|
||
# 手动触发（用于测试）
|
||
async def manual_trigger():
|
||
"""手动触发日报（测试用）"""
|
||
await run_daily_summary()
|