编辑采集器       启动采集器

属性	值	属性	值
名称	common_news	创建时间	2025-04-02 10:10:02
标题	common_news	更新时间	-
状态	stop	开始时间	-
停止类型		结束时间	-
信号		运行时长	-
进程ID		进程名称
作业ID		作业目录
日志文件	(查看日志)     (tail日志)
采集器类型	scrapy	文件	common_news_spider.py
起始urls(每行一个) (若为空则使用代码中 的start_urls)

+新增命令

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

+新增定时任务

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

历史作业 ID	开始时间	结束时间	运行时长	进程ID	进程名	日志
删除最后 选择条数 条历史作业 确定

代码

# import json # import time # import traceback # from datetime import datetime # from urllib.parse import urlparse, urljoin # # from playwright.sync_api import sync_playwright # # from spider.spiders.common.base_spider import BaseCommonNewsSpider # # # # 通用新闻采集器 # class CommonNewsSpider(BaseCommonNewsSpider): # # 最大重试次数 # max_retries = 3 # # 新闻总条数 # total_count = 0 # # # 线程池 # # thread_pool = CommonNewsThreadPool(4) # # futures = [] # # # 开始爬取 # def execute(self): # self.record_log(f'通用新闻采集器v1任务【{self.data.get("net_name")}】开始执行') # start_time = datetime.now() # # 循环遍历所有频道 # channels = json.loads(self.data.get('channel_url')) # with sync_playwright() as p: # browser = p.chromium.launch(headless=True) # context = browser.new_context() # for channel in channels: # channel_name = channel.get("channelName") # channel_url = channel.get("channelUrl") # self.record_log(f'频道【{channel_name}】开始采集') # retries = 0 # while retries < self.max_retries: # try: # # 1、打开浏览器，一个频道一个浏览器 # # headless = False if platform.system() == 'Windows' else True # # proxy = {'server': self.get_random_proxy()} # # browser = p.chromium.launch(headless=headless, proxy=proxy) # page = context.new_page() # page.goto(channel_url, wait_until='domcontentloaded', timeout=1200000) # # # 2、处理第一页 # self.record_log(f'开始解析频道，【{channel_url}】，页码【{1}】') # try: # href_list = self.parse_news_list(page, channel) # page.close() # for href in href_list: # # 校验是否可以采集 # is_parsed = self.check_news_parsed_url(href) # if not is_parsed: # continue # # 计数 # self.total_count += 1 # # 获取新闻数据 # news_data = self.parse_news_details(context, href) # # 保存数据 # self.save_data(news_data, channel) # except: # self.record_log(f'解析新闻列表异常：频道【{channel}】，exception：{traceback.format_exc()}') # # # 3、翻页 # self.roll_page(context, page, channel) # break # except: # self.record_log( # f'异常：第 {retries} 次尝试失败，{channel_name} {channel_url}，exception：{traceback.format_exc()}') # retries += 1 # if retries < self.max_retries: # time.sleep(5) # if retries == self.max_retries: # self.record_log(f"达到最大重试次数，无法访问目标 URL。{channel_name} {channel_url}") # context.close() # browser.close() # self.record_log(f'频道【{channel_url}】采集完成') # p.stop() # # 结束，打印汇总信息 # self.record_log(f'任务【{self.data.get("net_name")}】执行完成，共【{self.total_count}】条新增新闻') # end_time = datetime.now() # self.record_log( # f'通用新闻采集器v1任务【{self.data.get("net_name")}】执行结束，耗时：{(end_time - start_time).seconds} 秒') # # # 解析新闻列表 # def parse_news_list(self, page, channel): # channel_url = channel.get("channelUrl") # href_list = [] # try: # # 根据列表区域xpath，找下面的所有a标签 # # 列表区域xpath为空，获取频道域名，先根据是否包含频道域名来确定是否是新闻链接 # links = None # if self.data.get('list_xpath'): # list_element = page.locator(f'xpath={self.data.get("list_xpath")}') # if list_element.count() > 0: # list_element = list_element.nth(0) # links = list_element.locator('a') # else: # links = page.locator('a') # # if not links: # raise Exception(f'未找到新闻列表区域，请检查配置') # # # 匹配所有a标签，查找满足channel条件的a标签 # a_count = links.count() # for i in range(a_count): # a_element = links.nth(i) # if not a_element.is_visible(): # continue # href = a_element.get_attribute('href') # parsed = urlparse(href) # # 判断是否为相对路径，如果是相对路径，则拼接成绝对路径 # if not parsed.scheme and not parsed.netloc: # href = urljoin(page.url, href) # # 判断是否满足通用新闻条件 # if self.check_href(channel_url, href): # href_list.append(href) # # # 异步执行 # # future = self.thread_pool.submit_task(self.parse_news_details, channel, href) # # self.futures.append(future) # except: # raise Exception(f'解析新闻列表异常：频道【{channel}】，，exception：{traceback.format_exc()}') # return href_list # # # 检查是否是可以采集的新闻链接 # def check_href(self, channel_url, href): # hostname = urlparse(channel_url).hostname # if href.endswith('.jpg'): # return False # if href.endswith('.png'): # return False # if href.endswith('.gif'): # return False # if href.endswith('.jpeg'): # return False # if href.endswith('.svg'): # return False # if href.endswith('.mp4'): # return False # if href.endswith('.mp3'): # return False # if href.endswith('.pdf'): # return False # if href == channel_url: # return False # if hostname not in href: # return False # return True # # # 解析新闻详情 # def parse_news_details(self, context, href): # if "http" not in href: # self.record_log(f'非法链接：{href}') # return {} # # 开始采集 # retries = 0 # page = None # while retries < self.max_retries: # try: # # with sync_playwright() as p: # # headless = False if platform.system() == 'Windows' else True # # proxy = {'server': self.get_random_proxy()} # # browser = p.chromium.launch(headless=headless, proxy=proxy) # # browser = p.chromium.launch(headless=True) # # context = browser.new_context() # page = context.new_page() # page.goto(href, wait_until='domcontentloaded', timeout=1200000) # # # 解析新闻详情 # title = self.get_title(self.data.get('title_xpath'), page) # content = self.get_content(self.data.get('content_xpath'), page) # img_list = self.get_img_list(self.data.get('image_xpath'), page) # author = self.get_author(self.data.get('author_xpath'), page) # time_str = self.get_time(self.data.get('time_xpath'), page) # news_data = { # 'title': title, # 'content': content, # 'img_list': img_list, # 'author': author, # 'time_str': time_str if time_str else '', # 'href': href # } # return news_data # except: # self.record_log(f'解析新闻详情异常：{href}，exception：{traceback.format_exc()}') # retries += 1 # if retries < self.max_retries: # time.sleep(5) # finally: # # 关闭页面 # page.close() # if retries == self.max_retries: # raise Exception(f"达到最大重试次数，无法访问目标 URL。{href}") # return {} # # # 获取标题 # def get_title(self, title_xpath, page): # if not title_xpath: # return '' # # title_xpath = title_xpath.strip() # title = '' # try: # title_element = page.locator(f'xpath={title_xpath}') # if title_element.count() > 0: # title = title_element.nth(0).text_content() # except: # raise Exception(f'获取标题异常：{title_xpath}，exception：{traceback.format_exc()}') # return self.replace_space_and_line_break(title) # # # 获取内容 # def get_content(self, content_xpath, page): # if not content_xpath: # return '' # # content_xpath = content_xpath.strip() # content = '' # try: # content_element = page.locator(f'xpath={content_xpath}') # if content_element.count() > 0: # for i in range(content_element.count()): # content += content_element.nth(i).text_content() + '\r\n' # except: # raise Exception(f'获取内容异常：{content_xpath}，exception：{traceback.format_exc()}') # return content.strip() # # # 获取图片 # # xpath和正文p的父级一样 # def get_img_list(self, image_xpath, page): # if not image_xpath: # return [] # # image_xpath = image_xpath.strip() # img_list = [] # try: # img_list_element = page.locator(f'xpath={image_xpath}') # img_list_count = img_list_element.count() # if img_list_count > 0: # img_element = img_list_element.nth(0) # imgs = img_element.locator('img') # img_count = imgs.count() # for i in range(img_count): # img = imgs.nth(i) # img_url = img.get_attribute('src') # image_code = self.data.get("image_code") # if img_url and image_code: # img_url = eval(image_code) # if img_url and img_url not in img_list: # img_list.append(img_url) # except: # raise Exception(f'获取图片异常：{image_xpath}，exception：{traceback.format_exc()}') # return img_list # # # 获取作者 # def get_author(self, author_xpath, page): # if not author_xpath: # return '' # # author_xpath = author_xpath.strip() # author = '' # try: # author_element = page.locator(f'xpath={author_xpath}') # if author_element.count() > 0: # author = author_element.nth(0).text_content().strip(',') # author_code = self.data.get("author_code") # if author and author_code: # author = eval(author_code) # except: # raise Exception(f'获取作者异常：{author_xpath}，exception：{traceback.format_exc()}') # replace_list = ['作者：', 'BY:', 'By:', 'by:', '作者', 'BY', 'By', 'by'] # for replace_str in replace_list: # author = author.replace(replace_str, '') # return self.replace_space_and_line_break(author) # # # 获取时间 # def get_time(self, time_xpath, page): # if not time_xpath: # return '' # # time_xpath = time_xpath.strip() # time_str = '' # try: # time_element = page.locator(f'xpath={time_xpath}') # if time_element.count() > 0: # time_str = time_element.nth(0).text_content() # time_code = self.data.get("time_code") # if time_str and time_code: # time_str = eval(time_code) # except: # raise Exception(f'获取时间异常：{time_xpath}，exception：{traceback.format_exc()}') # return self.replace_space_and_line_break(time_str) # # # 翻页 # def roll_page(self, context, page, channel): # if self.max_page <= 1: # return # # channel = channel.get("channelUrl") # for i in range(self.max_page - 1): # page_num = i + 2 # self.record_log(f'开始翻页：{channel}，第{page_num}页') # try: # next_page_element = page.locator(f'xpath={self.data.get("next_page_xpath")}') # if not next_page_element.is_visible() or not next_page_element.is_enabled(): # self.record_log(f'翻页结束：{channel} 第{i + 1}页') # break # # next_page_element.scroll_into_view_if_needed() # next_page_element.click(force=True) # page.wait_for_load_state(state='domcontentloaded', timeout=1200000) # # 解析新闻列表，获取新闻链接列表，并逐个获取新闻详情，保存数据 # self.record_log(f'开始解析频道，【{channel}】，页码【{page_num}】') # try: # href_list = self.parse_news_list(page, channel) # for href in href_list: # # 获取新闻数据 # self.record_log(f'开始采集新闻详情【{href}】') # news_data = self.parse_news_details(context, href) # self.record_log(f'采集到的新闻详情数据：{json.dumps(news_data, ensure_ascii=False, indent=4)}') # # 保存数据 # self.save_data(news_data, channel) # except: # self.record_log(f'解析频道异常：{channel} 第{i + 1}页，exception：{traceback.format_exc()}') # except: # self.record_log(f'翻页异常：{channel} 第{i + 1}页，exception：{traceback.format_exc()}') # self.record_log(f'翻页完成：{channel}，第{i + 1}页')