import os import json import time import hashlib import random import string from typing import Dict, Optional import httpx from dotenv import load_dotenv # 加载环境变量 load_dotenv() CACHE_FILE = "data.json" TOKEN_EXPIRY_MINUTES = 100 # 100分钟过期 TICKET_EXPIRY_SECONDS = 7200 # ticket过期时间7200秒 class WeChatAuthService: def __init__(self): self.corpid = os.getenv("CORPID") self.corpsecret = os.getenv("CORPSECRET") if not self.corpid or not self.corpsecret: raise ValueError("CORPID or CORPSECRET not found in .env file") self.api_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={self.corpid}&corpsecret={self.corpsecret}" def _load_cache(self) -> Optional[Dict]: """从缓存文件加载token数据""" try: if os.path.exists(CACHE_FILE): with open(CACHE_FILE, 'r', encoding='utf-8') as f: return json.load(f) except (json.JSONDecodeError, IOError) as e: print(f"加载缓存失败: {e}") return None def _save_cache(self, token_data: Dict) -> None: """保存token数据到缓存文件""" try: with open(CACHE_FILE, 'w', encoding='utf-8') as f: json.dump(token_data, f, ensure_ascii=False, indent=2) except IOError as e: print(f"保存缓存失败: {e}") def _is_token_valid(self, token_data: Dict) -> bool: """检查token是否有效（未过期）""" if not token_data: return False current_time = time.time() fetch_time = token_data.get('fetch_time', 0) expiry_seconds = TOKEN_EXPIRY_MINUTES * 60 # 检查是否在有效期内 return current_time < (fetch_time + expiry_seconds) def _is_ticket_valid(self, ticket_data: Dict) -> bool: """检查ticket是否有效（未过期）""" if not ticket_data: return False current_time = time.time() fetch_time = ticket_data.get('fetch_time', 0) expiry_seconds = ticket_data.get('expires_in', TICKET_EXPIRY_SECONDS) # 检查是否在有效期内 return current_time < (fetch_time + expiry_seconds) def _fetch_new_token(self) -> Dict: """从企业微信API获取新的token""" try: response = httpx.get(self.api_url, timeout=10.0) response.raise_for_status() result = response.json() if result.get('errcode') != 0: raise Exception(f"API错误: {result.get('errmsg', '未知错误')}") access_token = result.get('access_token') expires_in = result.get('expires_in', 7200) # 默认7200秒 token_data = { 'access_token': access_token, 'fetch_time': time.time(), 'expires_in': expires_in, 'expiry_time': time.time() + (TOKEN_EXPIRY_MINUTES * 60) } return token_data except httpx.RequestError as e: raise Exception(f"请求失败: {e}") except httpx.HTTPStatusError as e: raise Exception(f"HTTP错误: {e}") def get_access_token(self) -> str: """获取有效的access_token""" # 尝试从缓存加载 cached_data = self._load_cache() # 检查缓存是否有效 if cached_data and self._is_token_valid(cached_data): print("使用缓存的token") return cached_data['access_token'] # 获取新token print("获取新的token...") new_token_data = self._fetch_new_token() # 保存到缓存 self._save_cache(new_token_data) # 计算并显示过期时间 expiry_time = new_token_data['expiry_time'] expiry_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(expiry_time)) print(f"Token获取成功，过期时间: {expiry_str}") return new_token_data['access_token'] def get_token_info(self) -> Dict: """获取token详细信息""" cached_data = self._load_cache() if not cached_data: return {"status": "no_token", "message": "未找到token"} current_time = time.time() fetch_time = cached_data.get('fetch_time', 0) expiry_time = cached_data.get('expiry_time', 0) is_valid = self._is_token_valid(cached_data) time_remaining = max(0, expiry_time - current_time) return { "status": "valid" if is_valid else "expired", "access_token": cached_data.get('access_token', ''), "fetch_time": time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(fetch_time)), "expiry_time": time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(expiry_time)), "time_remaining_minutes": round(time_remaining / 60, 2), "is_valid": is_valid } def _fetch_new_ticket(self, access_token: str) -> Dict: """从企业微信API获取新的jsapi_ticket""" try: api_url = f"https://qyapi.weixin.qq.com/cgi-bin/ticket/get?access_token={access_token}&type=agent_config" response = httpx.get(api_url, timeout=10.0) response.raise_for_status() result = response.json() if result.get('errcode') != 0: raise Exception(f"获取ticket API错误: {result.get('errmsg', '未知错误')}") ticket = result.get('ticket') expires_in = result.get('expires_in', TICKET_EXPIRY_SECONDS) ticket_data = { 'ticket': ticket, 'fetch_time': time.time(), 'expires_in': expires_in, 'expiry_time': time.time() + expires_in } return ticket_data except httpx.RequestError as e: raise Exception(f"请求ticket失败: {e}") except httpx.HTTPStatusError as e: raise Exception(f"HTTP错误: {e}") def get_jsapi_ticket(self) -> str: """获取有效的jsapi_ticket""" # 先获取access_token access_token = self.get_access_token() # 尝试从缓存加载 cached_data = self._load_cache() # 检查ticket缓存是否有效 if cached_data and 'ticket_data' in cached_data: ticket_data = cached_data['ticket_data'] if self._is_ticket_valid(ticket_data): print("使用缓存的ticket") return ticket_data['ticket'] # 获取新ticket print("获取新的ticket...") new_ticket_data = self._fetch_new_ticket(access_token) # 更新缓存 if cached_data is None: cached_data = {} cached_data['ticket_data'] = new_ticket_data self._save_cache(cached_data) # 计算并显示过期时间 expiry_time = new_ticket_data['expiry_time'] expiry_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(expiry_time)) print(f"Ticket获取成功，过期时间: {expiry_str}") return new_ticket_data['ticket'] def get_ticket_info(self) -> Dict: """获取ticket详细信息""" cached_data = self._load_cache() if not cached_data or 'ticket_data' not in cached_data: return {"status": "no_ticket", "message": "未找到ticket"} ticket_data = cached_data['ticket_data'] current_time = time.time() fetch_time = ticket_data.get('fetch_time', 0) expiry_time = ticket_data.get('expiry_time', 0) is_valid = self._is_ticket_valid(ticket_data) time_remaining = max(0, expiry_time - current_time) return { "status": "valid" if is_valid else "expired", "ticket": ticket_data.get('ticket', ''), "fetch_time": time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(fetch_time)), "expiry_time": time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(expiry_time)), "time_remaining_seconds": round(time_remaining, 2), "is_valid": is_valid } def generate_signature(self, url: str, timestamp: int = None, nonce_str: str = None) -> Dict: """生成应用签名 根据企业微信官方文档生成签名： 1. 获取jsapi_ticket 2. 生成随机字符串noncestr 3. 生成时间戳timestamp 4. 拼接字符串：jsapi_ticket、noncestr、timestamp、url 5. 按字典序排序后拼接 6. SHA1加密生成签名 Args: url: 当前页面的完整URL timestamp: 时间戳（秒），如果不提供则自动生成 nonce_str: 随机字符串，如果不提供则自动生成 Returns: 包含签名信息的字典 """ # 获取jsapi_ticket jsapi_ticket = self.get_jsapi_ticket() # 生成时间戳（秒） if timestamp is None: timestamp = int(time.time()) # 生成随机字符串 if nonce_str is None: nonce_str = ''.join(random.choices(string.ascii_letters + string.digits, k=16)) # 按字典序排序参数 params = { 'jsapi_ticket': jsapi_ticket, 'noncestr': nonce_str, 'timestamp': timestamp, 'url': url } # 按key排序后拼接字符串 param_str = '&'.join([f"{k}={params[k]}" for k in sorted(params.keys())]) # SHA1加密 signature = hashlib.sha1(param_str.encode('utf-8')).hexdigest() # 打印签名相关信息以便校验 print("\n" + "="*60) print("签名生成信息:") print("="*60) print(f"jsapi_ticket: {jsapi_ticket}") print(f"noncestr: {nonce_str}") print(f"timestamp: {timestamp}") print(f"url: {url}") print(f"param_str (排序后): {param_str}") print(f"signature (SHA1): {signature}") print("="*60 + "\n") return { 'signature': signature, 'timestamp': timestamp, 'nonceStr': nonce_str, 'url': url, 'jsapi_ticket': jsapi_ticket } # 创建全局实例 auth_service = WeChatAuthService() def get_access_token() -> str: """获取access_token的便捷函数""" return auth_service.get_access_token() def get_token_info() -> Dict: """获取token信息的便捷函数""" return auth_service.get_token_info() def get_jsapi_ticket() -> str: """获取jsapi_ticket的便捷函数""" return auth_service.get_jsapi_ticket() def get_ticket_info() -> Dict: """获取ticket信息的便捷函数""" return auth_service.get_ticket_info() def generate_signature(url: str, timestamp: int = None, nonce_str: str = None) -> Dict: """生成应用签名的便捷函数""" return auth_service.generate_signature(url, timestamp, nonce_str)