319 lines
11 KiB
Python
319 lines
11 KiB
Python
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)
|