fixed and version to 4.0.1

- open many chrome tab in a time when a risk control happened for media platform
- add time gap for wb fetching note detail
- solved wrong pb migrate setting for db

Author: bigbrother666sh

core/general_process.py

@@ -318,6 +318,7 @@ def wrapper():
                             if 'wb' not in recorder.item_source:
                                 recorder.item_source['wb'] = 0
                             recorder.item_source['wb'] += 1
+                            await asyncio.sleep(1)
                     elif source_name == KUAISHOU_PLATFORM_NAME:
                         existings[KUAISHOU_PLATFORM_NAME].update(un_related_urls)
                         for video in un_related_urls:
@@ -327,6 +328,8 @@ def wrapper():
                             if 'ks' not in recorder.item_source:
                                 recorder.item_source['ks'] = 0
                             recorder.item_source['ks'] += 1
+                            # it seems that kuaishou can afford more requests per second...
+                            # await asyncio.sleep(1)
                 except Exception as e:
                     wis_logger.error(f"Error Extracting Post List: {e}")
                     continue

core/wis/kuaishou/client.py

@@ -2,9 +2,9 @@
 import asyncio
 import json
 import traceback
-from typing import Callable, Dict, List, Optional, Union
+from typing import Callable, Dict, List, Optional, Union, cast
 from urllib.parse import urlencode
-
+import time
 import httpx
 from httpx import Response
 from tenacity import RetryError, retry, stop_after_attempt, wait_fixed
@@ -35,6 +35,8 @@ def __init__(
         self._graphql = KuaiShouGraphQL()
         self.account_with_ip_pool = account_with_ip_pool
         self.account_info: Optional[AccountWithIpModel] = None
+        self._account_update_lock = asyncio.Lock()
+        self._account_updated_at = None
 
     @property
     def headers(self):
@@ -66,20 +68,35 @@ async def update_account_info(self, force_login: bool = False):
         Returns:
 
         """
-        have_account = False
-        wis_logger.debug("try to get a new account")
-        account_info = await self.account_with_ip_pool.get_account_with_ip_info(force_login)
-        self.account_info = account_info
-        have_account = await self.pong()
-        if have_account:
+        if self._account_updated_at and time.time() - self._account_updated_at < 60:
             return
-        wis_logger.info(f"current account {account_info.account.account_name} is invalid, try to get a new one")
-        await self.mark_account_invalid(account_info)
-        account_info = await self.account_with_ip_pool.get_account_with_ip_info(force_login=True)
-        self.account_info = account_info
-        have_account = await self.pong()
-        if not have_account:
-            raise DataFetchError("cannot get any valid account, we have to quit")
+        
+        # Use async lock to ensure that only one account is being updated
+        async with self._account_update_lock:
+            if self.account_with_ip_pool.proxy_ip_pool:
+                try:
+                    await self.account_with_ip_pool.mark_ip_invalid(
+                        cast(IpInfoModel, self.account_info.ip_info)
+                    )
+                    proxy_ip_pool: ProxyIpPool = cast(
+                        ProxyIpPool, self.account_with_ip_pool.proxy_ip_pool
+                    )
+                    self.account_info.ip_info = await proxy_ip_pool.get_proxy()
+                    if await self.pong():
+                        self._account_updated_at = time.time()
+                        wis_logger.info("update account by changing ip success, will continue the job")
+                        return
+                except Exception as e:
+                    wis_logger.error(f"update IP failed, err:{e}")
+
+            # in this case, we change ip and account together to ensure success -- follow ajiang's suggestion
+            await self.mark_account_invalid(self.account_info)
+            account_info = await self.account_with_ip_pool.get_account_with_ip_info(force_login=force_login)
+            self.account_info = account_info
+            if not await self.pong():
+                raise DataFetchError("cannot get any valid account, we have to quit")
+            wis_logger.info("update account by changing ip and account success, will continue the job")
+            self._account_updated_at = time.time()
 
     async def mark_account_invalid(self, account_with_ip: AccountWithIpModel):
         """
@@ -156,48 +173,18 @@ async def get(self, uri: str, params=None, **kwargs) -> Dict:
                 method="GET", url=f"{KUAISHOU_API}{final_uri}", **kwargs
             )
         except RetryError as e:
-            # 获取原始异常
-            original_exception = e.last_attempt.exception()
-            traceback.print_exception(
-                type(original_exception),
-                original_exception,
-                original_exception.__traceback__,
-            )
-            wis_logger.error(
-                f"请求uri:{uri} 重试均失败了，尝试更换账号与IP再次发起重试"
-            )
+            wis_logger.info(
+                f"[KuaiShouApiClient.get] get uri:{uri} failed many times, will try to update account")
+            # let the bull fly for a while, in case the account is updating by other thread
+            await asyncio.sleep(1)
+            await self.update_account_info(force_login=True)
             try:
-                wis_logger.info(
-                    f"请求uri:{uri} 尝试更换IP再次发起重试..."
-                )
-                await self.account_with_ip_pool.mark_ip_invalid(
-                    self.account_info.ip_info
-                )
-                self.account_info.ip_info = (
-                    await self.account_with_ip_pool.proxy_ip_pool.get_proxy()
-                )
+                wis_logger.info("try to update account by changing ip")
                 return await self.request(
                     method="GET", url=f"{KUAISHOU_API}{final_uri}", **kwargs
                 )
             except Exception as e:
-                # 获取原始异常
-                """
-                # in version 4.0, we do not have ip proxy yet
-                original_exception = ee.last_attempt.exception()
-                traceback.print_exception(
-                    type(original_exception),
-                    original_exception,
-                    original_exception.__traceback__,
-                )
-                """
-                wis_logger.error(
-                    f"请求uri:{uri}，IP更换后还是失败，尝试更换账号与IP再次发起重试"
-                )
-                await self.mark_account_invalid(self.account_info)
-                await self.update_account_info(force_login=True)
-                return await self.request(
-                    method="GET", url=f"{KUAISHOU_API}{final_uri}", **kwargs
-                )
+                wis_logger.error(f"account and ip still invalid, have to quit, err:{e}")
 
     async def post(self, uri: str, data: dict, **kwargs) -> Dict:
         """
@@ -219,53 +206,21 @@ async def post(self, uri: str, data: dict, **kwargs) -> Dict:
                 headers=self.headers,
             )
         except RetryError as e:
-            # 获取原始异常
-            original_exception = e.last_attempt.exception()
-            traceback.print_exception(
-                type(original_exception),
-                original_exception,
-                original_exception.__traceback__,
-            )
-
-            wis_logger.error(
-                f"请求uri:{uri} 重试均失败了，尝试更换账号与IP再次发起重试"
-            )
+            wis_logger.info(
+                f"[KuaiShouApiClient.post] post uri:{uri} failed many times, will try to update account")
+            # let the bull fly for a while, in case the account is updating by other thread
+            await asyncio.sleep(1)
+            await self.update_account_info(force_login=True)
             try:
-                wis_logger.info(
-                    f"请求uri:{uri} 尝试更换IP再次发起重试..."
-                )
-                await self.account_with_ip_pool.mark_ip_invalid(
-                    self.account_info.ip_info
-                )
-                self.account_info.ip_info = (
-                    await self.account_with_ip_pool.proxy_ip_pool.get_proxy()
-                )
-
+                wis_logger.info("try to update account by changing ip")
                 return await self.request(
                     method="POST",
                     url=f"{KUAISHOU_API}{uri}",
                     data=json_str,
-                    **kwargs,
-                )
-            except Exception as ee:
-                # 获取原始异常
-                """
-                # in version 4.0, we do not have ip proxy yet
-                original_exception = ee.last_attempt.exception()
-                traceback.print_exception(
-                    type(original_exception),
-                    original_exception,
-                    original_exception.__traceback__,
-                )
-                """
-                wis_logger.error(
-                    "no IP proxy available, try to get a new account"
-                )
-                await self.mark_account_invalid(self.account_info)
-                await self.update_account_info(force_login=True)
-                return await self.request(
-                    method="POST", url=f"{KUAISHOU_API}{uri}", data=json_str, **kwargs
+                    headers=self.headers,
                 )
+            except Exception as e:
+                wis_logger.error(f"account and ip still invalid, have to quit, err:{e}")
 
     async def pong(self) -> bool:
         """

core/wis/kuaishou/core.py

@@ -45,7 +45,7 @@ async def async_initialize(self):
         await account_with_ip_pool.async_initialize()
 
         self.ks_client.account_with_ip_pool = account_with_ip_pool
-        await self.ks_client.update_account_info()
+        self.ks_client.account_info = await account_with_ip_pool.get_account_with_ip_info()
 
     async def posts_list(self,
                          keywords: List[str],

core/wis/llmuse.py

@@ -38,8 +38,11 @@ def perform_completion_with_backoff(messages: List, model: str = '', **kwargs):
             if hasattr(e, 'status_code'):
                 if e.status_code in [400, 401, 413]:
                     # client error, no need to retry
-                    error_msg = f"{model} Client error: {e.status_code}. Detail: {str(e)}"
-                    if 'Image url should be a valid url or should like data:image/TYPE;base64' not in str(e):
+                    error_msg = f"{model} API error: {e.status_code}. Detail: {str(e)}"
+                    if (
+                        'Image url should be a valid url or should like data:image/TYPE;base64' not in error_msg and
+                        'Failed to process image URL: data:' not in error_msg
+                        ):
                         # image url probility is that server cannot fetch the image, so we don't need to worry about it
                         wis_logger.error(error_msg)
                         wis_logger.info(f"messages: {messages}")

core/wis/mc_commen/account_pool.py

@@ -52,7 +52,7 @@ async def get_active_account(self, force_login: bool = False) -> AccountInfoMode
             account = self._account_list.pop(0)
             if account.status.value == AccountStatusEnum.NORMAL.value:
                 wis_logger.debug(
-                    f"from saved, get active account {account}"
+                    f"from account pool get active account {account}"
                 )
                 return account
 

core/wis/nodriver_helper.py

@@ -46,7 +46,7 @@ async def start(self):
             'headless': False,
             'browser_args': [
                 '--lang=zh-CN',
-                '--no-sandbox',
+                # '--no-sandbox',
                 '--disable-translate',  # 禁用翻译
                 '--no-first-run',  # 禁用首次运行向导
                 '--no-default-browser-check'

core/wis/weibo/client.py

@@ -13,10 +13,12 @@
 from .exception import DataFetchError
 from .field import SearchType
 from urllib.parse import parse_qs, unquote, urlencode
+import time
 
 
 class WeiboClient:
     account_info: AccountWithIpModel
+    _account_update_lock: asyncio.Lock
 
     def __init__(
         self,
@@ -34,6 +36,8 @@ def __init__(
         self.timeout = timeout
         # self._user_agent = user_agent or utils.get_user_agent()
         self.account_with_ip_pool = account_with_ip_pool
+        self._account_update_lock = asyncio.Lock()
+        self._account_updated_at = None
 
     @property
     def headers(self):
@@ -61,24 +65,40 @@ def _cookies(self):
 
     async def update_account_info(self, force_login: bool = False):
         """
-        更新客户端的账号信息, 该方法会一直尝试获取新的账号信息，直到获取到一个有效的账号信息
+        Args:
+            force_login: 是否强制登录获取新账号
         Returns:
-
         """
-        have_account = False
-        wis_logger.debug("try to get a new account")
-        account_info = await self.account_with_ip_pool.get_account_with_ip_info(force_login)
-        self.account_info = account_info
-        have_account = await self.pong()
-        if have_account:
+        if self._account_updated_at and time.time() - self._account_updated_at < 60:
             return
-        wis_logger.info(f"current account {account_info.account.account_name} is invalid, try to get a new one")
-        await self.mark_account_invalid(account_info)
-        account_info = await self.account_with_ip_pool.get_account_with_ip_info(force_login=True)
-        self.account_info = account_info
-        have_account = await self.pong()
-        if not have_account:
-            raise DataFetchError("cannot get any valid account, we have to quit")
+        
+        # Use async lock to ensure that only one account is being updated
+        async with self._account_update_lock:
+            if self.account_with_ip_pool.proxy_ip_pool:
+                try:
+                    await self.account_with_ip_pool.mark_ip_invalid(
+                        cast(IpInfoModel, self.account_info.ip_info)
+                    )
+                    proxy_ip_pool: ProxyIpPool = cast(
+                        ProxyIpPool, self.account_with_ip_pool.proxy_ip_pool
+                    )
+                    self.account_info.ip_info = await proxy_ip_pool.get_proxy()
+                    if await self.pong():
+                        wis_logger.info("update account by changing ip success, will continue the job")
+                        self._account_updated_at = time.time()
+                        return
+                except Exception as e:
+                    wis_logger.error(f"update IP failed, err:{e}")
+
+            # in this case, we change ip and account together to ensure success -- follow ajiang's suggestion
+            await self.mark_account_invalid(self.account_info)
+            account_info = await self.account_with_ip_pool.get_account_with_ip_info(force_login=force_login)
+            self.account_info = account_info
+            if not await self.pong():
+                raise DataFetchError("cannot get any valid account, we have to quit")
+
+            wis_logger.info("update account by changing ip and account success, will continue the job")
+            self._account_updated_at = time.time()
 
     async def mark_account_invalid(self, account_with_ip: AccountWithIpModel):
         """
@@ -172,28 +192,17 @@ async def get(self, uri: str, params=None, **kwargs) -> Union[Response, Dict]:
             )
             return res
         except RetryError:
-            wis_logger.error(
-                f"[WeiboClient.get] 请求uri:{uri} 重试均失败了，尝试更换账号与IP再次发起重试"
-            )
+            wis_logger.info(
+                f"[WeiboClient.get] get uri:{uri} failed many times, will try to update account")
+            # let the bull fly for a while, in case the account is updating by other thread
+            await asyncio.sleep(1)
+            await self.update_account_info(force_login=True)
             try:
-                wis_logger.info("尝试更换IP再次发起重试...")
-                await self.account_with_ip_pool.mark_ip_invalid(
-                    cast(IpInfoModel, self.account_info.ip_info)
-                )
-                proxy_ip_pool: ProxyIpPool = cast(
-                    ProxyIpPool, self.account_with_ip_pool.proxy_ip_pool
-                )
-                self.account_info.ip_info = await proxy_ip_pool.get_proxy()
                 return await self.request(
                     method="GET", url=f"{WEIBO_API_URL}{final_uri}", **kwargs
                 )
             except Exception as e:
-                wis_logger.error(f"尝试更换账号再次发起重试, err:{e}")
-                await self.mark_account_invalid(self.account_info)
-                await self.update_account_info(force_login=True)
-                return await self.request(
-                    method="GET", url=f"{WEIBO_API_URL}{final_uri}", **kwargs
-                )
+                wis_logger.error(f"account and ip still invalid, have to quit, err:{e}")
 
     async def post(self, uri: str, data: Dict, **kwargs) -> Union[Response, Dict]:
         """
@@ -212,30 +221,20 @@ async def post(self, uri: str, data: Dict, **kwargs) -> Union[Response, Dict]:
             )
             return res
         except RetryError:
-            wis_logger.error(
-                f"[WeiboClient.post] 请求uri:{uri} 重试均失败了，尝试更换IP再次发起重试"
+            wis_logger.info(
+                f"[WeiboClient.post] post uri:{uri} failed many times, will try to update account"
             )
+            await asyncio.sleep(1)
+            await self.update_account_info(force_login=True)
             try:
-                await self.account_with_ip_pool.mark_ip_invalid(
-                    cast(IpInfoModel, self.account_info.ip_info)
-                )
-                proxy_ip_pool: ProxyIpPool = cast(
-                    ProxyIpPool, self.account_with_ip_pool.proxy_ip_pool
-                )
-                self.account_info.ip_info = await proxy_ip_pool.get_proxy()
                 return await self.request(
                     method="POST",
                     url=f"{WEIBO_API_URL}{uri}",
                     data=json_str,
                     **kwargs,
                 )
             except Exception as e:
-                wis_logger.error(f"尝试更换账号再次发起重试, err:{e}")
-                await self.mark_account_invalid(self.account_info)
-                await self.update_account_info(force_login=True)
-                return await self.request(
-                    method="POST", url=f"{WEIBO_API_URL}{uri}", data=json_str, **kwargs
-                )
+                wis_logger.error(f"account and ip still invalid, have to quit, err:{e}")
 
     async def pong(self) -> bool:
         """get a note to check if login state is ok"""