编辑采集器       启动采集器

属性	值	属性	值
名称	us_x	创建时间	2025-04-02 10:10:02
标题	美国-Twitter	更新时间	-
状态	stop	开始时间	2025-07-04 00:26:13
停止类型	finished	结束时间	2025-07-05 01:27:11
信号		运行时长	25.0小时
进程ID	3842	进程名称	Process-8
作业ID	57d0cfac-0658-4895-8cdf-a82da1affac3	作业目录	records/jobs/57d0cfac-0658-4895-8cdf-a82da1affac3
日志文件	records/logs/57d0cfac-0658-4895-8cdf-a82da1affac3.log     (查看日志)     (tail日志)
采集器类型	scrapy	文件	us_x_spider.py
起始urls(每行一个) (若为空则使用代码中 的start_urls)

+新增命令

启动命令	名称	类型	命令内容	状态	创建时间	启动时间	结束时间	时长	修改	删除

+新增定时任务

定时任务	类型	定时时间	更新时间	激活	修改	删除

历史作业 ID	开始时间	结束时间	运行时长	进程ID	进程名	日志
b6a783aa-0e23-4d31-95d7-ca032e95b3b5	2025-06-18 10:51:29	2025-06-21 02:21:35	63.5小时	3262243	Process-8	查看日志	删除记录
619009c7-79a8-42dc-9d42-6d15d5bfb53d	2025-06-13 10:09:08	2025-06-18 18:51:25	128.7小时	2831229	Process-5	查看日志	删除记录
删除最后 选择条数 条历史作业 确定

代码

import json import os import platform import random import re import threading import time import traceback from enum import Enum from bs4 import BeautifulSoup import dao import utils from spider.spiders.common.base_spider import BaseListToDetailsSpider from spider.spiders.common.models import File_Type, Language, Nation, Post, Post_Content_Type, User, \ PostModel, File, BaseModel from spider.spiders.common.utils.download_utils import download_image from spider.spiders.resource.douyin.utils.download_python import download from playwright.sync_api import sync_playwright, expect, Response # 创建临时目录 user_home = os.path.expanduser('~') temp_dir = os.path.join(user_home, 'temp_data_video') if not os.path.exists(temp_dir): os.makedirs(temp_dir) class CrawlerType(Enum): SEARCH = 'keywords' # 按关键词搜索 USER = 'accounts' # 以用户ID为入口 POST = 'posts' # 以推文url为入口 # us_x_spider class UsXSpider(BaseListToDetailsSpider): name = 'us_x' desc = '美国-Twitter' net_name = 'Twitter' net_url = 'www.x.com' language = Language.EN nation_category = Nation.OUT_SIDE channel_name = 'TwitterSearch' channel_url = 'https://x.com/search' channel_desc = 'Twitter>TwitterSearch' # 已处理推文 processed_post = {} # 已处理username processed_username = {} # 已处理视频链接 processed_url = {} # 推文是否有视频 post_has_video = {} # 总处理推文数 sum = 0 # 重复推文数 repeat_num = 0 # 成功处理推文数 success_num = 0 # 跳过数量 skip_num = 0 # 广告数 ad_num = 0 # 错误数（包含广告和其他错误） error_num = 0 # 图像数量 image_num = 0 # 头像数量 profile_num = 0 # 视频数量 video_num = 0 # ZIP数量 zip_num = 0 # 其他文件数量 txt_num = 0 cmd_id = 0 # 最大翻页数 max_page = 5 # 评论最大深度 max_depth = 3 timeout_count = 0 # 读取队列最大超时次数 max_timeout_count = 6 # 是否按用户ID发散 diverge_user = False cookies = [ { "domain": ".x.com", "expirationDate": 1770346464.234086, "hostOnly": False, "httpOnly": True, "name": "auth_token", "path": "/", "secure": True, "session": False, "storeId": None, "value": "c9f225135cd8c2c56c46a9f7503dd8f297f2fcb6" }, { "domain": ".x.com", "expirationDate": 1735795371.223896, "hostOnly": False, "httpOnly": False, "name": "gt", "path": "/", "secure": True, "session": False, "storeId": None, "value": "1874650032982655395" }, { "domain": ".x.com", "expirationDate": 1770346371.113944, "hostOnly": False, "httpOnly": False, "name": "guest_id", "path": "/", "secure": True, "session": False, "storeId": None, "value": "v1%3A173578637181748532" }, { "domain": ".x.com", "expirationDate": 1767322662.861446, "hostOnly": False, "httpOnly": False, "name": "twid", "path": "/", "secure": True, "session": False, "storeId": None, "value": "u%3D1739218826342899712" }, { "domain": "x.com", "hostOnly": True, "httpOnly": False, "name": "lang", "path": "/", "secure": False, "session": True, "storeId": None, "value": "en" }, { "domain": ".x.com", "expirationDate": 1767322661.4684, "hostOnly": False, "httpOnly": False, "name": "night_mode", "path": "/", "secure": True, "session": False, "storeId": None, "value": "2" }, { "domain": ".x.com", "expirationDate": 1735872864.967413, "hostOnly": False, "httpOnly": True, "name": "att", "path": "/", "secure": True, "session": False, "storeId": None, "value": "1-QW1LlKS2DDeQfqkwmbzx4bTMa0pOUBONURApK6vs" }, { "domain": ".x.com", "expirationDate": 1770346464.616771, "hostOnly": False, "httpOnly": False, "name": "ct0", "path": "/", "secure": True, "session": False, "storeId": None, "value": "211b6b251c5823365f10ca2c6ad9bb9ab2ebf8fee4f4aeecd6de28ff8168c0765fe911e2d6685b3ddb336b4367532c3ae8bd1c7d8a7fe115e773fd42b1e01dedd96b2ac57b0c3ad32635c6a992307766" }, { "domain": ".x.com", "expirationDate": 1770346370.87792, "hostOnly": False, "httpOnly": False, "name": "dnt", "path": "/", "secure": True, "session": False, "storeId": None, "value": "1" }, { "domain": ".x.com", "expirationDate": 1770346662.861367, "hostOnly": False, "httpOnly": False, "name": "guest_id_ads", "path": "/", "secure": True, "session": False, "storeId": None, "value": "v1%3A173578637181748532" }, { "domain": ".x.com", "expirationDate": 1770346662.861418, "hostOnly": False, "httpOnly": False, "name": "guest_id_marketing", "path": "/", "secure": True, "session": False, "storeId": None, "value": "v1%3A173578637181748532" }, { "domain": ".x.com", "expirationDate": 1770346464.233817, "hostOnly": False, "httpOnly": True, "name": "kdt", "path": "/", "secure": True, "session": False, "storeId": None, "value": "v4H2N173AUXTDXR7zBRmQxI7syePjxuXxDOVPDdC" }, { "domain": ".x.com", "expirationDate": 1767322466.849394, "hostOnly": False, "httpOnly": False, "name": "personalization_id", "path": "/", "secure": True, "session": False, "storeId": None, "value": "\"v1_BAwnLnRKn8OJ/pmiRwigLg==\"" } ] @staticmethod def get_user_url(user_id): return f'https://x.com/{user_id}' @staticmethod def get_search_url(keyword): return f'https://x.com/search?q={keyword}&src=recent_search_click' @staticmethod def get_default_targets(): # 未配置命令时，默认搜索关键词 return ['Philippines', 'Scarborough Shoal', 'Second Thomas Shoal', 'Thitu Island', 'Fiery Cross Reef', 'Subi Reef', 'Mischief Reef', 'Sabina Shoal','南沙群岛','Paracel Islands','Spratly Islands','西沙群岛','中沙群岛','Macclesfield Bank','黄岩岛','太平岛','Itu Aba','永暑礁', 'Fiery Cross Reef','美济礁','Mischief Reef','仁爱礁','Second Thomas Shoal','九段线','Nine-Dash Line','专属经济区','Exclusive Economic Zone','EEZ','大陆架','Continental Shelf','联合国海洋法公约','United Nations Convention on the Law of the Sea','UNCLOS','南海仲裁案','South China Sea Arbitration', '航行自由','Freedom of Navigation','军事化','Militarization','岛礁建设','Island Building','地缘政治','Geopolitics','战略航道','Strategic Waterway','海洋资源','Marine Resources','主权声索','Territorial Claim','Panatag Shoal', 'Bajo de Masinloc','BRP Sierra Madre'] def start_requests(self): cmd_info_op = CrawlerType.SEARCH.value cmd_infos = dao.get_spider_cmd_wait_infos(self.name) if not cmd_infos: utils.log_print(f"没有待处理的命令") targets = self.get_default_targets() else: # 拿取创建时间最早的命令 cmd_info = cmd_infos[0] utils.log_print(f'开始爬取, cmd_info: {cmd_info}') self.cmd_id = cmd_info.get('id') dao.update_spider_cmd_info(self.name, self.cmd_id, 'create_time', time.time()) cmd_info_op = cmd_info.get('op') cmd_info_str = cmd_info.get('cmd_str') targets = cmd_info_str.split('\r\n') detail_thread = threading.Thread(target=self.handle_details, args=()) detail_thread.start() with sync_playwright() as p: headless = False if platform.system() == 'Windows' else True browser = p.chromium.launch(headless=headless) context = browser.new_context() context.add_cookies(self.cookies) if CrawlerType.SEARCH.value == cmd_info_op: for target in targets: self.handle_search(context, target) elif CrawlerType.USER.value == cmd_info_op: for target in targets: self.handle_user(context, target) elif CrawlerType.POST.value == cmd_info_op: for target in targets: self.handle_post(context, target) else: utils.log_print(f"op参数不支持：{cmd_info_op}") detail_thread.join() utils.log_print(f'sum: {self.sum}, repeat_num: {self.repeat_num}, success_num: {self.success_num}, ad_num: {self.ad_num}, skip_num: {self.skip_num}, error_num: {self.error_num}') utils.log_print(f'zip_num: {self.zip_num}, image_num: {self.image_num}, profile_num: {self.profile_num}, video_num: {self.video_num}, txt_num: {self.txt_num}') self.update_count() dao.update_spider_cmd_info(self.name, self.cmd_id, 'stop_time', time.time()) dao.update_spider_cmd_info(self.name, self.cmd_id, 'status', 'stop') return [] def update_count(self): dao.update_spider_cmd_info(self.name, self.cmd_id, 'page_count', self.sum) dao.update_spider_cmd_info(self.name, self.cmd_id, 'txt_count', self.txt_num) dao.update_spider_cmd_info(self.name, self.cmd_id, 'zip_count', self.zip_num) dao.update_spider_cmd_info(self.name, self.cmd_id, 'img_count', self.image_num) dao.update_spider_cmd_info(self.name, self.cmd_id, 'prof_img_count', self.profile_num) dao.update_spider_cmd_info(self.name, self.cmd_id, 'video_count', self.video_num) data = f'page_count:{self.sum}, txt_count:{self.txt_num}, zip_count:{self.zip_num}, img_count:{self.image_num}, prof_img_count:{self.profile_num}, video_count:{self.video_num}' # todo utils.log_print(f'同步更新zk, data: {data}') def handle_search(self, context, target): page = context.new_page() page.goto(self.get_search_url(target)) for page_num in range(self.max_page): utils.log_print(f"开始第 {page_num} 次循环") time.sleep(10) # 等待页面加载 expect(page.locator('article').nth(0)).to_be_visible(timeout=30000) # 获取article数量 article_count = page.locator('article').count() for article_num in range(article_count): self.sum += 1 if self.sum % 10 == 0: self.update_count() utils.log_print("=" * 30) utils.log_print(f"第 {page_num} 次循环，({article_num + 1}/{article_count})正在获取推文。") try: article = page.locator("article").nth(article_num) soup = BeautifulSoup(article.inner_html(), "lxml") # 简单判断是否为广告 if 'style="text-overflow: unset;">Ad</span>' in str(soup): self.ad_num += 1 raise ValueError("这条推文是广告。") # 找出相关信息 time_element = soup.find("time") # publish_time = time_element.get("datetime") publish_url = "https://x.com" + time_element.find_parent().get("href") if self.processed_post.get(publish_url): self.repeat_num += 1 utils.log_print(f'推文已重复，processed_post: {publish_url}') else: utils.log_print(f'推文未重复，publish_url: {publish_url}') self.processed_post[publish_url] = time.time() post_dict = { "post_url": publish_url, "keyword": target, "parent_url": None, "root_url": publish_url, "depth": 1, } utils.log_print(f"推文加入队列: {post_dict}") dao.put_spider_work_queue(self.name, json.dumps(post_dict)) if article_num + 1 == article_count: utils.log_print(f"到达{article_count}, 开始滚动") page.screenshot(path=os.path.join(temp_dir, f'x_screenshot_{page_num}.png')) page.evaluate( "const list = document.querySelectorAll('article');list[list.length -1].scrollIntoView(true)") except Exception as e: self.error_num += 1 utils.log_print(f"({article_num + 1}/{article_count})推文获取出错： {e}") if article_num + 1 == article_count: utils.log_print(f"到达{article_count}, 开始滚动") page.screenshot(path=os.path.join(temp_dir, f'x_screenshot_{page_num}.png')) page.evaluate( "const list = document.querySelectorAll('article');list[list.length -1].scrollIntoView(true)") page.close() def handle_user(self, context, target): page = context.new_page() page.goto(self.get_user_url(target)) for page_num in range(self.max_page): utils.log_print(f"开始第 {page_num} 次循环") time.sleep(10) # 等待页面加载 expect(page.locator('article').nth(0)).to_be_visible(timeout=30000) # 获取article数量 article_count = page.locator('article').count() for article_num in range(article_count): self.sum += 1 if self.sum % 10 == 0: self.update_count() utils.log_print("=" * 30) utils.log_print(f"第 {page_num} 次循环，({article_num + 1}/{article_count})正在获取推文。") try: article = page.locator("article").nth(article_num) soup = BeautifulSoup(article.inner_html(), "lxml") # 简单判断是否为广告 if 'style="text-overflow: unset;">Ad</span>' in str(soup): self.ad_num += 1 raise ValueError("这条推文是广告。") # 找出相关信息 time_element = soup.find("time") # publish_time = time_element.get("datetime") publish_url = "https://x.com" + time_element.find_parent().get("href") if self.processed_post.get(publish_url): self.repeat_num += 1 utils.log_print(f'推文已重复，processed_post: {publish_url}') else: utils.log_print(f'推文未重复，publish_url: {publish_url}') self.processed_post[publish_url] = time.time() post_dict = { "post_url": publish_url, "keyword": target, "parent_url": None, "root_url": publish_url, "depth": 1, } utils.log_print(f"推文加入队列: {post_dict}") dao.put_spider_work_queue(self.name, json.dumps(post_dict)) if article_num + 1 == article_count: utils.log_print(f"到达{article_count}, 开始滚动") page.screenshot(path=os.path.join(temp_dir, f'x_screenshot_{page_num}.png')) page.evaluate( "const list = document.querySelectorAll('article');list[list.length -1].scrollIntoView(true)") except Exception as e: self.error_num += 1 utils.log_print(f"--- ({article_num + 1}/{article_count})推文获取出错： {e}") if article_num + 1 == article_count: utils.log_print(f"到达{article_count}, 开始滚动") page.screenshot(path=os.path.join(temp_dir, f'x_screenshot_{page_num}.png')) page.evaluate( "const list = document.querySelectorAll('article');list[list.length -1].scrollIntoView(true)") page.close() def handle_post(self, context, target): self.sum += 1 if self.sum % 10 == 0: self.update_count() if self.processed_post.get(target): self.repeat_num += 1 utils.log_print(f'推文已重复，processed_post: {target}') else: utils.log_print(f'推文未重复，publish_url: {target}') self.processed_post[target] = time.time() post_dict = { "post_url": target, "keyword": target, "parent_url": None, "root_url": target, "depth": 1, } utils.log_print(f"推文加入队列: {post_dict}") dao.put_spider_work_queue(self.name, json.dumps(post_dict)) def handle_details(self): with sync_playwright() as p: headless = False if platform.system() == 'Windows' else True browser = p.chromium.launch(headless=headless) context = browser.new_context() context.add_cookies(self.cookies) while True: time.sleep(random.randint(3, 5)) try: post_dict = dao.get_spider_work_queue_wait(self.name, 30) if not post_dict: self.timeout_count += 1 if self.timeout_count > self.max_timeout_count: utils.log_print(f"超过{self.max_timeout_count}次未获取到待处理推文，退出") break utils.log_print(f"未获取到待处理推文， timeout_count: {self.timeout_count}") continue self.timeout_count = 0 post_dict = json.loads(post_dict) publish_url = post_dict.get("post_url") depth = post_dict.get("depth") if depth > self.max_depth: self.skip_num += 1 utils.log_print(f"超过最大深度，跳过") continue utils.log_print(f"读取到publish_url: {publish_url}") self.parse_detail(context, post_dict) except Exception as e: utils.log_print(f"解析publish_url: {publish_url}详情出错： {e}") traceback.print_exc() def parse_detail(self, context, post_dict): detail_page = context.new_page() publish_url = post_dict.get("post_url") try: self.post_has_video[publish_url] = True detail_page.goto(publish_url) locator_article = detail_page.locator('article') expect(locator_article.nth(0)).to_be_visible(timeout=30000) # expect(detail_page.get_by_test_id('tweet')).to_be_visible(timeout=30000) detail_article = detail_page.locator('article[tabindex="-1"]').first detail_soup = BeautifulSoup(detail_article.inner_html(), "lxml") detail_time_element = detail_soup.find("time") publish_time = detail_time_element.get("datetime") detail_page.on('response', lambda response: self.process_response(response, publish_url, detail_page,publish_time)) # 要注意推文是否没有正文 tweet_text = detail_soup.find("div", attrs={"data-testid": "tweetText"}) publish_content = tweet_text.get_text() if tweet_text else "" publish_title = publish_content.split('\n')[0] if publish_content else "" # 还要注意是否没有图片 tweet_photo = detail_soup.findAll("div", attrs={"data-testid": "tweetPhoto"}) publish_images = [] for photo in tweet_photo: find_img = photo.find("img") if find_img: photo_url = find_img.get("src") if photo_url.startswith("http"): publish_images.append(photo_url) user_name__find_all = detail_soup.find("div", attrs={"data-testid": "User-Name"}).find_all('span') author = user_name__find_all[0].get_text() user_id = user_name__find_all[-1].get_text() if not user_id.startswith('@'): user_id = user_name__find_all[-2].get_text() author += user_id avatar = detail_soup.find("div", attrs={"data-testid": "Tweet-User-Avatar"}) avatar_images = [img.get("src") for img in avatar.find_all("img")] if avatar else [] avatar_image = avatar_images[0] if avatar_images else "" # 还要注意是否没有视频 tweet_video = detail_soup.find("div", attrs={"data-testid": "videoComponent"}) publish_video = [video.get("src") for video in tweet_video.find_all("video")] if tweet_video else [] if not tweet_video: self.post_has_video[publish_url] = False utils.log_print(f"---> 这条推文没有视频，publish_url: {publish_url}") reply = detail_soup.find("button", attrs={"data-testid": "reply"}).get("aria-label") reply_num = int(reply.split(' ')[0]) if reply else 0 repost = detail_soup.find("button", attrs={"data-testid": "retweet"}).get("aria-label") repost_num = int(repost.split(' ')[0]) if repost else 0 like = detail_soup.find("button", attrs={"data-testid": "like"}).get("aria-label") like_num = int(like.split(' ')[0]) if like else 0 bookmark = detail_soup.find("button", attrs={"data-testid": "bookmark"}).get("aria-label") bookmark_num = int(bookmark.split(' ')[0]) if bookmark else 0 utils.log_print( f"---> reply_num: {reply_num}, repost_num: {repost_num}, like_num: {like_num}, bookmark_num: {bookmark_num}") tweet_dict = {"publish_time": publish_time, "author": author, "publish_url": publish_url, "publish_title": publish_title, "publish_content": publish_content, "publish_images": publish_images, "publish_video": publish_video, "avatar_image": avatar_image, "reply_num": reply_num, "repost_num": repost_num, "like_num": like_num, "bookmark_num": bookmark_num} self.save_detail(tweet_dict, detail_page, post_dict) self.success_num += 1 time.sleep(5) if not tweet_video: detail_page.close() except Exception as e: self.error_num += 1 utils.log_print(f"解析publish_url:{publish_url}详情出错： {e}") traceback.print_exc() finally: detail_page.close() def process_response(self, response: Response, post_url, detail_page,publish_time): url = response.url if self.post_has_video.get(post_url) and url.find(".mp4") > 0: utils.log_print(f'---> find mp4 response url: {url}') if self.processed_url.get(url): utils.log_print(f'已重复，processed_url: {url}') return self.processed_url[url] = time.time() self.download_video(post_url, url,publish_time) # 延迟关闭页面，防止文本图片评论等信息未解析完 threading.Thread(target=self.close_page(detail_page, 30)).start() def close_page(self, page, wait_time=30): try: time.sleep(wait_time) page.close() except Exception as e: utils.log_print(f"关闭页面出错: {e}") def download_video(self, post_url, download_addr,publish_time): utils.log_print(f"download_video开始执行，post_url: {post_url}, download_addr: {download_addr}") utils.log_print(f"开始下载视频: {download_addr}, and temp_dir: {temp_dir}") username = post_url.split('/')[-3] user_info = User(user_url=self.get_user_url(username), username=username) post_info = Post(post_url=post_url,pub_time=publish_time) post_model = PostModel(self, post=post_info, author=user_info, content_type=Post_Content_Type.VIDEO) file_ext = '.mp4' mp4_file_path = os.path.join(temp_dir, f'{utils.str_to_md5(download_addr)}{file_ext}') mp4_file_name = os.path.basename(mp4_file_path) mp4_file_tmp_path = os.path.join(temp_dir, mp4_file_name) download(mp4_file_tmp_path, url=download_addr) file_info = File(file_url=download_addr, file_ext=file_ext, file_type=File_Type.VIDEO) save_model = post_model.create_batch_save_model() save_model.add_file(file=file_info) save_model.save_zip(file_path=mp4_file_tmp_path, file=file_info) self.video_num += 1 self.zip_num += 1 def save_detail(self, tweet_dict, detail_page, post_dict): utils.log_print("save_detail开始执行") utils.log_print(tweet_dict) publish_time = tweet_dict['publish_time'] publish_url = tweet_dict['publish_url'] publish_title = tweet_dict['publish_title'] publish_content = tweet_dict['publish_content'] publish_images = tweet_dict['publish_images'] author = tweet_dict['author'] nickname, username = self.split_by_last_at(author) avatar_image = tweet_dict['avatar_image'] reply_num = tweet_dict['reply_num'] repost_num = tweet_dict['repost_num'] like_num = tweet_dict['like_num'] bookmark_num = tweet_dict['bookmark_num'] parent_url = post_dict.get("parent_url") root_url = post_dict.get("root_url") depth = post_dict.get("depth") followers_num, following_num = self.parse_follow_and_fan(detail_page, username) user_info = User(user_url=self.get_user_url(username), username=username, nickname=nickname, follow_count=following_num, fans_count=followers_num) post_info = Post(post_url=publish_url, content=publish_content, title=publish_title, pub_time=publish_time, content_type=Post_Content_Type.IMG, comment_count=reply_num, forward_count=repost_num, like_count=like_num, favorite_count=bookmark_num, root_url=root_url, parent_url=parent_url, user_id=user_info.id) post_model = PostModel(self, post=post_info, author=user_info, content_type=Post_Content_Type.IMG) if avatar_image: utils.log_print(f"开始下载头像: {avatar_image}, and temp_dir: {temp_dir}") file_ext = '.jpg' img_file_tmp_path = os.path.join(temp_dir, f'{utils.str_to_md5(avatar_image)}{file_ext}') result = download_image(img_file_tmp_path, avatar_image) if result == 0: file_info = File(file_url=avatar_image, origin_url=avatar_image, file_ext=file_ext, file_type=File_Type.IMG, title=f"{username}的头像") user_info.set_user_profile(file_info) save_model = BaseModel(self) save_model.add_user(user=user_info) save_model.add_file(file=file_info) save_model.save_zip(file_path=img_file_tmp_path, file=file_info) self.profile_num += 1 self.zip_num += 1 else: utils.log_print(f"下载头像失败: {avatar_image}") for item in publish_images: if item == "": continue utils.log_print(f"开始下载图片: {item}, and temp_dir: {temp_dir}") file_ext = '.jpg' img_file_tmp_path = os.path.join(temp_dir, f'{utils.str_to_md5(item)}{file_ext}') result = download_image(img_file_tmp_path, item) if result == 0: file_info = File(file_url=item, origin_url=item, file_ext=file_ext, file_type=File_Type.IMG, title=publish_title) save_model = post_model.create_batch_save_model() save_model.add_file(file=file_info) save_model.save_zip(file_path=img_file_tmp_path, file=file_info) self.image_num += 1 self.zip_num += 1 post_model.save_content() self.txt_num += 1 # 处理评论，每条评论也是一个推文 if reply_num > 0: self.parse_comments(detail_page, root_url, publish_url, depth, reply_num) def parse_follow_and_fan(self, detail_page, username): time.sleep(random.randint(3, 5)) # 处理粉丝数和关注数 following_num = None followers_num = None user_page = detail_page.context.new_page() try: user_page.goto(self.get_user_url(username)) expect(user_page.locator('article').nth(0)).to_be_visible(timeout=30000) utils.log_print(f"获取用户粉丝数和关注数") following_soup = BeautifulSoup(user_page.locator(f'a[href="/{username}/following"]').inner_html(), "lxml") followers_soup = BeautifulSoup( user_page.locator(f'a[href="/{username}/verified_followers"]').inner_html(), "lxml") following_num = following_soup.find_all('span')[0].get_text() followers_num = followers_soup.find_all('span')[0].get_text() utils.log_print(f" ---> 用户关注数: {following_num}, 粉丝数: {followers_num}") except Exception as e: utils.log_print(f"获取用户粉丝数和关注数出错: {e}") finally: user_page.close() return self.convert_int(followers_num), self.convert_int(following_num) @staticmethod def convert_int(text:str): # 去除数字中的逗号 text = text.replace(',', '') number_match = re.search(r"(\d+(?:\.\d+)?)(K|M|万|亿|$)", text) if number_match: value = float(number_match.group(1)) unit = number_match.group(2) if unit == '万': return int(value * 10000) elif unit == '亿': return int(value * 100000000) elif unit == 'K': return int(value * 1000) elif unit == 'M': return int(value * 1000000) else: return int(value) return 0 def parse_comments(self, detail_page, root_url, parent_url, depth, reply_num): utils.log_print(f'高度：{detail_page.evaluate("() => document.documentElement.scrollHeight")}') detail_page.evaluate("() => window.scrollTo(0, document.documentElement.scrollHeight)") time.sleep(5) # detail_page.screenshot(path=os.path.join(temp_dir, f'x_comment_screenshot_{parent_url.split("/")[-1]}.png')) locator_article = detail_page.locator('article[tabindex="0"]') locator_article_count = locator_article.count() utils.log_print(f'开始解析评论, {locator_article_count}') for comment_index in range(locator_article_count): self.sum += 1 if self.sum % 10 == 0: self.update_count() try: comment_soup = BeautifulSoup(locator_article.nth(comment_index).inner_html(), "lxml") if 'style="text-overflow: unset;">Ad</span>' in str(comment_soup): self.ad_num += 1 continue detail_time_element = comment_soup.find("time") # publish_time = detail_time_element.get("datetime") publish_url = "https://x.com" + detail_time_element.find_parent().get("href") if self.diverge_user: user_name__find_all = comment_soup.find("div", attrs={"data-testid": "User-Name"}).find_all('span') user_id = user_name__find_all[-1].get_text() if not user_id.startswith('@'): utils.log_print(f"-1未定位到用户ID，使用昵称: {user_name__find_all[-2].get_text()}") user_id = user_name__find_all[-2].get_text() if user_id.startswith('@'): user_id = user_id[1:] utils.log_print(f"开始按用户ID: {user_id}发散") if self.processed_username.get(user_id): utils.log_print(f'用户ID已重复，processed_username: {user_id}') else: utils.log_print(f'用户ID未重复，processed_username: {user_id}') self.processed_username[user_id] = time.time() self.handle_user(detail_page.context, user_id) if self.processed_post.get(publish_url): self.repeat_num += 1 utils.log_print(f'评论已重复，processed_post: {publish_url}') else: utils.log_print(f'评论未重复，publish_url: {publish_url}') self.processed_post[publish_url] = time.time() post_dict = { "post_url": publish_url, "parent_url": parent_url, "root_url": root_url, "depth": depth + 1, } utils.log_print(f"评论加入队列: {post_dict}") dao.put_spider_work_queue(self.name, json.dumps(post_dict)) except Exception as e: self.error_num += 1 utils.log_print(f"解析评论出错: {e}") traceback.print_exc() @staticmethod def split_by_last_at(author:str): last_at_index = author.rfind('@') if last_at_index != -1: return author[:last_at_index], author[last_at_index + 1:] else: # 如果没有找到 @，可以根据实际需求返回合适的值，这里返回原字符串和空字符串 return author, ""