mz/video-gen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
root
2026-07-06 13:39:08 +08:00
parent 16b3d223c1
commit c68215c177
33 changed files with 3250 additions and 732 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-api/app/utils/datetime_util.py
+270
View File
@@ -0,0 +1,270 @@
from datetime import datetime, timedelta, timezone, date
from typing import Optional, Union, Tuple
from dateutil import parser
# ===================== 时区与格式化常量（PHP风格映射） =====================
# 北京时间 东八区 UTC+8
BEIJING_TZ = timezone(timedelta(hours=8))
UTC_TZ = timezone.utc
# Python strftime 标准格式
FORMAT_DATETIME = "%Y-%m-%d %H:%M:%S"
FORMAT_DATE = "%Y-%m-%d"
FORMAT_TIME = "%H:%M:%S"
# PHP格式 -> Python格式 映射，贴近PHP使用习惯
PHP_FMT_MAP = {
"Y-m-d": FORMAT_DATE,
"Y-m-d H:i:s": FORMAT_DATETIME,
"H:i:s": FORMAT_TIME
}
# ===================== 私有公共工具函数（内部复用） =====================
def _normalize_date_str(date_str: str) -> str:
"""
预处理中文格式日期字符串，统一转为横杠分隔标准格式
支持：2025年01月01日 12时30分00秒 / 2025：01：01 等格式
"""
if not isinstance(date_str, str):
return ""
s = date_str.strip()
# 中文、全角符号替换清洗
s = s.replace("：", ":") \
.replace("年", "-") \
.replace("月", "-") \
.replace("日", " ") \
.replace("时", ":") \
.replace("分", ":") \
.replace("秒", "")
return s
def _get_python_fmt(fmt: str) -> str:
"""兼容PHP格式字符串，自动转为Python strftime格式"""
return PHP_FMT_MAP.get(fmt, fmt)
def _convert_ts_to_datetime(timestamp: Union[int, float], tz: timezone = BEIJING_TZ) -> datetime:
"""
统一处理 10位秒 / 13位毫秒 时间戳 -> 带时区datetime
"""
ts = float(timestamp)
# 毫秒时间戳兼容
if ts > 10 ** 12:
ts /= 1000
return datetime.fromtimestamp(ts, tz=tz)
def _parse_datetime_with_tz(dt_str: str, fmt: str, tz: timezone = BEIJING_TZ) -> Optional[datetime]:
"""
通用：固定格式日期字符串解析+绑定时区，支持中文日期预处理
仅支持指定fmt格式，不支持带时区后缀字符串
"""
try:
fmt = _get_python_fmt(fmt)
normalize_str = _normalize_date_str(dt_str)
dt = datetime.strptime(normalize_str, fmt)
return dt.replace(tzinfo=tz)
except (ValueError, TypeError):
return None
def _smart_parse_datetime(dt_str: str, base_dt: datetime, is_timezone: bool) -> Optional[datetime]:
"""
私有：dateutil智能解析日期（用于带时区、相对时间、中文复杂日期）
"""
clean_str = _normalize_date_str(dt_str)
try:
parsed_dt = parser.parse(clean_str, default=base_dt, tzinfos=None)
if is_timezone:
# 保留原始时区，无时区则兜底北京时间
if parsed_dt.tzinfo is None:
parsed_dt = parsed_dt.replace(tzinfo=BEIJING_TZ)
else:
# 普通日期强制绑定北京时间
parsed_dt = parsed_dt.replace(tzinfo=BEIJING_TZ)
return parsed_dt
except (parser.ParserError, ValueError, TypeError):
return None
# ===================== 对外工具方法（对标PHP + dateutil增强） =====================
def time() -> int:
"""
对标 PHP time()
获取当前北京时间 10位 秒级时间戳
"""
return int(datetime.now(tz=BEIJING_TZ).timestamp())
def microtime(get_as_float: bool = False) -> Union[float, str]:
"""
对标 PHP microtime()
:param get_as_float: True 返回浮点时间戳，False 返回 "微秒 秒" 字符串
"""
now = datetime.now(tz=BEIJING_TZ)
ts = now.timestamp()
if get_as_float:
return ts
sec = int(ts)
usec = int((ts - sec) * 1000000)
return f"{usec:06d} {sec}"
def strtotime(
time_str: str,
now: Optional[int] = None,
is_timezone: bool = False
) -> Optional[int]:
"""
【增强版 对标 PHP strtotime】依托 dateutil 支持相对时间、带时区时间解析
:param time_str: 待解析日期/相对时间字符串
:param now: 基准时间戳，默认当前北京时间
:param is_timezone: 是否为带时区格式的时间字符串（如 2026-07-03 10:01:29.621351+08）
- True：优先使用字符串自带时区，无时区则兜底北京时间
- False：默认按北京时间解析普通日期字符串
:return: 秒级时间戳，解析失败返回 None
"""
if not isinstance(time_str, str) or not time_str.strip():
return None
# 基准时间：默认当前北京时间
base_dt = datetime.fromtimestamp(now, tz=BEIJING_TZ) if now else datetime.now(tz=BEIJING_TZ)
parsed_dt = _smart_parse_datetime(time_str, base_dt, is_timezone)
if not parsed_dt:
return None
return int(parsed_dt.timestamp())
def date(fmt: str, timestamp: Optional[int] = None) -> str:
"""
对标 PHP date()
时间戳格式化，默认北京时间
:param fmt: 支持 Y-m-d / Y-m-d H:i:s 或 %Y-%m-%d 原生格式
:param timestamp: 秒时间戳，None则取当前时间
"""
if timestamp is None:
dt = datetime.now(tz=BEIJING_TZ)
else:
dt = _convert_ts_to_datetime(timestamp)
python_fmt = _get_python_fmt(fmt)
return dt.strftime(python_fmt)
def timestamp_to_datetime(timestamp: Union[int, float], fmt: str = "Y-m-d H:i:s") -> str:
"""
时间戳（秒/毫秒）→ 北京时间格式化字符串
:param timestamp: 10位秒 /13位毫秒
:param fmt: PHP风格格式 Y-m-d / Y-m-d H:i:s
"""
dt = _convert_ts_to_datetime(timestamp)
python_fmt = _get_python_fmt(fmt)
return dt.strftime(python_fmt)
def datetime_to_timestamp(dt_str: str, fmt: str = "Y-m-d H:i:s") -> Optional[int]:
"""
日期字符串(含中文日期) → 北京时间秒时间戳
:param dt_str: 日期字符串
:param fmt: PHP风格格式化模板
:return: 秒时间戳，解析失败返回None
"""
dt = _parse_datetime_with_tz(dt_str, fmt)
if not dt:
return None
return int(dt.timestamp())
def get_today_start_timestamp() -> int:
"""获取今日 00:00:00 北京时间 秒时间戳"""
today: date = datetime.now(tz=BEIJING_TZ).date()
start_dt = datetime.combine(today, datetime.min.time(), tzinfo=BEIJING_TZ)
return int(start_dt.timestamp())
def get_today_end_timestamp() -> int:
"""获取今日 23:59:59.999999 北京时间 秒时间戳"""
today: date = datetime.now(tz=BEIJING_TZ).date()
end_dt = datetime.combine(today, datetime.max.time(), tzinfo=BEIJING_TZ)
return int(end_dt.timestamp())
def parse_date_range(date_list: list, fmt: str = "Y-m-d", is_timezone: bool = False) -> Optional[Tuple[datetime, datetime]]:
"""
时间范围解析：["2025-01-01","2026-01-01"] → (开始0点,结束23:59:59.999999) 带北京时间
:param date_list: 长度为2的日期字符串数组
:param fmt: 日期格式，默认Y-m-d
:param is_timezone: 是否为带时区格式的时间字符串（如 2026-07-03 10:01:29.621351+08）
- True：使用dateutil智能解析，优先保留字符串自带时区，无时区兜底北京时间
- False：按指定fmt格式精准解析，强制北京时间
:return: (start_dt, end_dt) 格式非法返回None
"""
# 强参数校验
if not isinstance(date_list, list) or len(date_list) != 2:
return None
start_str, end_str = date_list[0].strip(), date_list[1].strip()
if not start_str or not end_str:
return None
base_now = datetime.now(tz=BEIJING_TZ)
if is_timezone:
# 带时区场景：智能解析，支持 2026-07-03 08:16:26.88303+08
start_dt = _smart_parse_datetime(start_str, base_now, is_timezone=True)
end_dt = _smart_parse_datetime(end_str, base_now, is_timezone=True)
else:
# 普通前端日期：固定格式解析
start_dt = _parse_datetime_with_tz(start_str, fmt, BEIJING_TZ)
end_dt = _parse_datetime_with_tz(end_str, fmt, BEIJING_TZ)
if not start_dt or not end_dt:
return None
# 结束时间补全到当日最后一毫秒
end_dt = end_dt.replace(hour=23, minute=59, second=59, microsecond=999999)
return start_dt, end_dt
def db_tz_str_to_datetime(db_datetime_str: str) -> Optional[datetime]:
"""
数据库timestamptz格式字符串（2026-07-03 08:16:26.88303+08）转为带时区datetime
"""
try:
dt = datetime.fromisoformat(db_datetime_str)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=BEIJING_TZ)
return dt
except (ValueError, TypeError):
return None
def datetime_to_db_tz_str(dt: datetime) -> str:
"""
带时区datetime转为数据库timestamptz标准字符串，统一北京时间存储
"""
if dt.tzinfo is None:
dt = dt.replace(tzinfo=BEIJING_TZ)
else:
dt = dt.astimezone(BEIJING_TZ)
return dt.isoformat()
# 导出列表
__all__ = [
"BEIJING_TZ",
"UTC_TZ",
"FORMAT_DATETIME",
"FORMAT_DATE",
"FORMAT_TIME",
"time",
"microtime",
"strtotime",
"date",
"timestamp_to_datetime",
"datetime_to_timestamp",
"get_today_start_timestamp",
"get_today_end_timestamp",
"parse_date_range",
"db_tz_str_to_datetime",
"datetime_to_db_tz_str"
]
video-gen-api/app/utils/douyinRequest.py
+123 -91
View File
@@ -5,8 +5,14 @@ from typing import Any, Dict, Optional, Tuple
import httpx
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
from datetime import datetime, timezone, timedelta
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
@@ -40,50 +46,75 @@ class DouyinRequest:
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 cache_str:
cache = json.loads(cache_str)
token = cache.get("token")
expired_at_str = cache.get("expired_at")
if token and expired_at_str:
expired_at = datetime.fromisoformat(expired_at_str).replace(tzinfo=timezone.utc)
if expired_at > datetime.now(timezone.utc):
return token
except Exception as e:
raise ValueError(f"获取Redis缓存失败: {e}")
if not cache_str:
return None
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:
cache = {
"token": token,
"expired_at": expired_at.isoformat(),
}
await redis.hset(self._redis_key, oauth_id, json.dumps(cache))
await redis.hset(self._redis_key, oauth_id, json.dumps(cache, ensure_ascii=False))
except Exception as e:
raise ValueError(f"设置Redis缓存失败: {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缓存失败: {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:
@@ -101,7 +132,17 @@ class DouyinRequest:
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)
@@ -118,9 +159,7 @@ class DouyinRequest:
)
await db.commit()
related_oauth_ids = await db.execute(
select(UserOAuth.id).where(where_cond)
)
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)
@@ -128,12 +167,14 @@ class DouyinRequest:
new_token, new_expired_at = await self.refresh_access_token(db, oauth_id, oauth_data.appid, oauth_data.refresh_token)
return new_token
if oauth_data.access_token_expired and oauth_data.access_token_expired > datetime.now(timezone.utc):
# 数据库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)
@@ -146,7 +187,7 @@ class DouyinRequest:
select(UserOAuth.access_token, UserOAuth.access_token_expired).where(
where_cond,
UserOAuth.access_token_expired.is_not(None),
UserOAuth.access_token_expired > datetime.now(timezone.utc),
UserOAuth.access_token_expired > now_beijing,
).limit(1)
)
related_oauth = related_oauths.first()
@@ -155,9 +196,18 @@ class DouyinRequest:
await self._set_redis_token(oauth_id, token, expired_at)
return token
if oauth_data.refresh_token_expired and oauth_data.refresh_token_expired < datetime.now(timezone.utc):
raise ValueError("授权已过期，请重新授权")
# =========【核心业务规则实现：判断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
@@ -171,7 +221,7 @@ class DouyinRequest:
oauth_info = oauth_info.first()
if not oauth_info:
raise ValueError("无效的oauth_id")
account_username, account_userid = oauth_info
result = await db.execute(
@@ -203,6 +253,7 @@ class DouyinRequest:
code = data.get('code', 0)
if code != 0:
#刷新token
raise ValueError(f"刷新access_token失败,接口返回：{data}")
data = data.get('data', {})
@@ -211,7 +262,10 @@ class DouyinRequest:
expires_in = data.get('expires_in', 0)
refresh_token_expires_in = data.get('refresh_token_expires_in', 0)
new_expired_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in)
# 【改动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:
@@ -222,20 +276,16 @@ class DouyinRequest:
where_cond = where_cond & (UserOAuth.account_userid == account_userid)
await db.execute(
update(UserOAuth).where(
where_cond,
).values(
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=datetime.now(timezone.utc) + timedelta(seconds=refresh_token_expires_in),
refresh_token_expired=new_refresh_expired,
)
)
await db.commit()
related_oauth_ids = await db.execute(
select(UserOAuth.id).where(where_cond)
)
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:
@@ -243,92 +293,79 @@ class DouyinRequest:
return new_access_token, new_expired_at
# 有token请求
async def request_with_token_with_context(
self,
oauth_id: str,
url: str,
method: str = 'GET',
options: any = None,
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):
# 有一些错误是触发频次管理的，需要重试
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:
# 移除可能错误设置的 Content-Type，让库自动生成 multipart 头
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:
data = response.json()
except json.JSONDecodeError:
data = {'code': 0, 'data': response.text, 'msg':'JSON解析失败'}
if 'code' not in data:
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 = data.get('code', 0)
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 data
return resp_data
except httpx.HTTPStatusError as e:
await asyncio.sleep(i * 10)
continue
except httpx.RequestError as e:
except (httpx.HTTPStatusError, httpx.RequestError):
await asyncio.sleep(i * 10)
continue
# 重试耗尽，日志记录
options_log = {}
if options:
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(data, ensure_ascii=False) if 'data' in locals() else ''
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 'data' in locals() and data.get('code', 0) != 0:
raise ValueError(f'接口返回错误[code:{data.get("code", "接口编码")}]{data.get("message", "接口返回错误")}')
else:
raise ValueError('网络错误，稍后重试。')
if resp_data and resp_data.get("code", 0) != 0:
raise ValueError(f'接口返回错误[code:{resp_data.get("code")}]{resp_data.get("message", "接口异常")}')
raise ValueError("网络请求失败，请稍后重试")
# 无token请求
async def request_with_context(
self,
url: str,
@@ -336,8 +373,9 @@ class DouyinRequest:
options: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
options = options or {}
resp_data: Optional[Dict[str, Any]] = None
for i in range(1, 2):
for i in range(1, 3):
try:
headers = options.get('headers', {}).copy()
headers.setdefault('Content-Type', 'application/json')
@@ -347,30 +385,24 @@ class DouyinRequest:
response.raise_for_status()
try:
data = response.json()
resp_data = response.json()
except json.JSONDecodeError:
data = {'code': 0, 'data': response.text}
resp_data = {'code': 0, 'data': response.text}
if 'code' not in data:
if 'code' not in resp_data:
await asyncio.sleep(i * 5)
continue
code = data.get('code', 0)
if code >= 50000:
if resp_data.get("code", 0) >= 50000:
await asyncio.sleep(i * 10)
continue
return resp_data
return data
except httpx.HTTPStatusError as e:
await asyncio.sleep(i * 10)
continue
except httpx.RequestError as e:
except (httpx.HTTPStatusError, httpx.RequestError):
await asyncio.sleep(i * 10)
continue
res = json.dumps(data) if 'data' in locals() else ''
res = json.dumps(resp_data, ensure_ascii=False) if resp_data else ''
raise RuntimeError(
f'DouYin API request failed after 5 retries. '
f'url:{url};options:{json.dumps(options)};response:{res}'
f'DouYin API request failed after 2 retries. '
f'url:{url};options:{json.dumps(options, ensure_ascii=False)};response:{res}'
)
video-gen-api/app/utils/kuaishouApi.py
+139
View File
@@ -0,0 +1,139 @@
from typing import Any, Dict, Optional
from app.utils.kuaishouRequest import KuaishouRequest
from app.models.user_oauth import UserOAuth
class KuaishouApi:
def __init__(self):
self.request = KuaishouRequest()
# 获取广告主信息
async def get_advertiser_info(self, oauth_id: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
if not oauth_id:
raise RuntimeError('OAuth ID is not set.')
url = "https://ad.e.kuaishou.com/rest/openapi/v1/advertiser/info"
return await self.request.request_with_token_with_context(
oauth_id,
url,
'GET',
{'params': params or {}}
)
# 上传图片素材
async def upload_image_material(self, oauth_id: str, data: Optional[Dict[str, Any]] = None, files: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
if not oauth_id:
raise RuntimeError('OAuth ID is not set.')
url = "https://ad.e.kuaishou.com/rest/openapi/v1/file/ad/image/upload"
options: Dict[str, Any] = {}
if data:
options['data'] = data
if files:
options['files'] = files
return await self.request.request_with_token_with_context(
oauth_id,
url,
'POST',
options,
request_count=3
)
# 上传视频素材
async def upload_video_material(self, oauth_id: str, data: Optional[Dict[str, Any]] = None, files: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
if not oauth_id:
raise RuntimeError('OAuth ID is not set.')
url = "https://ad.e.kuaishou.com/rest/openapi/v1/file/ad/video/upload"
options: Dict[str, Any] = {}
if data:
options['data'] = data
if files:
options['files'] = files
return await self.request.request_with_token_with_context(
oauth_id,
url,
'POST',
options,
request_count=3
)
# 获取地域信息
async def get_area(self, oauth_id: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
if not oauth_id:
raise RuntimeError('OAuth ID is not set.')
url = "https://ad.e.kuaishou.com/rest/openapi/v1/tools/admin/info"
return await self.request.request_with_token_with_context(
oauth_id,
url,
'GET',
{'params': params or {}}
)
# 获取素材消耗（报表查询）
async def get_material_cost(self, oauth_id: str, params: any, request_count: int = 3) -> Dict[str, Any]:
if not oauth_id:
raise RuntimeError('OAuth ID is not set.')
url = "https://ad.e.kuaishou.com/rest/openapi/v1/report/get"
return await self.request.request_with_token_with_context(
oauth_id,
url,
'GET',
{'params': params or {}},
request_count=request_count
)
# 获取账户信息
async def get_account_info(self, oauth_id: str, params: any) -> Dict[str, Any]:
if not oauth_id:
raise RuntimeError('OAuth ID is not set.')
url = "https://ad.e.kuaishou.com/rest/openapi/v1/advertiser/info"
return await self.request.request_with_token_with_context(
oauth_id,
url,
'GET',
{'params': params or {}}
)
# 获取广告计划列表
async def get_ad_list(self, oauth_id: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
if not oauth_id:
raise RuntimeError('OAuth ID is not set.')
url = "https://ad.e.kuaishou.com/rest/openapi/v1/ad/list"
return await self.request.request_with_token_with_context(
oauth_id,
url,
'GET',
{'params': params or {}}
)
# 获取广告组列表
async def get_ad_unit_list(self, oauth_id: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
if not oauth_id:
raise RuntimeError('OAuth ID is not set.')
url = "https://ad.e.kuaishou.com/rest/openapi/v2/ad_unit/list"
return await self.request.request_with_token_with_context(
oauth_id,
url,
'GET',
{'params': params or {}}
)
# 获取账户余额
async def get_account_balance(self, oauth_id: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
if not oauth_id:
raise RuntimeError('OAuth ID is not set.')
url = "https://ad.e.kuaishou.com/rest/openapi/v1/advertiser/balance"
return await self.request.request_with_token_with_context(
oauth_id,
url,
'GET',
{'params': params or {}}
)
video-gen-api/app/utils/kuaishouRequest.py
+276
View File
@@ -0,0 +1,276 @@
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, timezone, timedelta
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("kuaishou_request", "kuaishou_request")
class KuaishouRequest:
def __init__(self, platform: str = "kuaishou"):
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]:
redis = get_redis()
if not redis:
raise ValueError("Redis连接未配置")
try:
cache_str = await redis.hget(self._redis_key, oauth_id)
if cache_str:
cache = json.loads(cache_str)
token = cache.get("token")
expired_at_str = cache.get("expired_at")
if token and expired_at_str:
expired_at = datetime.fromisoformat(expired_at_str).replace(tzinfo=timezone.utc)
if expired_at > datetime.now(timezone.utc):
return token
except Exception as e:
raise ValueError(f"获取Redis缓存失败: {e}")
return None
async def _set_redis_token(self, oauth_id: str, token: str, expired_at: datetime):
redis = get_redis()
if not redis:
raise ValueError("Redis连接未配置")
try:
cache = {
"token": token,
"expired_at": expired_at.isoformat(),
}
await redis.hset(self._redis_key, oauth_id, json.dumps(cache))
except Exception as e:
raise ValueError(f"设置Redis缓存失败: {e}")
async def _delete_redis_token(self, oauth_id: str):
redis = get_redis()
if not redis:
raise ValueError("Redis连接未配置")
try:
await redis.hdel(self._redis_key, oauth_id)
except Exception as e:
raise ValueError(f"删除Redis缓存失败: {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")
if force_refresh:
await db.execute(
update(UserOAuth).where(UserOAuth.id == oauth_id).values(
access_token=None,
access_token_expired=None,
)
)
await db.commit()
await self._delete_redis_token(oauth_id)
new_token, new_expired_at = await self.refresh_access_token(db, oauth_id, oauth_data.appid, oauth_data.refresh_token)
await self._set_redis_token(oauth_id, new_token, new_expired_at)
return new_token
if oauth_data.access_token_expired and oauth_data.access_token_expired > datetime.now(timezone.utc):
token = oauth_data.access_token
expired_at = oauth_data.access_token_expired
await self._set_redis_token(oauth_id, token, expired_at)
return token
if oauth_data.refresh_token_expired and oauth_data.refresh_token_expired < datetime.now(timezone.utc):
raise ValueError("授权已过期，请重新授权")
new_token, new_expired_at = await self.refresh_access_token(db, oauth_id, oauth_data.appid, oauth_data.refresh_token)
await self._set_redis_token(oauth_id, new_token, new_expired_at)
return new_token
async def refresh_access_token(self, db: AsyncSession, oauth_id: str, appid: str, refresh_token: str) -> Tuple[str, datetime]:
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("应用已被删除或禁用")
# 快手刷新token接口
response = await self.client.request(
'POST',
'https://ad.e.kuaishou.com/rest/openapi/oauth2/authorize/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:
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('access_token_expires_in', 0)
refresh_token_expires_in = data.get('refresh_token_expires_in', 0)
new_expired_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in)
await db.execute(
update(UserOAuth).where(
UserOAuth.id == oauth_id,
).values(
access_token=new_access_token,
refresh_token=new_refresh_token,
access_token_expired=new_expired_at,
refresh_token_expired=datetime.now(timezone.utc) + timedelta(seconds=refresh_token_expires_in),
)
)
await db.commit()
return new_access_token, new_expired_at
# 有token请求
async def request_with_token_with_context(
self,
oauth_id: str,
url: str,
method: str = 'GET',
options: any = None,
request_count: int = 3,
) -> Any:
options = options or {}
token = await self.get_access_token(oauth_id)
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:
data = response.json()
except json.JSONDecodeError:
data = {'code': 0, 'data': response.text, 'msg': 'JSON解析失败'}
# 如果没有code，直接返回数据，快手触发接口频次会返回空
if 'code' not in data:
await asyncio.sleep(i * 5)
continue
code = data.get('code', 0)
# 检查是否需要刷新令牌
if code in [402000, 400003, 402007, 402005, 402004, 401000]:
# 如果message包含 '该账户不在您的代理商下'，说明账户已经转走了，不需要刷新token，直接返回
message = data.get('message', '')
if '该账户不在您的代理商下' in message:
return data
await asyncio.sleep(i * 5)
token = await self.get_access_token(oauth_id, force_refresh=True)
continue
# 检查是否触发接口频次
if code in [400001, 402007, 402008, 410000, 410001]:
await asyncio.sleep(i * 5)
continue
# 服务端错误
if code >= 500000:
await asyncio.sleep(i * 5)
continue
# 其他错误直接返回数据
return data
except httpx.HTTPStatusError as e:
await asyncio.sleep(i * 10)
continue
except httpx.RequestError as e:
await asyncio.sleep(i * 10)
continue
options_log = {}
if options:
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(data, ensure_ascii=False) if 'data' in locals() else ''
logger.error(
f'KuaiShou 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 'data' in locals() and data.get('code', 0) != 0:
raise ValueError(f'接口返回错误[code:{data.get("code", "接口编码")}]{data.get("message", "接口返回错误")}')
else:
raise ValueError('网络错误，稍后重试。')