97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
from chuangliangTool.db_base import Chuangliang_ad
|
||
from typing import Optional, Union, Dict
|
||
|
||
|
||
class GetBaiduToken:
|
||
# 定义两种查询方式的SQL模板
|
||
TOKEN_QUERY_BY_NAME = """
|
||
SELECT advertiser_name, access_token
|
||
FROM chuangliang_ad.media_account
|
||
FORCE INDEX(idx_media_type_advertiser_status_is_active)
|
||
WHERE media_type = 'baidu'
|
||
AND is_delete = 0
|
||
AND advertiser_name = %s
|
||
AND product_version = 0
|
||
LIMIT 1
|
||
"""
|
||
|
||
TOKEN_QUERY_BY_ID = """
|
||
SELECT advertiser_name, access_token
|
||
FROM chuangliang_ad.media_account
|
||
FORCE INDEX(idx_media_type_advertiser_status_is_active)
|
||
WHERE media_type = 'baidu'
|
||
AND is_delete = 0
|
||
AND media_account_id = %s
|
||
AND product_version = 0
|
||
LIMIT 1
|
||
"""
|
||
|
||
def __init__(self, identifier: Union[str, int]):
|
||
"""
|
||
初始化方法
|
||
:param identifier: 可以是广告主名称(str)或媒体账户ID(int/str)
|
||
"""
|
||
self.identifier = identifier
|
||
self.header = None # 缓存header数据避免重复查询
|
||
|
||
def get_header(self) -> Optional[Dict[str, str]]:
|
||
"""获取包含用户名和令牌的header字典，优先返回缓存值"""
|
||
if self.header is not None:
|
||
return self.header
|
||
|
||
try:
|
||
result = self._query_token()
|
||
if result:
|
||
advertiser_name, access_token = result[0]
|
||
self.header = {
|
||
"userName": advertiser_name,
|
||
"accessToken": access_token
|
||
}
|
||
return self.header
|
||
else:
|
||
print(f"No active token found for: {self.identifier}")
|
||
return None
|
||
except Exception as e:
|
||
print(f"Error fetching token: {str(e)}")
|
||
return None
|
||
|
||
def _query_token(self) -> list:
|
||
"""根据标识符类型执行参数化SQL查询"""
|
||
# 判断标识符类型并选择相应的SQL模板
|
||
if self._is_media_account_id():
|
||
sql = self.TOKEN_QUERY_BY_ID
|
||
param = int(self.identifier) # 确保转换为整数
|
||
else:
|
||
sql = self.TOKEN_QUERY_BY_NAME
|
||
param = self.identifier
|
||
|
||
return Chuangliang_ad.query_params(sql, (param,))
|
||
|
||
def _is_media_account_id(self) -> bool:
|
||
"""判断标识符是否符合media_account_id的特征"""
|
||
# 检查是否为11位纯数字
|
||
if isinstance(self.identifier, int):
|
||
return len(str(self.identifier)) == 11
|
||
|
||
if isinstance(self.identifier, str):
|
||
return self.identifier.isdigit() and len(self.identifier) == 11
|
||
|
||
return False
|
||
|
||
|
||
# 使用示例
|
||
if __name__ == "__main__":
|
||
# 使用广告主名称查询
|
||
# token_getter1 = GetBaiduToken("原生-SLG-乱世-安卓12A20KA00006")
|
||
# header1 = token_getter1.get_header()
|
||
# print(header1)
|
||
|
||
# 使用媒体账户ID查询（字符串形式）
|
||
token_getter2 = GetBaiduToken("12466757256")
|
||
header2 = token_getter2.get_header()
|
||
print(header2)
|
||
#
|
||
# # 使用媒体账户ID查询（整数形式）
|
||
# token_getter3 = GetBaiduToken(12466757256)
|
||
# header3 = token_getter3.get_header()
|
||
# print(header3) |