408 lines
16 KiB
Python
408 lines
16 KiB
Python
import json
|
|
import asyncio
|
|
from typing import Any, Dict, Optional, Tuple
|
|
|
|
import httpx
|
|
from sqlalchemy import select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from datetime import datetime, timedelta
|
|
|
|
# 【改动1：引入自定义北京时间工具类】
|
|
from app.utils.datetime_util import (
|
|
BEIJING_TZ,
|
|
datetime_to_db_tz_str,
|
|
db_tz_str_to_datetime
|
|
)
|
|
from app.models.user_oauth import UserOAuth
|
|
from app.models.user_oauth_app import UserOAuthApp
|
|
from app.models.base import async_session
|
|
from app.utils.redis import get_redis
|
|
from app.utils.logger import get_logger
|
|
|
|
logger = get_logger("douyin_request", "douyin_request")
|
|
|
|
|
|
class DouyinRequest:
|
|
def __init__(self, platform: str = "douyin"):
|
|
self._client: Optional[httpx.AsyncClient] = None
|
|
self._platform = platform
|
|
|
|
@property
|
|
def client(self) -> httpx.AsyncClient:
|
|
if not self._client:
|
|
self._client = httpx.AsyncClient(
|
|
timeout=httpx.Timeout(30.0),
|
|
follow_redirects=True,
|
|
)
|
|
return self._client
|
|
|
|
@property
|
|
def _redis_key(self) -> str:
|
|
return f"{self._platform}:tokens"
|
|
|
|
async def close(self):
|
|
if self._client:
|
|
await self._client.aclose()
|
|
self._client = None
|
|
|
|
async def _get_redis_token(self, oauth_id: str) -> Optional[str]:
|
|
"""
|
|
【核心改动：北京时间解析+脏缓存自动清理】
|
|
1. 缓存格式异常/字段缺失：直接删除脏缓存
|
|
2. Token过期则不返回，交由上层判断refresh是否过期决定是否删除缓存
|
|
"""
|
|
redis = get_redis()
|
|
if not redis:
|
|
raise ValueError("Redis连接未配置")
|
|
|
|
try:
|
|
cache_str = await redis.hget(self._redis_key, oauth_id)
|
|
if not cache_str:
|
|
return None
|
|
|
|
cache = json.loads(cache_str)
|
|
token = cache.get("token")
|
|
expired_at_str = cache.get("expired_at")
|
|
|
|
# 脏缓存：关键字段缺失，直接删除
|
|
if not (token and expired_at_str):
|
|
await redis.hdel(self._redis_key, oauth_id)
|
|
logger.warning(f"oauth_id:{oauth_id} Redis缓存字段缺失，已清理脏数据")
|
|
return None
|
|
|
|
# 【改动3：使用工具类解析带时区时间，不再强制覆盖UTC】
|
|
expired_at = db_tz_str_to_datetime(expired_at_str)
|
|
now_beijing = datetime.now(tz=BEIJING_TZ)
|
|
|
|
if expired_at > now_beijing:
|
|
return token
|
|
|
|
# AccessToken已过期，返回None，上层会校验refresh_token状态决定是否清理缓存
|
|
return None
|
|
|
|
except json.JSONDecodeError:
|
|
# JSON格式损坏，清理脏缓存
|
|
await redis.hdel(self._redis_key, oauth_id)
|
|
logger.error(f"oauth_id:{oauth_id} Redis缓存JSON格式异常，已清理脏数据")
|
|
return None
|
|
except Exception as e:
|
|
raise ValueError(f"获取Redis缓存失败: {str(e)}")
|
|
|
|
async def _set_redis_token(self, oauth_id: str, token: str, expired_at: datetime):
|
|
redis = get_redis()
|
|
if not redis:
|
|
raise ValueError("Redis连接未配置")
|
|
# 【改动4：统一转为北京时间序列化存入Redis，和数据库时区对齐】
|
|
cache = {
|
|
"token": token,
|
|
"expired_at": datetime_to_db_tz_str(expired_at),
|
|
}
|
|
try:
|
|
await redis.hset(self._redis_key, oauth_id, json.dumps(cache, ensure_ascii=False))
|
|
except Exception as e:
|
|
raise ValueError(f"设置Redis缓存失败: {str(e)}")
|
|
|
|
async def _delete_redis_token(self, oauth_id: str):
|
|
"""删除单个oauth_id缓存（授权彻底失效时调用）"""
|
|
redis = get_redis()
|
|
if not redis:
|
|
raise ValueError("Redis连接未配置")
|
|
try:
|
|
await redis.hdel(self._redis_key, oauth_id)
|
|
logger.info(f"oauth_id:{oauth_id} 授权失效，已清理Redis缓存脏数据")
|
|
except Exception as e:
|
|
raise ValueError(f"删除Redis缓存失败: {str(e)}")
|
|
|
|
async def get_access_token(self, oauth_id: str, force_refresh: bool = False) -> str:
|
|
# 优先读取缓存
|
|
if not force_refresh:
|
|
token = await self._get_redis_token(oauth_id)
|
|
if token:
|
|
return token
|
|
|
|
async with async_session() as db:
|
|
oauth_data = await db.execute(
|
|
select(UserOAuth).where(
|
|
UserOAuth.id == oauth_id,
|
|
UserOAuth.deleted_at.is_(None),
|
|
).limit(1)
|
|
)
|
|
oauth_data = oauth_data.scalar_one_or_none()
|
|
|
|
if not oauth_data:
|
|
raise ValueError("无效的oauth_id")
|
|
|
|
# 【改动5：统一北京时间当前时间】
|
|
now_beijing = datetime.now(tz=BEIJING_TZ)
|
|
|
|
if force_refresh:
|
|
# 强制刷新：先检查refresh_token是否为空
|
|
if not oauth_data.refresh_token:
|
|
# RefreshToken为空 → 授权已失效，清理Redis脏缓存
|
|
await self._delete_redis_token(oauth_id)
|
|
raise ValueError("授权已过期，请重新授权登录")
|
|
|
|
# 清空同条件下所有账号缓存与数据库token
|
|
where_cond = UserOAuth.deleted_at.is_(None)
|
|
if oauth_data.appid:
|
|
where_cond = where_cond & (UserOAuth.appid == oauth_data.appid)
|
|
if oauth_data.account_username:
|
|
where_cond = where_cond & (UserOAuth.account_username == oauth_data.account_username)
|
|
if oauth_data.account_userid:
|
|
where_cond = where_cond & (UserOAuth.account_userid == oauth_data.account_userid)
|
|
|
|
await db.execute(
|
|
update(UserOAuth).where(where_cond).values(
|
|
access_token=None,
|
|
access_token_expired=None,
|
|
)
|
|
)
|
|
await db.commit()
|
|
|
|
related_oauth_ids = await db.execute(select(UserOAuth.id).where(where_cond))
|
|
related_oauth_ids = [row[0] for row in related_oauth_ids.all()]
|
|
for related_id in related_oauth_ids:
|
|
await self._delete_redis_token(related_id)
|
|
|
|
new_token, new_expired_at = await self.refresh_access_token(db, oauth_id, oauth_data.appid, oauth_data.refresh_token)
|
|
return new_token
|
|
|
|
# 数据库AccessToken未过期，写入缓存直接返回
|
|
if oauth_data.access_token_expired and oauth_data.access_token_expired > now_beijing:
|
|
token = oauth_data.access_token
|
|
expired_at = oauth_data.access_token_expired
|
|
await self._set_redis_token(oauth_id, token, expired_at)
|
|
return token
|
|
|
|
# 查找同账号下未过期有效token复用
|
|
where_cond = UserOAuth.deleted_at.is_(None)
|
|
if oauth_data.appid:
|
|
where_cond = where_cond & (UserOAuth.appid == oauth_data.appid)
|
|
if oauth_data.account_username:
|
|
where_cond = where_cond & (UserOAuth.account_username == oauth_data.account_username)
|
|
if oauth_data.account_userid:
|
|
where_cond = where_cond & (UserOAuth.account_userid == oauth_data.account_userid)
|
|
|
|
related_oauths = await db.execute(
|
|
select(UserOAuth.access_token, UserOAuth.access_token_expired).where(
|
|
where_cond,
|
|
UserOAuth.access_token_expired.is_not(None),
|
|
UserOAuth.access_token_expired > now_beijing,
|
|
).limit(1)
|
|
)
|
|
related_oauth = related_oauths.first()
|
|
if related_oauth:
|
|
token, expired_at = related_oauth
|
|
await self._set_redis_token(oauth_id, token, expired_at)
|
|
return token
|
|
|
|
# =========【核心业务规则实现：判断RefreshToken是否过期】=========
|
|
if not oauth_data.refresh_token:
|
|
# RefreshToken为空 → 授权已失效，清理Redis脏缓存
|
|
await self._delete_redis_token(oauth_id)
|
|
raise ValueError("授权已过期，请重新授权登录")
|
|
|
|
if oauth_data.refresh_token_expired and oauth_data.refresh_token_expired < now_beijing:
|
|
# RefreshToken过期 → 授权彻底失效，清理Redis脏缓存
|
|
await self._delete_redis_token(oauth_id)
|
|
raise ValueError("授权已过期，请重新授权登录")
|
|
|
|
# RefreshToken有效，执行刷新并更新缓存
|
|
new_token, new_expired_at = await self.refresh_access_token(db, oauth_id, oauth_data.appid, oauth_data.refresh_token)
|
|
return new_token
|
|
|
|
async def refresh_access_token(self, db: AsyncSession, oauth_id: str, appid: str, refresh_token: str) -> Tuple[str, datetime]:
|
|
oauth_info = await db.execute(
|
|
select(UserOAuth.account_username, UserOAuth.account_userid).where(
|
|
UserOAuth.id == oauth_id,
|
|
UserOAuth.deleted_at.is_(None),
|
|
).limit(1)
|
|
)
|
|
oauth_info = oauth_info.first()
|
|
if not oauth_info:
|
|
raise ValueError("无效的oauth_id")
|
|
|
|
account_username, account_userid = oauth_info
|
|
|
|
result = await db.execute(
|
|
select(UserOAuthApp.secret).where(
|
|
UserOAuthApp.app_id == appid,
|
|
UserOAuthApp.status == 1,
|
|
UserOAuthApp.deleted_at.is_(None),
|
|
).limit(1)
|
|
)
|
|
app_secret = result.scalar_one_or_none()
|
|
|
|
if not app_secret:
|
|
raise ValueError("应用已被删除或禁用")
|
|
|
|
response = await self.client.request(
|
|
'POST',
|
|
'https://api.oceanengine.com/open_api/oauth2/refresh_token/',
|
|
data={
|
|
'app_id': appid,
|
|
'secret': app_secret,
|
|
'refresh_token': refresh_token,
|
|
},
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
if 'code' not in data:
|
|
raise ValueError("刷新access_token失败，接口未返回code")
|
|
|
|
code = data.get('code', 0)
|
|
if code != 0:
|
|
#刷新token
|
|
raise ValueError(f"刷新access_token失败,接口返回：{data}")
|
|
|
|
data = data.get('data', {})
|
|
new_access_token = data.get('access_token', '')
|
|
new_refresh_token = data.get('refresh_token', '')
|
|
expires_in = data.get('expires_in', 0)
|
|
refresh_token_expires_in = data.get('refresh_token_expires_in', 0)
|
|
|
|
# 【改动6：北京时间计算过期时间】
|
|
now_beijing = datetime.now(tz=BEIJING_TZ)
|
|
new_expired_at = now_beijing + timedelta(seconds=expires_in)
|
|
new_refresh_expired = now_beijing + timedelta(seconds=refresh_token_expires_in)
|
|
|
|
where_cond = UserOAuth.deleted_at.is_(None)
|
|
if appid:
|
|
where_cond = where_cond & (UserOAuth.appid == appid)
|
|
if account_username:
|
|
where_cond = where_cond & (UserOAuth.account_username == account_username)
|
|
if account_userid:
|
|
where_cond = where_cond & (UserOAuth.account_userid == account_userid)
|
|
|
|
await db.execute(
|
|
update(UserOAuth).where(where_cond).values(
|
|
access_token=new_access_token,
|
|
refresh_token=new_refresh_token,
|
|
access_token_expired=new_expired_at,
|
|
refresh_token_expired=new_refresh_expired,
|
|
)
|
|
)
|
|
await db.commit()
|
|
|
|
related_oauth_ids = await db.execute(select(UserOAuth.id).where(where_cond))
|
|
related_oauth_ids = [row[0] for row in related_oauth_ids.all()]
|
|
|
|
for related_id in related_oauth_ids:
|
|
await self._set_redis_token(related_id, new_access_token, new_expired_at)
|
|
|
|
return new_access_token, new_expired_at
|
|
|
|
async def request_with_token_with_context(
|
|
self,
|
|
oauth_id: str,
|
|
url: str,
|
|
method: str = 'GET',
|
|
options: Optional[Dict[str, Any]] = None,
|
|
request_count: int = 1,
|
|
) -> Any:
|
|
options = options or {}
|
|
token = await self.get_access_token(oauth_id)
|
|
resp_data: Optional[Dict[str, Any]] = None
|
|
|
|
for i in range(1, request_count + 1):
|
|
try:
|
|
headers = options.get('headers', {}).copy()
|
|
headers['Access-Token'] = token
|
|
|
|
has_files = 'files' in options
|
|
if has_files:
|
|
headers.pop('Content-Type', None)
|
|
else:
|
|
headers.setdefault('Content-Type', 'application/json')
|
|
|
|
options['headers'] = headers
|
|
response = await self.client.request(method, url, **options)
|
|
response.raise_for_status()
|
|
|
|
try:
|
|
resp_data = response.json()
|
|
except json.JSONDecodeError:
|
|
resp_data = {'code': 0, 'data': response.text, 'msg': 'JSON解析失败'}
|
|
|
|
if 'code' not in resp_data:
|
|
await asyncio.sleep(i * 5)
|
|
continue
|
|
|
|
code = resp_data.get('code', 0)
|
|
if code in [40102, 40104]:
|
|
await asyncio.sleep(i * 5)
|
|
token = await self.get_access_token(oauth_id, force_refresh=True)
|
|
continue
|
|
if code in [40100, 40110]:
|
|
wait_time = min(2 * (2 ** (i - 1)), 10)
|
|
await asyncio.sleep(wait_time)
|
|
continue
|
|
if code == 50000:
|
|
await asyncio.sleep(i * 10)
|
|
continue
|
|
|
|
return resp_data
|
|
|
|
except (httpx.HTTPStatusError, httpx.RequestError):
|
|
await asyncio.sleep(i * 10)
|
|
continue
|
|
|
|
# 重试耗尽，日志记录
|
|
options_log = {}
|
|
for key, value in options.items():
|
|
if key == 'files':
|
|
options_log[key] = {k: (v[0], 'bytes_content', v[2]) for k, v in value.items()}
|
|
else:
|
|
options_log[key] = value
|
|
|
|
res = json.dumps(resp_data, ensure_ascii=False) if resp_data else ''
|
|
logger.error(
|
|
f'DouYin API request failed after {request_count} retries. '
|
|
f'url:{url};method:{method};oauth_id:{oauth_id};options:{json.dumps(options_log, ensure_ascii=False)};response:{res}'
|
|
)
|
|
|
|
if resp_data and resp_data.get("code", 0) != 0:
|
|
raise ValueError(f'接口返回错误[code:{resp_data.get("code")}]{resp_data.get("message", "接口异常")}')
|
|
raise ValueError("网络请求失败，请稍后重试")
|
|
|
|
async def request_with_context(
|
|
self,
|
|
url: str,
|
|
method: str = 'GET',
|
|
options: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
options = options or {}
|
|
resp_data: Optional[Dict[str, Any]] = None
|
|
|
|
for i in range(1, 3):
|
|
try:
|
|
headers = options.get('headers', {}).copy()
|
|
headers.setdefault('Content-Type', 'application/json')
|
|
options['headers'] = headers
|
|
|
|
response = await self.client.request(method, url, **options)
|
|
response.raise_for_status()
|
|
|
|
try:
|
|
resp_data = response.json()
|
|
except json.JSONDecodeError:
|
|
resp_data = {'code': 0, 'data': response.text}
|
|
|
|
if 'code' not in resp_data:
|
|
await asyncio.sleep(i * 5)
|
|
continue
|
|
if resp_data.get("code", 0) >= 50000:
|
|
await asyncio.sleep(i * 10)
|
|
continue
|
|
return resp_data
|
|
|
|
except (httpx.HTTPStatusError, httpx.RequestError):
|
|
await asyncio.sleep(i * 10)
|
|
continue
|
|
|
|
res = json.dumps(resp_data, ensure_ascii=False) if resp_data else ''
|
|
raise RuntimeError(
|
|
f'DouYin API request failed after 2 retries. '
|
|
f'url:{url};options:{json.dumps(options, ensure_ascii=False)};response:{res}'
|
|
) |