编辑采集器       终止采集器

属性	值	属性	值
名称	us_the_intercept_news	创建时间	2025-04-02 10:10:02
标题	美国-The Intercept拦截者---(US-1-采)(zxc)	更新时间	-
状态	running	开始时间	2025-07-04 00:26:05
停止类型		结束时间	-
信号		运行时长	90.2小时
进程ID	3823	进程名称	Process-1
作业ID	1f40dea3-8a2a-41c5-b7a6-55c77050438c	作业目录	records/jobs/1f40dea3-8a2a-41c5-b7a6-55c77050438c
日志文件	records/logs/1f40dea3-8a2a-41c5-b7a6-55c77050438c.log     (查看日志)     (tail日志)
采集器类型	scrapy	文件	us_the_intercept_news_spider.py
起始urls(每行一个) (若为空则使用代码中 的start_urls)

+新增命令

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

+新增定时任务

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

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

代码

import json from urllib.parse import urlparse, parse_qs import bs4 import requests from bs4 import BeautifulSoup from scrapy.http import Response import utils from spider.spiders.common.base_spider import BaseQueueListToDetailsSpider import urllib.parse import re from spider.spiders.common.helper import update_url class UsTheInterceptNewsSpider(BaseQueueListToDetailsSpider): name = 'us_the_intercept_news' desc = '美国-The Intercept拦截者---(US-1-采)(zxc)' net_name = 'The Intercept,拦截者新闻,美国' net_url = 'theintercept.com/' language = 'en' nation_category = 1 channel_name = '新闻列表' channel_url = 'https://theintercept.com/' channel_desc = '拦截者新闻>新闻列表' max_page = 1000000 now_page = 0 from_reached_url = False host_key = 'theintercept.com' exclude_urls = [] start_urls = [ { 'url': 'https://theintercept.com/politics/', 'max_page': 1, 'language': 'en', 'nation_category': 1, 'channel_name': 'politics', 'channel_url': 'https://theintercept.com/politics/', 'channel_desc': 'theintercept>politics' } , { 'url': 'https://theintercept.com/collections/end-of-roe/', 'max_page': 1, 'language': 'en', 'nation_category': 1, 'channel_name': 'end-of-roe', 'channel_url': 'https://theintercept.com/collections/end-of-roe/', 'channel_desc': 'theintercept>end-of-roe' }, { 'url': 'https://theintercept.com/justice/', 'max_page': 1, 'language': 'en', 'nation_category': 1, 'channel_name': 'justice', 'channel_url': 'https://theintercept.com/justice/', 'channel_desc': 'theintercept>justice' } ,{ 'url': 'https://theintercept.com/collections/israel-palestine/', 'max_page': 1, 'language': 'en', 'nation_category': 1, 'channel_name': 'israel-palestine', 'channel_url': 'https://theintercept.com/collections/israel-palestine/', 'channel_desc': 'theintercept>israel-palestine' },{ 'url': 'https://theintercept.com/technology/', 'max_page': 1, 'language': 'en', 'nation_category': 1, 'channel_name': 'technology', 'channel_url': 'https://theintercept.com/technology/', 'channel_desc': 'theintercept>technology' },{ 'url': 'https://theintercept.com/environment/', 'max_page': 1, 'language': 'en', 'nation_category': 1, 'channel_name': 'environment', 'channel_url': 'https://theintercept.com/environment/', 'channel_desc': 'theintercept>environment' }, { 'url': 'https://theintercept.com/collections/end-of-roe/', 'max_page': 1, 'language': 'en', 'nation_category': 1, 'channel_name': 'end-of-roe', 'channel_url': 'https://theintercept.com/collections/end-of-roe/', 'channel_desc': 'theintercept>end-of-roe' } ,{ 'url': 'https://theintercept.com/special-investigations/', 'max_page': 1, 'language': 'en', 'nation_category': 1, 'channel_name': 'special-investigations', 'channel_url': 'https://theintercept.com/special-investigations/', 'channel_desc': 'theintercept>special-investigations' },{ 'url': 'https://theintercept.com/voices/', 'max_page': 1, 'language': 'en', 'nation_category': 1, 'channel_name': 'voices', 'channel_url': 'https://theintercept.com/voices/', 'channel_desc': 'theintercept>voices' } ] def build_next_list_url(self, response, list_url_item): if response: updated_url = '' use_url = response.url print('use_url') print(use_url) if 'page' in use_url: # 使用正则表达式提取 page 参数的数字 match = re.search(r'/page/(\d+)/', use_url) print('match') if match: last_number = int(match.group(1)) # 获取最后的数字 print(f'last_number:{last_number}') new_number = last_number + 1 # 原有数字加 1 print(f'new_number:{new_number}') # 生成新的 URL updated_url = use_url.replace(f'/page/{last_number}/', f'/page/{new_number}/') print(f'updated_url:{updated_url}') else: updated_url = use_url + 'page/2' list_url_item[ 'use_url'] = updated_url return list_url_item def build_detail_urls(self, response: Response): detail_url_list = [] if response: soup = bs4.BeautifulSoup(response.text,'lxml') head_div = soup.find('div',class_ = 'flex flex-col gap-12') if head_div: print('找到head_div') articles = head_div.find_all('article', class_='wide-story-card',recursive=False) if len(articles)>0: for article in articles: if article.find('a', attrs={'href': True}): detail_url = article.find('a', attrs={'href': True})['href'] detail_url_list.append(detail_url) return detail_url_list def get_img(soup: BeautifulSoup): img_url_list = [] try: content_div = soup.find('div', class_ = 'entry-content__content') if content_div: img_urls = content_div.find_all('img',attrs={'src':True}) if len(img_urls)>0: for img in img_urls: img_url = img['src'] img_url_list.append(img_url) except Exception as e: utils.log_error(f'get img fail: {e}') return img_url_list img_href_rules = [ { 'op': 'func', 'name': get_img } ] title_rules = [ { 'op': 'find', 'tag': 'h1' }, { 'op': 'text' } ] # propos -> pageProps -> globalContent -> promo_items -> basic -> credits -> by -> name def get_author(soup: BeautifulSoup): name = '' try: title_str = soup.find('a',attrs={'rel':'author'}).text.strip().replace('\n', '').replace('\t', '') if title_str: if '-' in title_str: name_str = title_str.split('-')[0].strip() if 'By' in name_str: name = name_str.replace('By', '') if 'by' in name_str: name = name_str.replace('by', '') name = title_str print(name) else: meta_name = soup.find('meta',attrs={'name':'author','content':True}) if meta_name: name = meta_name['content'] 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 = soup.find('time',attrs={'datetime':True}).text if time: time = time.strip().replace('\n','').replace('\t','') except Exception as e: utils.log_error(f'get img fail: {e}') return time time_rules = [ { 'op': 'func', 'name': get_time } ] # propos -> pageProps -> globalContent -> content_elements -> {url} / {content...} (img /content) def get_content(soup: BeautifulSoup): content = '' try: content_div = soup.find('div', class_='entry-content__content') if content_div: content_ps = content_div.find_all('p') if len(content_ps)>0: for p in content_ps: content = content +'\n\n'+ p.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 } ]