编辑采集器       启动采集器

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

+新增命令

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

+新增定时任务

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

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

代码

from urllib.parse import urlparse, parse_qs from bs4 import BeautifulSoup from scrapy.http import Response import bs4 import json import utils from spider.spiders.common.base_spider import BaseQueueListToDetailsSpider # TheWallStreetJournal class UsNBCNewsSpider(BaseQueueListToDetailsSpider): name = 'us_cnbc_news' desc = '美国-消费者新闻与商业频道(zxc1.2)' net_name = 'CNBC News,消费者新闻与商业频道,美国' net_url = 'www.nbcnews.com' language = 'en' nation_category = 1 channel_name = '每日最新' channel_url = 'https://www.cnbc.com' channel_desc = '消费者新闻与商业频道>每日最新' max_page = 100000 now_page = 0 from_reached_url = False host_key = 'cnbc.com' exclude_urls = [] start_urls = [ { 'url': 'https://www.cnbc.com/business/', 'max_page': 1, 'language': 'en', 'nation_category': 1, 'channel_name': 'business', 'channel_url': 'https://www.cnbc.com/business/', 'channel_desc': 'cnbc>business' },{ 'url': 'https://www.cnbc.com/technology/', 'max_page': 1, 'language': 'en', 'nation_category': 1, 'channel_name': 'technology', 'channel_url': 'https://www.cnbc.com/technology/', 'channel_desc': 'cnbc>technology' },{ 'url': 'https://www.cnbc.com/politics/', 'max_page': 1, 'language': 'en', 'nation_category': 1, 'channel_name': 'politics', 'channel_url': 'https://www.cnbc.com/politics/', 'channel_desc': 'cnbc>politics' },{ 'url': 'https://www.cnbc.com/tv/', 'max_page': 1, 'language': 'en', 'nation_category': 1, 'channel_name': 'tv', 'channel_url': 'https://www.cnbc.com/tv/', 'channel_desc': 'cnbc>tv' } ] """ """ def build_next_list_url(self, response, list_url_item): return list_url_item def build_detail_urls(self, response: Response): detail_url_list = [] if response: try: soup = BeautifulSoup(response.text, 'lxml') a_list = soup.find_all('a', {'href': True}) for a in a_list: href = a['href'] path = urlparse(href).path last_part = path.split('/')[-1] hyphen_count = last_part.count('-') if hyphen_count > 2 and 'https://' in href: print(f'a[href]:{href}') detail_url_list.append(href) except Exception as e: utils.log_error(f'get img fail: {e}') return detail_url_list # props initialState front curation layouts[1] packages[1] items 每一个computedValues下的 url def get_img(soup :BeautifulSoup): img_urls = [] try: #文章配图 print('get img') figure_hrefs = soup.find_all('picture',{'data-test':'Picture'}) if figure_hrefs: print('pictures') for figure_href in figure_hrefs: if figure_href.find('img', attrs={'src': True}): print('picture-img') a_href = figure_href.find('img', attrs={'src': True}) img_url = a_href['href'] print(img_url) img_urls.append(img_url) #从meta中获取图片 if len(img_urls) == 0: image_metas = soup.find_all('meta',{'itemprop':'image','content':True}) if image_metas: for image_meta in image_metas: img_url = image_meta['content'] if img_url not in img_urls: img_urls.append(img_url) scripts = soup.find_all('script') if scripts: for script in scripts: data = script.text if '.mp4' in data: data_dict = json.loads(data) # 提取 .mp4 链接 content_url = data_dict['video']['contentUrl'] print("提取的 .mp4 文件：") print(content_url) img_urls.append(content_url) #PDF下载 pdf_link = soup.find('a', attrs={'href': True,'title':'DOWNLOAD PDF'}) if pdf_link: pdf_url = pdf_link['href'] img_urls.append(pdf_url) except Exception as e: utils.log_error(f'get img fail: {e}') return img_urls img_href_rules = [ { 'op': 'func', 'name': get_img } ] title_rules = [ { 'op': 'find', 'tag': 'h1' }, { 'op': 'text' } ] def get_author(soup: BeautifulSoup): name = '' try: author_names = soup.find_all('div',class_ = 'ArticleHeader-author') for author in author_names: if name == '': name = author.find('a',class_ = 'Author-authorName').text.strip() else: name = name + ',' + author.find('a', class_='Author-authorName').text.strip() if name == '': name_div = soup.find('div', class_='Author-author') if name_div: name = name_div.text.strip() except Exception as e: utils.log_error(f'get img fail: {e}') return name author_rules = [ { 'op': 'func', 'name': get_author } ] def get_time(soup: BeautifulSoup): time = '' try: time_span = soup.find('time', attrs={'datetime':True}) if time_span: time = time_span['datetime'] except Exception as e: utils.log_error(f'get img fail: {e}') return time time_rules = [ { 'op': 'func', 'name': get_time } ] def get_content(soup: BeautifulSoup): content = '' try: if soup.find('div', class_='ClipPlayer-clipPlayerIntroSummary'): video_content = soup.find('div', class_='ClipPlayer-clipPlayerIntroSummary').text.strip() content = content + video_content key_div = soup.find('div',class_ = 'RenderKeyPoints-list') if key_div: li_list = key_div.find_all('li') if li_list: print('=====find li_list=======') for li in li_list: content = content +'\n\n'+ li.text.strip().replace('\n', '').replace('\t', '') content_div = soup.find('div',class_ = 'ArticleBody-articleBody') if content_div: p_list = content_div.find_all('p') if p_list: print('=====find p_list=======') for p in p_list: content = content +'\n\n'+ p.text.strip().replace('\n', '').replace('\t', '') pdf_content = soup.find('div',class_ = 'privacy-body print-only') if pdf_content: content = content + '\n\n' + pdf_content.text.strip().replace('\n', '').replace('\t', '') except Exception as e: utils.log_error(f'get img fail: {e}') return content content_rules = [ { 'op': 'func', 'name': get_content } ]