From 36ba9338f1b0704fa83dc9190eee9c860bde7861 Mon Sep 17 00:00:00 2001 From: v6ole Date: Fri, 26 Jun 2026 09:37:13 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=BC=81=E4=B8=9A=E5=BE=AE=E4=BF=A1=20?= =?UTF-8?q?Phase=201=20=E2=80=94=20token=20=E7=BB=AD=E6=9C=9F=20+=20?= =?UTF-8?q?=E5=9B=9E=E8=B0=83=E9=AA=8C=E8=AF=81=20+=20=E8=B0=83=E5=BA=A6?= =?UTF-8?q?=E5=99=A8=E6=B3=A8=E5=86=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit wecom.py: - Token 自动续期 (7200s 过期 + 60s 提前量 + 42001 重试) - POST 请求自动重试 (最多 3 次，含速率限制退避) - 新增 send_markdown_message() / send_template_card() 富消息方法 wecom.py API: - GET /api/wecom/callback: 企微回调 URL 验证 (SHA1 签名 + AES 解密) - POST /api/wecom/callback: 事件接收占位 scheduler.py: - 仅推送给已绑定 wecom_userid 的经理 - 消息内容优化 (已填报/未填报人数统计) main.py: - APScheduler 注册每日 17:30 自动检查填报 已验证: WECOM_TOKEN 获取成功 Co-Authored-By: Claude --- backend/app/api/wecom.py | 70 +++++++++++++- backend/app/main.py | 19 +++- backend/app/services/scheduler.py | 29 ++++-- backend/app/services/wecom.py | 148 +++++++++++++++++++++++++----- backend/requirements.txt | 1 + 5 files changed, 232 insertions(+), 35 deletions(-) diff --git a/backend/app/api/wecom.py b/backend/app/api/wecom.py index 5f03204..eddbd14 100644 --- a/backend/app/api/wecom.py +++ b/backend/app/api/wecom.py @@ -1,9 +1,13 @@ +import hashlib +import struct +import base64 import uuid from typing import Optional -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Query, Request from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select +from app.config import settings from app.database import get_db from app.middleware.auth import get_current_user, require_director from app.models.user import User @@ -22,6 +26,68 @@ class AnnouncementRequest(BaseModel): content: str +# ── Callback verification helpers ── + +def _verify_signature(token: str, timestamp: str, nonce: str, encrypted: str, signature: str) -> bool: + """Verify WeChat Work callback signature.""" + params = sorted([token, timestamp, nonce, encrypted]) + raw = "".join(params) + computed = hashlib.sha1(raw.encode()).hexdigest() + return computed == signature + + +def _decrypt_echostr(encrypted: str) -> str: + """Decrypt WeChat Work callback echostr. Returns plaintext.""" + key = base64.b64decode(settings.WECOM_ENCODING_AES_KEY + "=") + ciphertext = base64.b64decode(encrypted) + + # AES-256-CBC decrypt + from Crypto.Cipher import AES + cipher = AES.new(key, AES.MODE_CBC, iv=key[:16]) + plaintext = cipher.decrypt(ciphertext) + + # Strip PKCS#7 padding + pad_len = plaintext[-1] + plaintext = plaintext[:-pad_len] + + # Format: random(16) + msg_len(4) + msg + corpid + msg_len = struct.unpack(">I", plaintext[16:20])[0] + msg = plaintext[20:20 + msg_len].decode("utf-8") + return msg + + +@router.get("/callback") +async def wecom_callback_verify( + msg_signature: str = Query(...), + timestamp: str = Query(...), + nonce: str = Query(...), + echostr: str = Query(...), +): + """WeChat Work callback URL verification (GET).""" + token = settings.WECOM_TOKEN + if not token or not settings.WECOM_ENCODING_AES_KEY: + raise HTTPException(status_code=500, detail="WeCom callback not configured") + + if not _verify_signature(token, timestamp, nonce, echostr, msg_signature): + raise HTTPException(status_code=403, detail="Signature verification failed") + + try: + plaintext = _decrypt_echostr(echostr) + return plaintext + except Exception as e: + raise HTTPException(status_code=500, detail=f"Decrypt failed: {e}") + + +@router.post("/callback") +async def wecom_callback_event(request: Request): + """WeChat Work callback event receiver (POST). Placeholder for future use.""" + body = await request.body() + # TODO: decrypt and handle events (app install, user enter, etc.) + return "ok" + + +# ── Manual actions ── + @router.post("/remind") async def send_reminder( data: RemindRequest, @@ -29,7 +95,6 @@ async def send_reminder( db: AsyncSession = Depends(get_db), ): """Director manually sends reminder to specific managers.""" - # Get wecom_userids for the selected users result = await db.execute( select(User.wecom_userid).where(User.id.in_([uuid.UUID(uid) for uid in data.user_ids])) ) @@ -48,7 +113,6 @@ async def send_announcement( db: AsyncSession = Depends(get_db), ): """Director sends an announcement to all team members.""" - # Get all wecom_userids in the department result = await db.execute(select(User.wecom_userid).where(User.wecom_userid.isnot(None))) wecom_ids = [r[0] for r in result.all()] diff --git a/backend/app/main.py b/backend/app/main.py index 6c394c1..49d7002 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,11 +1,21 @@ from contextlib import asynccontextmanager from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware +from apscheduler.schedulers.asyncio import AsyncIOScheduler from app.config import settings -from app.database import engine, Base +from app.database import engine, Base, async_session from app.api import router as api_router from app.api import auth, users, customers, visits, work_plans, mini_business, key_visits from app.api import dashboard, upload, export, import_data, wecom, daily_notes, ai_summary +from app.services.scheduler import check_daily_reporting + +_scheduler = AsyncIOScheduler() + + +async def _scheduled_check(): + """Wrapper for APScheduler: create a fresh session and run the daily check.""" + async with async_session() as db: + await check_daily_reporting(db) @asynccontextmanager @@ -28,8 +38,15 @@ async def lifespan(app: FastAPI): await conn.run_sync(lambda c, t=tbl: c.exec_driver_sql( f"ALTER TABLE {t} ADD COLUMN IF NOT EXISTS edit_log JSONB DEFAULT '[]'" )) + + # Start daily reporting scheduler (17:30 CST = 09:30 UTC) + _scheduler.add_job(_scheduled_check, "cron", hour=17, minute=30, id="daily_check") + _scheduler.start() + yield + # Shutdown + _scheduler.shutdown(wait=False) await engine.dispose() diff --git a/backend/app/services/scheduler.py b/backend/app/services/scheduler.py index a70f2c5..a2d8c02 100644 --- a/backend/app/services/scheduler.py +++ b/backend/app/services/scheduler.py @@ -15,8 +15,10 @@ async def check_daily_reporting(db: AsyncSession) -> dict: if weekday >= 5: # Skip weekends return {"status": "weekend", "date": str(today)} - # Get all managers - result = await db.execute(select(User).where(User.role == "manager")) + # Get all managers with wecom_userid + result = await db.execute( + select(User).where(User.role == "manager", User.wecom_userid.isnot(None)) + ) managers = result.scalars().all() # Get managers who have reported today @@ -31,16 +33,25 @@ async def check_daily_reporting(db: AsyncSession) -> dict: reported_map[str(uid)] = True not_reported = [] + reported_names = [] for m in managers: - if str(m.id) not in reported_map: + if str(m.id) in reported_map: + reported_names.append(m.name) + else: not_reported.append(m) - if not_reported and managers: - content = f"📋 今日填报提醒（{today}）\n\n以下同事尚未提交今日拜访记录：\n" - for m in not_reported: - content += f"• {m.name}\n" - content += "\n请尽快完成今日拜访填报 🙏" - + # Send markdown message to not-reported managers + if not_reported: + names = "、".join(m.name for m in not_reported) + content = ( + f"## 📋 今日填报提醒\n\n" + f"> 日期：{today}\n" + f"> 已填报：{len(reported_names)} 人\n" + f"> 未填报：**{len(not_reported)} 人**\n\n" + f"以下同事尚未提交今日拜访记录：\n" + + "".join(f"- **{m.name}**\n" for m in not_reported) + + f"\n请尽快完成今日拜访填报 🙏" + ) user_ids = [m.wecom_userid for m in not_reported if m.wecom_userid] if user_ids: await wecom_client.send_text_message(user_ids, content) diff --git a/backend/app/services/wecom.py b/backend/app/services/wecom.py index 9f51b38..d2fcc40 100644 --- a/backend/app/services/wecom.py +++ b/backend/app/services/wecom.py @@ -1,55 +1,159 @@ +import time import httpx from app.config import settings class WecomClient: - """Minimal WeChat Work API client for sending app messages.""" + """WeChat Work API client — token management, message sending, OAuth.""" def __init__(self): self.corp_id = settings.WECOM_CORP_ID self.agent_id = settings.WECOM_AGENT_ID self.secret = settings.WECOM_SECRET self._access_token: str | None = None + self._token_expires_at: float = 0 # epoch seconds async def _get_token(self) -> str: - if self._access_token: + """Get a valid access token, refreshing if expired.""" + # Token valid for 7200s; refresh 60s early to be safe + if self._access_token and time.time() < self._token_expires_at - 60: return self._access_token + url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={self.corp_id}&corpsecret={self.secret}" async with httpx.AsyncClient() as client: resp = await client.get(url, timeout=10) data = resp.json() if data.get("errcode") == 0: self._access_token = data["access_token"] + self._token_expires_at = time.time() + data.get("expires_in", 7200) return self._access_token raise Exception(f"Failed to get wecom token: {data}") + async def _post_with_retry(self, url: str, body: dict, max_retries: int = 3) -> dict: + """POST with retry on network errors and automatic token refresh on 42001.""" + last_error = None + for attempt in range(max_retries): + try: + token = await self._get_token() + full_url = f"{url}?access_token={token}" + async with httpx.AsyncClient() as client: + resp = await client.post(full_url, json=body, timeout=10) + data = resp.json() + errcode = data.get("errcode", 0) + + if errcode == 0: + return data + + # Token expired mid-request — clear and retry once + if errcode == 42001: + self._access_token = None + self._token_expires_at = 0 + if attempt < max_retries - 1: + continue + + # Rate limit — wait and retry + if errcode == 45009: + if attempt < max_retries - 1: + await httpx.AsyncClient().aclose() + time.sleep(1 * (attempt + 1)) + continue + + last_error = data + except (httpx.TimeoutException, httpx.ConnectError) as e: + last_error = {"errcode": -1, "errmsg": str(e)} + if attempt < max_retries - 1: + time.sleep(0.5 * (attempt + 1)) + + raise Exception(f"WeCom API error after {max_retries} retries: {last_error}") + async def get_userinfo_by_code(self, code: str) -> dict | None: """Exchange OAuth2 code for userid (used in silent login).""" - token = await self._get_token() - url = f"https://qyapi.weixin.qq.com/cgi-bin/auth/getuserinfo?access_token={token}&code={code}" - async with httpx.AsyncClient() as client: - resp = await client.get(url, timeout=10) - data = resp.json() - if data.get("errcode") == 0: - return data + try: + token = await self._get_token() + url = f"https://qyapi.weixin.qq.com/cgi-bin/auth/getuserinfo?access_token={token}&code={code}" + async with httpx.AsyncClient() as client: + resp = await client.get(url, timeout=10) + data = resp.json() + if data.get("errcode") == 0: + return data + return None + except Exception: return None async def send_text_message(self, user_ids: list[str], content: str) -> bool: """Send a text app message to specified users.""" if not settings.WECOM_AGENT_ID: - return False # Not configured, skip silently in dev - token = await self._get_token() - url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}" - body = { - "touser": "|".join(user_ids), - "msgtype": "text", - "agentid": int(settings.WECOM_AGENT_ID), - "text": {"content": content}, - } - async with httpx.AsyncClient() as client: - resp = await client.post(url, json=body, timeout=10) - data = resp.json() - return data.get("errcode") == 0 + return False + try: + body = { + "touser": "|".join(user_ids), + "msgtype": "text", + "agentid": int(settings.WECOM_AGENT_ID), + "text": {"content": content}, + } + await self._post_with_retry( + "https://qyapi.weixin.qq.com/cgi-bin/message/send", body + ) + return True + except Exception: + return False + + async def send_markdown_message(self, content: str) -> bool: + """Send a markdown message to all users in the app (broadcast).""" + if not settings.WECOM_AGENT_ID: + return False + try: + body = { + "touser": "@all", + "msgtype": "markdown", + "agentid": int(settings.WECOM_AGENT_ID), + "markdown": {"content": content}, + } + await self._post_with_retry( + "https://qyapi.weixin.qq.com/cgi-bin/message/send", body + ) + return True + except Exception: + return False + + async def send_template_card( + self, + user_ids: list[str], + title: str, + desc: str, + url: str, + btn_text: str = "查看详情", + ) -> bool: + """Send a text_notice template card with a deep-link button.""" + if not settings.WECOM_AGENT_ID: + return False + try: + body = { + "touser": "|".join(user_ids), + "msgtype": "template_card", + "agentid": int(settings.WECOM_AGENT_ID), + "template_card": { + "card_type": "text_notice", + "main_title": {"title": title, "desc": desc}, + "card_action": { + "type": 1, # jump to URL + "url": url, + }, + "button_list": [ + { + "text": btn_text, + "style": 1, # primary + "key": "open_url", + } + ], + }, + } + await self._post_with_retry( + "https://qyapi.weixin.qq.com/cgi-bin/message/send", body + ) + return True + except Exception: + return False wecom_client = WecomClient() diff --git a/backend/requirements.txt b/backend/requirements.txt index e830a11..a7eb09e 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -12,3 +12,4 @@ minio==7.2.10 openpyxl==3.1.5 apscheduler==3.11.0 python-dotenv==1.0.1 +pycryptodome