diff --git a/app/config.py b/app/config.py
index 10e2613..98a4fee 100644
--- a/app/config.py
+++ b/app/config.py
@@ -63,6 +63,7 @@ class WeChatConfig:
test_uids: str = "" # 演示模式：逗号分隔的企微userid，如 "zhangsan,lisi"
service_base_url: str = "" # 公网地址，如 https://vsp.viewshanghai.com
group_chat_id: str = "" # 告警群聊ID（通过企微API创建或手动指定）
+ group_robot_key: str = "" # 群机器人 Webhook key（用于日报等模板卡片推送）
@dataclass
@@ -184,6 +185,7 @@ def load_settings() -> Settings:
test_uids=os.getenv("WECHAT_TEST_UIDS", ""),
service_base_url=os.getenv("SERVICE_BASE_URL", ""),
group_chat_id=os.getenv("WECHAT_GROUP_CHAT_ID", ""),
+ group_robot_key=os.getenv("WECHAT_GROUP_ROBOT_KEY", ""),
),
agent=AgentConfig(
vlm_api_key=os.getenv("DASHSCOPE_API_KEY", ""),
diff --git a/app/services/daily_report_service.py b/app/services/daily_report_service.py
index b98650d..e95774b 100644
--- a/app/services/daily_report_service.py
+++ b/app/services/daily_report_service.py
@@ -299,8 +299,64 @@ async def _build_daily_report_data() -> Optional[Dict]:
return report
+def _build_template_card(report: Dict) -> Dict:
+ """构建 text_notice 模板卡片（群机器人 Webhook 专用）"""
+ s = report["summary"]
+ tops = report["tops"]
+ click_url = settings.wechat.service_base_url or "https://work.weixin.qq.com"
+
+ # 大号数字：昨日新增
+ emphasis_desc = f"昨日新增（{report['change_str']}）"
+
+ # 键值对列表（最多 6 条，挑最重要的）
+ kv_list = [
+ {"keyname": "安保 / 保洁", "value": f"{s['security_count']} / {s['clean_count']}"},
+ {"keyname": "已完成", "value": f"{s['completed_count']}"},
+ {"keyname": "待处理", "value": f"{s['backlog_count']}（遗留 {s['carry_over_count']}）" if s['backlog_count'] > 0 else "0 全部清零"},
+ {"keyname": "首响 / 完结", "value": f"{s['avg_resp']} / {s['avg_close']}"},
+ ]
+
+ # 第5条：最高发告警类型
+ if tops["alarm_types"] != "暂无数据":
+ kv_list.append({"keyname": "告警热点", "value": tops["alarm_types"]})
+
+ # 第6条：高发设备
+ if tops["cameras"] != "暂无数据":
+ kv_list.append({"keyname": "高发设备", "value": tops["cameras"]})
+
+ # 副标题
+ sub_title = ""
+ if report["top_overdue"]:
+ sub_title = f"需关注：{report['top_overdue'][0]}"
+ elif s["backlog_count"] == 0:
+ sub_title = "昨日工单全部处理完毕，运营良好"
+
+ card = {
+ "card_type": "text_notice",
+ "source": {
+ "desc": "VSP物业平台",
+ "desc_color": 0,
+ },
+ "main_title": {
+ "title": report["title"],
+ },
+ "emphasis_content": {
+ "title": str(s["yesterday_total"]),
+ "desc": emphasis_desc,
+ },
+ "sub_title_text": sub_title,
+ "horizontal_content_list": kv_list,
+ "card_action": {
+ "type": 1,
+ "url": click_url,
+ },
+ }
+
+ return card
+
+
def _build_markdown(report: Dict) -> str:
- """构建单条企微 markdown 日报（领导视角，简洁直观）"""
+ """构建单条企微 markdown 日报（降级方案）"""
if report.get("empty"):
return (
f"**{report['title']}**\n\n"
@@ -314,13 +370,11 @@ def _build_markdown(report: Dict) -> str:
lines = [
f"**{report['title']}**",
"",
- # ── 总览 ──
f">昨日新增 {s['yesterday_total']} 条（{report['change_str']}）",
f">安保 {s['security_count']}｜保洁 {s['clean_count']}｜"
f"已完成 {s['completed_count']}｜误报 {s['false_alarm_rate']}",
]
- # 待处理：0 用绿色，>0 用橙色警示
if backlog == 0:
lines.append(f">待处理 0 条，全部清零")
else:
@@ -329,14 +383,12 @@ def _build_markdown(report: Dict) -> str:
f"（其中遗留 {s['carry_over_count']}）"
)
- # ── 效率 ──
lines.append("")
lines.append(
f">响应效率：首响 {s['avg_resp']}"
f"｜完结 {s['avg_close']}"
)
- # ── 风险分布（仅有数据时展示）──
tops = report["tops"]
risk_items = []
if tops["alarm_types"] != "暂无数据":
@@ -353,7 +405,6 @@ def _build_markdown(report: Dict) -> str:
lines.append("**热点分布**")
lines.extend(risk_items)
- # ── 超时跟进（仅有遗留时展示）──
if report["top_overdue"]:
lines.append("")
lines.append(f"**需关注（{s['carry_over_count']}条超时）**")
@@ -364,7 +415,7 @@ def _build_markdown(report: Dict) -> str:
async def generate_daily_report() -> Optional[str]:
- """生成日报 markdown 内容（供预览和发送）"""
+ """生成日报 markdown 内容（供预览和降级发送）"""
report = await _build_daily_report_data()
if not report:
return None
@@ -372,26 +423,39 @@ async def generate_daily_report() -> Optional[str]:
async def _send_daily_report():
- """生成并发送单条 markdown 日报到企微群聊"""
+ """发送日报：优先用群机器人 Webhook 模板卡片，降级为 markdown"""
from app.services.wechat_service import get_wechat_service
chat_id = settings.wechat.group_chat_id
- if not chat_id:
- logger.warning("日报发送跳过：未配置 group_chat_id")
+ robot_key = settings.wechat.group_robot_key
+ if not chat_id and not robot_key:
+ logger.warning("日报发送跳过：未配置 group_chat_id 或 group_robot_key")
return
try:
- content = await generate_daily_report()
- if not content:
+ report = await _build_daily_report_data()
+ if not report:
logger.info("日报生成内容为空，跳过发送")
return
wechat_svc = get_wechat_service()
- ok = await wechat_svc.send_group_markdown(chat_id, content)
- if ok:
- logger.info("日报已发送到企微群聊")
- else:
- logger.error("日报发送失败")
+
+ # 优先：群机器人 Webhook 发送 text_notice 模板卡片
+ if robot_key:
+ card = _build_template_card(report)
+ ok = await wechat_svc.send_webhook_template_card(robot_key, card)
+ if ok:
+ logger.info("日报模板卡片已通过 Webhook 发送")
+ return
+
+ # 降级：应用群聊发送 markdown
+ if chat_id:
+ content = _build_markdown(report)
+ ok = await wechat_svc.send_group_markdown(chat_id, content)
+ if ok:
+ logger.info("日报已通过 markdown 发送到群聊")
+ else:
+ logger.error("日报发送失败")
except Exception:
logger.exception("日报生成或发送异常")
diff --git a/app/services/wechat_service.py b/app/services/wechat_service.py
index b16b92b..5d90593 100644
--- a/app/services/wechat_service.py
+++ b/app/services/wechat_service.py
@@ -617,6 +617,29 @@ class WeChatService:
logger.error(f"发送群聊文本卡片异常: {e}")
return False
+ async def send_webhook_template_card(self, webhook_key: str, card: Dict) -> bool:
+ """通过群机器人 Webhook 发送 template_card（text_notice / news_notice）"""
+ if not webhook_key:
+ logger.warning("Webhook key 为空，跳过发送")
+ return False
+ url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={webhook_key}"
+ payload = {
+ "msgtype": "template_card",
+ "template_card": card,
+ }
+ try:
+ async with httpx.AsyncClient(timeout=10) as client:
+ resp = await client.post(url, json=payload)
+ data = resp.json()
+ if data.get("errcode") != 0:
+ logger.error(f"Webhook template_card 发送失败: {data}")
+ return False
+ logger.info("Webhook template_card 已发送")
+ return True
+ except Exception as e:
+ logger.error(f"Webhook template_card 异常: {e}")
+ return False
+
async def send_group_image(self, chat_id: str, media_id: str) -> bool:
"""发送图片消息到群聊"""
if not self._enabled: