Merge pull request #608 from Bowenwin/bili_expand

Bili_function_expand

Author: NanmiCoder

config/base_config.py

@@ -193,4 +193,17 @@
 # 是否开启按每一天进行爬取的选项，仅支持 bilibili 关键字搜索
 # 若为 False，则忽略 START_DAY 与 END_DAY 设置的值
 # 若为 True，则按照 START_DAY 至 END_DAY 按照每一天进行筛选，这样能够突破 1000 条视频的限制，最大程度爬取该关键词下的所有视频
-ALL_DAY = False
+ALL_DAY = False
+
+#!!! 下面仅支持 bilibili creator搜索
+# 爬取评论creator主页还是爬取creator动态和关系列表(True为前者)
+CREATOR_MODE = True
+
+# 爬取creator粉丝列表时起始爬取页数
+START_CONTACTS_PAGE = 1
+
+# 爬取作者粉丝和关注列表数量控制(单作者)
+CRAWLER_MAX_CONTACTS_COUNT_SINGLENOTES = 100
+
+# 爬取作者动态数量控制(单作者)
+CRAWLER_MAX_DYNAMICS_COUNT_SINGLENOTES = 50

media_platform/bilibili/client.py

@@ -21,6 +21,7 @@
 import httpx
 from playwright.async_api import BrowserContext, Page
 
+import config
 from base.base_crawler import AbstractApiClient
 from tools import utils
 
@@ -337,3 +338,162 @@ async def get_creator_videos(self, creator_id: str, pn: int, ps: int = 30, order
             "order": order_mode,
         }
         return await self.get(uri, post_data)
+
+    async def get_creator_info(self, creator_id: int) -> Dict:
+        """
+        get creator info
+        :param creator_id: 作者 ID
+        """
+        uri = "/x/space/wbi/acc/info"
+        post_data = {
+            "mid": creator_id,
+        }
+        return await self.get(uri, post_data)
+
+    async def get_creator_fans(self,
+                               creator_id: int,
+                               pn: int,
+                               ps: int = 24,
+                               ) -> Dict:
+        """
+        get creator fans
+        :param creator_id: 创作者 ID
+        :param pn: 开始页数
+        :param ps: 每页数量
+        :return:
+        """
+        uri = "/x/relation/fans"
+        post_data = {
+            'vmid': creator_id,
+            "pn": pn,
+            "ps": ps,
+            "gaia_source": "main_web",
+
+        }
+        return await self.get(uri, post_data)
+
+    async def get_creator_followings(self,
+                                     creator_id: int,
+                                     pn: int,
+                                     ps: int = 24,
+                                     ) -> Dict:
+        """
+        get creator followings
+        :param creator_id: 创作者 ID
+        :param pn: 开始页数
+        :param ps: 每页数量
+        :return:
+        """
+        uri = "/x/relation/followings"
+        post_data = {
+            "vmid": creator_id,
+            "pn": pn,
+            "ps": ps,
+            "gaia_source": "main_web",
+        }
+        return await self.get(uri, post_data)
+
+    async def get_creator_dynamics(self, creator_id: int, offset: str = ""):
+        """
+        get creator comments
+        :param creator_id: 创作者 ID
+        :param offset: 发送请求所需参数
+        :return:
+        """
+        uri = "/x/polymer/web-dynamic/v1/feed/space"
+        post_data = {
+            "offset": offset,
+            "host_mid": creator_id,
+            "platform": "web",
+        }
+
+        return await self.get(uri, post_data)
+
+    async def get_creator_all_fans(self, creator_info: Dict, crawl_interval: float = 1.0,
+                                   callback: Optional[Callable] = None,
+                                   max_count: int = 100) -> List:
+        """
+        get creator all fans
+        :param creator_info:
+        :param crawl_interval:
+        :param callback:
+        :param max_count: 一个up主爬取的最大粉丝数量
+
+        :return: up主粉丝数列表
+        """
+        creator_id = creator_info["id"]
+        result = []
+        pn = config.START_CONTACTS_PAGE
+        while len(result) < max_count:
+            fans_res: Dict = await self.get_creator_fans(creator_id, pn=pn)
+            fans_list: List[Dict] = fans_res.get("list", [])
+
+            pn += 1
+            if len(result) + len(fans_list) > max_count:
+                fans_list = fans_list[:max_count - len(result)]
+            if callback:  # 如果有回调函数，就执行回调函数
+                await callback(creator_info, fans_list)
+            await asyncio.sleep(crawl_interval)
+            if not fans_list:
+                break
+            result.extend(fans_list)
+        return result
+
+    async def get_creator_all_followings(self, creator_info: Dict, crawl_interval: float = 1.0,
+                                         callback: Optional[Callable] = None,
+                                         max_count: int = 100) -> List:
+        """
+        get creator all followings
+        :param creator_info:
+        :param crawl_interval:
+        :param callback:
+        :param max_count: 一个up主爬取的最大关注者数量
+
+        :return: up主关注者列表
+        """
+        creator_id = creator_info["id"]
+        result = []
+        pn = config.START_CONTACTS_PAGE
+        while len(result) < max_count:
+            followings_res: Dict = await self.get_creator_followings(creator_id, pn=pn)
+            followings_list: List[Dict] = followings_res.get("list", [])
+
+            pn += 1
+            if len(result) + len(followings_list) > max_count:
+                followings_list = followings_list[:max_count - len(result)]
+            if callback:  # 如果有回调函数，就执行回调函数
+                await callback(creator_info, followings_list)
+            await asyncio.sleep(crawl_interval)
+            if not followings_list:
+                break
+            result.extend(followings_list)
+        return result
+
+    async def get_creator_all_dynamics(self, creator_info: Dict, crawl_interval: float = 1.0,
+                                       callback: Optional[Callable] = None,
+                                       max_count: int = 20) -> List:
+        """
+        get creator all followings
+        :param creator_info:
+        :param crawl_interval:
+        :param callback:
+        :param max_count: 一个up主爬取的最大动态数量
+
+        :return: up主关注者列表
+        """
+        creator_id = creator_info["id"]
+        result = []
+        offset = ""
+        has_more = True
+        while has_more and len(result) < max_count:
+            dynamics_res = await self.get_creator_dynamics(creator_id, offset)
+            dynamics_list: List[Dict] = dynamics_res["items"]
+            has_more = dynamics_res["has_more"]
+            offset = dynamics_res["offset"]
+            if len(result) + len(dynamics_list) > max_count:
+                dynamics_list = dynamics_list[:max_count - len(result)]
+            if callback:
+                await callback(creator_info, dynamics_list)
+            await asyncio.sleep(crawl_interval)
+            result.extend(dynamics_list)
+        return result

media_platform/bilibili/core.py

@@ -89,8 +89,11 @@ async def start(self):
                 # Get the information and comments of the specified post
                 await self.get_specified_videos(config.BILI_SPECIFIED_ID_LIST)
             elif config.CRAWLER_TYPE == "creator":
-                for creator_id in config.BILI_CREATOR_ID_LIST:
-                    await self.get_creator_videos(int(creator_id))
+                if config.CREATOR_MODE:
+                    for creator_id in config.BILI_CREATOR_ID_LIST:
+                        await self.get_creator_videos(int(creator_id))
+                else:
+                    await self.get_all_creator_details(config.BILI_CREATOR_ID_LIST)
             else:
                 pass
             utils.logger.info(
@@ -125,7 +128,7 @@ async def get_pubtime_datetime(start: str = config.START_DAY, end: str = config.
             end_day = end_day + timedelta(days=1) - timedelta(seconds=1)  # 则将 end_day 设置为 end_day + 1 day - 1 second
         # 将其重新转换为时间戳
         return str(int(start_day.timestamp())), str(int(end_day.timestamp()))
-        
+
     async def search(self):
         """
         search bilibili video with keywords
@@ -466,3 +469,121 @@ async def get_bilibili_video(self, video_item: Dict, semaphore: asyncio.Semaphor
         extension_file_name = f"video.mp4"
         await bilibili_store.store_video(aid, content, extension_file_name)
 
+    async def get_all_creator_details(self, creator_id_list: List[int]):
+        """
+        creator_id_list: get details for creator from creator_id_list
+        """
+        utils.logger.info(
+            f"[BilibiliCrawler.get_creator_details] Crawling the detalis of creator")
+        utils.logger.info(
+            f"[BilibiliCrawler.get_creator_details] creator ids:{creator_id_list}")
+
+        semaphore = asyncio.Semaphore(config.MAX_CONCURRENCY_NUM)
+        task_list: List[Task] = []
+        try:
+            for creator_id in creator_id_list:
+                task = asyncio.create_task(self.get_creator_details(
+                    creator_id, semaphore), name=creator_id)
+                task_list.append(task)
+        except Exception as e:
+            utils.logger.warning(
+                f"[BilibiliCrawler.get_all_creator_details] error in the task list. The creator will not be included. {e}")
+
+        await asyncio.gather(*task_list)
+
+    async def get_creator_details(self, creator_id: int, semaphore: asyncio.Semaphore):
+        """
+        get details for creator id
+        :param creator_id:
+        :param semaphore:
+        :return:
+        """
+        async with semaphore:
+            creator_unhandled_info: Dict = await self.bili_client.get_creator_info(creator_id)
+            creator_info: Dict = {
+                "id": creator_id,
+                "name": creator_unhandled_info.get("name"),
+                "sign": creator_unhandled_info.get("sign"),
+                "avatar": creator_unhandled_info.get("face"),
+            }
+        await self.get_fans(creator_info, semaphore)
+        await self.get_followings(creator_info, semaphore)
+        await self.get_dynamics(creator_info, semaphore)
+
+    async def get_fans(self, creator_info: Dict, semaphore: asyncio.Semaphore):
+        """
+        get fans for creator id
+        :param creator_info:
+        :param semaphore:
+        :return:
+        """
+        creator_id = creator_info["id"]
+        async with semaphore:
+            try:
+                utils.logger.info(
+                    f"[BilibiliCrawler.get_fans] begin get creator_id: {creator_id} fans ...")
+                await self.bili_client.get_creator_all_fans(
+                    creator_info=creator_info,
+                    crawl_interval=random.random(),
+                    callback=bilibili_store.batch_update_bilibili_creator_fans,
+                    max_count=config.CRAWLER_MAX_CONTACTS_COUNT_SINGLENOTES,
+                )
+
+            except DataFetchError as ex:
+                utils.logger.error(
+                    f"[BilibiliCrawler.get_fans] get creator_id: {creator_id} fans error: {ex}")
+            except Exception as e:
+                utils.logger.error(
+                    f"[BilibiliCrawler.get_fans] may be been blocked, err:{e}")
+
+    async def get_followings(self, creator_info: Dict, semaphore: asyncio.Semaphore):
+        """
+        get followings for creator id
+        :param creator_info:
+        :param semaphore:
+        :return:
+        """
+        creator_id = creator_info["id"]
+        async with semaphore:
+            try:
+                utils.logger.info(
+                    f"[BilibiliCrawler.get_followings] begin get creator_id: {creator_id} followings ...")
+                await self.bili_client.get_creator_all_followings(
+                    creator_info=creator_info,
+                    crawl_interval=random.random(),
+                    callback=bilibili_store.batch_update_bilibili_creator_followings,
+                    max_count=config.CRAWLER_MAX_CONTACTS_COUNT_SINGLENOTES,
+                )
+
+            except DataFetchError as ex:
+                utils.logger.error(
+                    f"[BilibiliCrawler.get_followings] get creator_id: {creator_id} followings error: {ex}")
+            except Exception as e:
+                utils.logger.error(
+                    f"[BilibiliCrawler.get_followings] may be been blocked, err:{e}")
+
+    async def get_dynamics(self, creator_info: Dict, semaphore: asyncio.Semaphore):
+        """
+        get dynamics for creator id
+        :param creator_info:
+        :param semaphore:
+        :return:
+        """
+        creator_id = creator_info["id"]
+        async with semaphore:
+            try:
+                utils.logger.info(
+                    f"[BilibiliCrawler.get_dynamics] begin get creator_id: {creator_id} dynamics ...")
+                await self.bili_client.get_creator_all_dynamics(
+                    creator_info=creator_info,
+                    crawl_interval=random.random(),
+                    callback=bilibili_store.batch_update_bilibili_creator_dynamics,
+                    max_count=config.CRAWLER_MAX_DYNAMICS_COUNT_SINGLENOTES,
+                )
+
+            except DataFetchError as ex:
+                utils.logger.error(
+                    f"[BilibiliCrawler.get_dynamics] get creator_id: {creator_id} dynamics error: {ex}")
+            except Exception as e:
+                utils.logger.error(
+                    f"[BilibiliCrawler.get_dynamics] may be been blocked, err:{e}")