89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
import httpx
|
|
from datetime import datetime, timedelta
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.models.base import async_session
|
|
from app.models.user_oauth import UserOAuth
|
|
from app.models.user_oauth_account import UserOAuthAccount
|
|
from app.models.user_oauth_app import UserOAuthApp
|
|
from app.tasks.async_runner import run_async
|
|
from app.tasks.celery_app import celery_app
|
|
from app.utils.douyinApi import DouyinApi
|
|
from app.utils.douyinRequest import DouyinRequest
|
|
from app.utils.id_gen import generate_id
|
|
|
|
|
|
async def _update_oauth_accounts(account_id: str, account_userid: str, current_user_id: str, db: async_session):
|
|
conditions = [UserOAuth.user_id == current_user_id, UserOAuth.deleted_at.is_(None)]
|
|
if account_id:
|
|
conditions.append(UserOAuth.account_id == account_id)
|
|
if account_userid:
|
|
conditions.append(UserOAuth.account_userid == account_userid)
|
|
result = await db.execute(
|
|
select(UserOAuth).where(
|
|
*conditions
|
|
)
|
|
)
|
|
oauth_records = result.scalars().all()
|
|
if not oauth_records:
|
|
return True
|
|
for oauth in oauth_records:
|
|
if oauth.port_type == 1:
|
|
return await update_juliang(oauth, db)
|
|
|
|
async def update_juliang(oauth: UserOAuth, db: async_session):
|
|
if oauth.account_role == 'AGENT':
|
|
#通过代理商获取账户列表
|
|
cursor : int = 0
|
|
count : int = 10
|
|
while True:
|
|
params = {
|
|
'advertiser_id': oauth.account_id,
|
|
'count': count,
|
|
}
|
|
if cursor:
|
|
params['cursor'] = cursor
|
|
response = await DouyinApi().get_advertiser_by_agent(oauth.id, params)
|
|
if response.get('code', 0) != 0:
|
|
#记录错误日志
|
|
break
|
|
|
|
data = response['data']['list'] or []
|
|
if not data:
|
|
break
|
|
account_source = response['data']['account_source'] or ''
|
|
|
|
account_list = []
|
|
for item in data:
|
|
account_list.append(UserOAuthAccount(
|
|
id=generate_id(),
|
|
oauth_id=oauth.id,
|
|
advertiser_id=str(item),
|
|
advertiser_name="",
|
|
advertiser_role=account_source,
|
|
))
|
|
|
|
if account_list:
|
|
db.add_all(account_list)
|
|
await db.commit()
|
|
|
|
cursor = response['data'].get('cursor_page_info', {}).get('cursor')
|
|
has_more = response['data'].get('cursor_page_info', {}).get('has_more', False)
|
|
if not has_more:
|
|
break
|
|
|
|
|
|
# if celery_app:
|
|
# @celery_app.task(name="user_oauth.update_oauth_accounts", bind=True, max_retries=3, default_retry_delay=60)
|
|
# def update_oauth_accounts(self, account_id: str, account_userid: str, current_user_id: str, db):
|
|
# return run_async(_update_oauth_accounts(account_id, account_userid, current_user_id, db))
|
|
# else:
|
|
# class _DisabledTask:
|
|
# def delay(self, *args, **kwargs):
|
|
# pass
|
|
|
|
# def apply_async(self, *args, **kwargs):
|
|
# pass
|
|
|
|
# update_oauth_accounts = _DisabledTask() |