Youtube User / Channel Scraper

1"""
2This module serves as the entry point for executing the Apify Actor. It handles the configuration of logging
3settings. The `main()` coroutine is then executed using `asyncio.run()`.
4
5Feel free to modify this file to suit your specific needs.
6"""
7
8import asyncio
9import logging
10
11from apify.log import ActorLogFormatter
12
13from .main import main
14
15# Configure loggers
16handler = logging.StreamHandler()
17handler.setFormatter(ActorLogFormatter())
18
19apify_client_logger = logging.getLogger('apify_client')
20apify_client_logger.setLevel(logging.INFO)
21apify_client_logger.addHandler(handler)
22
23apify_logger = logging.getLogger('apify')
24apify_logger.setLevel(logging.DEBUG)
25apify_logger.addHandler(handler)
26
27# Execute the Actor main coroutine
28asyncio.run(main())

1"""
2This module defines the `main()` coroutine for the Apify Actor, executed from the `__main__.py` file.
3
4Feel free to modify this file to suit your specific needs.
5
6To build Apify Actors, utilize the Apify SDK toolkit, read more at the official documentation:
7https://docs.apify.com/sdk/python
8"""
9
10# Beautiful Soup - library for pulling data out of HTML and XML files, read more at
11# https://www.crummy.com/software/BeautifulSoup/bs4/doc
12
13# HTTPX - library for making asynchronous HTTP requests in Python, read more at https://www.python-httpx.org/
14import json
15import re
16
17import requests
18from lxml import etree
19
20# Apify SDK - toolkit for building Apify Actors, read more at https://docs.apify.com/sdk/python
21from apify import Actor
22from apify.storages import KeyValueStore
23
24
25def get_count(text):
26    """
27     提取数量
28    :param text: 提取对象字符串
29    :return: 数量
30    """
31    if text:
32        count = text.replace(",", "").split(" ")[0]
33        if "K" in count:
34            count = float(count.split("K")[0]) * 1000
35        elif "M" in count:
36            count = float(count.split("M")[0]) * 1000000
37        return int(count)
38
39
40def get_link_dict(links):
41    """
42     提取链接详情
43    :param links: 提取对象列表
44    :return: 提取后的数据列表
45    """
46    link_list = []
47    for link in links:
48        item = dict()
49        item["title"] = link["channelExternalLinkViewModel"]["title"]["content"]
50        item["link"] = link["channelExternalLinkViewModel"]["link"]["content"]
51        link_list.append(item)
52    return link_list
53
54
55async def main() -> None:
56    """
57    The main coroutine is being executed using `asyncio.run()`, so do not attempt to make a normal function
58    out of it, it will not work. Asynchronous execution is required for communication with Apify platform,
59    and it also enhances performance in the field of web scraping significantly.
60    """
61    async with Actor:
62        # Structure of input is defined in input_schema.json
63        actor_input = await Actor.get_input() or {}
64        start_urls = actor_input.get('start_urls', [])
65        ids = actor_input.get('ids')
66        Actor.log.info(f'ids: {ids}')
67        if ids:
68            id_urls = [{"url":"https://www.youtube.com/channel/" + sid} if "@" not in sid else {"url":"https://www.youtube.com/" + sid} for sid in ids.split(',')]
69        else:
70            id_urls = []
71        start_urls.extend(id_urls)
72        headings = []
73        # Create an asynchronous HTTPX client
74        ua = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36"
75        headers = {
76            "authority": "www.youtube.com",
77            'accept': "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,"
78                      "*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
79            "accept-language": "en-US,en;q=0.5",
80            "cache-control": "no-cache",
81            "pragma": "no-cache",
82            "user-agent": ua
83        }
84        for url in start_urls:
85            Actor.log.info(f'crawling: {url}')
86            response = requests.get(url.get("url"), headers=headers)
87            if "This account has been terminated" in response.text:
88                print(f"该账号已经被封禁：{response.url}")
89            else:
90                html = etree.HTML(response.text)
91                name = html.xpath("//title/text()")[0].split(" -")[0]
92                verified = 1 if 'CHECK_CIRCLE_THICK' in response.text else 0
93                avatar = re.search('\"avatar\":{\"thumbnails\":\[{\"url\":\"(.*?)\"', response.text).group(1)
94                kol_token = re.findall('\"token\":\"(.*?)\"', response.text)[-1]
95                payload = json.dumps({"context": {
96                    "client": {"gl": "US", "deviceMake": "Apple", "deviceModel": "",
97                               "userAgent": ua,
98                               "clientName": "WEB", "clientVersion": "2.20240224.11.00", "osName": "Macintosh",
99                               }, "user": {"lockedSafetyMode": False},
100                    "request": {"useSsl": True, "internalExperimentFlags": [], "consistencyTokenJars": []},
101                }, "continuation": kol_token})
102                url = "https://www.youtube.com/youtubei/v1/browse?prettyPrint=false"
103                response = requests.request("POST", url, headers=headers, data=payload)
104                info_dict = response.json()["onResponseReceivedEndpoints"][0]["appendContinuationItemsAction"][
105                    "continuationItems"][0]["aboutChannelRenderer"]["metadata"]["aboutChannelViewModel"]
106                # Extract all headings from the page (tag name and text)
107                item = {}
108                item["channelId"] = info_dict.get("channelId")  # 博主ID
109                item["avatar"] = avatar  # 用户头像
110                item["title"] = name  # 博主昵称
111                item["verified"] = verified  # 是否经过认证
112                item["hasbusinessEmail"] = 1 if info_dict.get("signInForBusinessEmail") else 0  # 是否有电子邮箱
113                item["indexUrl"] = info_dict.get("canonicalChannelUrl")  # 用户主页URL
114                item["channelUrl"] = "https://www.youtube.com/channel/" + info_dict.get("channelId")
115                item["description"] = info_dict.get("description")  # 用户简介
116                item["joinDate"] = info_dict.get("joinedDateText").get("content").split("Joined ")[-1]  # 注册时间
117                item["country"] = info_dict.get("country")  # 国家
118                links = get_link_dict(info_dict.get("links", []))
119                item["links"] = links if links else None  # 获取链接详情字典
120                item["viewCount"] = get_count(info_dict.get("viewCountText"))  # 视频播放次数
121                item["videoCount"] = get_count(info_dict.get("videoCountText"))  # 视频数量
122                item["subscriberCount"] = get_count(info_dict.get("subscriberCountText"))  # 订阅数量
123                Actor.log.info(f'Extracted heading: {item}')
124                headings.append(item)
125        # Save headings to Dataset - a table-like storage
126        await Actor.push_data(headings)

1# Feel free to add your Python dependencies below. For formatting guidelines, see:
2# https://pip.pypa.io/en/latest/reference/requirements-file-format/
3
4apify ~= 1.6.0
5requests
6lxml