[Python] 纯文本查看 复制代码
import logging
from pyppeteer import launch
from pyppeteer.errors import TimeoutError
import asyncio
import json
from os import makedirs
from os.path import exists
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(levelname)s: %(message)s')
INDEX_URL = 'https://spa2.scrape.center/page/{page}'
TIMEOUT = 10
TOTAL_PAGE = 10
WINDOW_WIDTH, WINDOW_HEIGHT = 1366, 768 # 浏览器大小
HEADLESS = False # 定义变量，用来指定是否启用Pyppeteer的无头模式，如果其值为False，则启动后弹出窗口
browser, tab = None, None # 声明变量，浏览器对象和新建页面选项卡，后面设置为全局变量
async def init():
global browser, tab
browser = await launch(headless=HEADLESS,
args=['--disable-infobars',
f'--window-size={WINDOW_WIDTH},{WINDOW_HEIGHT}']) # 给headless传入参数设置非无头模式，通过args参数指定隐藏提示条和设置浏览器窗口的宽高
tab = await browser.newPage()
await tab.setViewport({'width': WINDOW_WIDTH, 'height': WINDOW_HEIGHT})
async def scrape_page(url, selector): # 链接，选择器
logging.info('scraping %s', url)
try:
await tab.goto(url)
await tab.waitForSelector(selector, options={
'timeout': TIMEOUT * 1000
}) # 传入选择器，规定最大等待时长
except TimeoutError:
logging.error('error occurred while scraping %s', url, exc_info=True)
async def scrape_index(page):
url = INDEX_URL.format(page=page)
await scrape_page(url, '.item .name') # 传入的选择器时电影的名称，意味着电影名称加载出来就代表页面加载成功了
async def parse_index():
return await tab.querySelectorAllEval('.item .name', 'nodes => nodes.map(node => node.href)')
async def scrape_detail(url):
await scrape_page(url, 'h2')
async def parse_detail():
url = tab.url
name = await tab.querySelectorEval('h2', 'node => node.innerText')
categories = await tab.querySelectorAllEval('.categories button span', 'nodes => nodes.map(node => node.innerText)')
cover = await tab.querySelectorEval('.cover', 'node => node.src')
score = await tab.querySelectorEval('.score', 'node => node.innerText')
drama = await tab.querySelectorEval('.drama p', 'node => node.innerText')
return {
'url': url,
'name': name,
'categories': categories,
'cover': cover,
'score': score,
'drama': drama
}
RESULTS_DIR = 'results' # 定义目录
exists(RESULTS_DIR) or makedirs(RESULTS_DIR) # 确定路径在不在，没有新建
async def save_data(data): # 储存
name = data.get('name')
data_path = f'{RESULTS_DIR}/{name}.json'
json.dump(data, open(data_path, 'w', encoding='utf-8'), ensure_ascii=False, indent=2)
async def main():
await init()
try:
for page in range(1, TOTAL_PAGE + 1):
await scrape_index(page)
detail_urls = await parse_index()
for detail_url in detail_urls:
await scrape_detail(detail_url)
detail_data = await parse_detail()
await save_data(detail_data)
logging.info('detail_urls %s', detail_data)
finally:
await browser.close()
if __name__ == '__main__':
asyncio.get_event_loop().run_until_complete(main())