import json
from os import makedirs
from os.path import exists
import requests
import logging
import re
from urllib.parse import urljoin
import multiprocessing
# 日志配置：记录时间、级别、内容
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(levelname)s: %(message)s')
# 核心常量
BASE_URL = 'https://ssr1.scrape.center'
TOTAL_PAGE = 10
RESULTS_DIR = 'results' # 存储JSON的文件夹
if not exists(RESULTS_DIR):
makedirs(RESULTS_DIR)
def scrape_page(url):
"""
通用「套娃」爬取函数：统一处理请求、状态码、异常
:param url: 目标URL
:return: 成功返回HTML，失败返回None
"""
logging.info('正在爬取 %s...', url)
try:
# 坑1：增加 timeout（10秒）防止死等
response = requests.get(url, timeout=10)
if response.status_code == 200:
return response.text
logging.error('无效状态码 %d，URL：%s', response.status_code, url)
except requests.RequestException:
logging.error('请求异常！URL：%s', url, exc_info=True)
def scrape_index(page):
"""
构造并爬取指定页码的列表页
:param page: 页码
:return: 列表页HTML
"""
index_url = f'{BASE_URL}/page/{page}'
return scrape_page(index_url)
def parse_index(html):
"""
解析列表页，返回详情页URL生成器（节省内存）
:param html: 列表页HTML
:return: 详情页URL生成器
"""
if not html: return [] # 坑2：空值保护，防止后续正则崩溃
# 坑3：所有正则前加 r'' 前缀，解决转义警告
pattern = re.compile(r'<a.*?href="(.*?)".*?class="name">')
items = re.findall(pattern, html)
if not items:
return []
for item in items:
detail_url = urljoin(BASE_URL, item) # 自动拼接base_url（处理相对路径）
logging.info('获取到详情页链接：%s', detail_url)
yield detail_url
def scrape_detail(url):
"""
爬取详情页
:param url: 详情页URL
:return: 详情页HTML
"""
return scrape_page(url)
def parse_detail(html):
"""
解析详情页，返回电影信息字典
:param html: 详情页HTML
:return: 电影信息字典
"""
if not html: return None # 空值保护
# 逐个定义正则，加re.S让.匹配换行符
cover_pattern = re.compile(r'class="item.*?<img.*?src="(.*?)".*?class="cover">', re.S)
name_pattern = re.compile(r'<h2.*?>(.*?)</h2>')
categories_pattern = re.compile(r'<button.*?category.*?<span>(.*?)</span>.*?</button>', re.S)
published_at_pattern = re.compile(r'(\d{4}-\d{2}-\d{2})\s?上映')
drama_pattern = re.compile(r'<div.*?drama.*?>.*?<p.*?>(.*?)</p>', re.S)
score_pattern = re.compile(r'<p.*?score.*?>(.*?)</p>', re.S)
# 逐个提取字段，找不到给默认值，保证程序不会崩
name_match = re.search(name_pattern, html)
name = name_match.group(1).strip() if name_match else "未命名电影"
cover_match = re.search(cover_pattern, html)
cover = cover_match.group(1).strip() if cover_match else None
categories = re.findall(categories_pattern, html)
pub_match = re.search(published_at_pattern, html)
published_at = pub_match.group(1) if pub_match else None
drama_match = re.search(drama_pattern, html)
drama = drama_match.group(1).strip() if drama_match else None
score_match = re.search(score_pattern, html)
score = float(score_match.group(1).strip()) if score_match else None
return {
'cover': cover,
'name': name,
'categories': categories,
'published_at': published_at,
'drama': drama,
'score': score
}
def save_data(data):
"""
保存单条电影数据到JSON文件
:param data: 电影信息字典
"""
if not data: return
name = data.get('name')
# 坑4：完整清洗文件名非法字符（Windows/Linux通用）
safe_name = re.sub(r'[\\/:*?"<>|]', '_', name)
data_path = f'{RESULTS_DIR}/{safe_name}.json'
# 坑5：用with open自动关闭文件，避免资源泄漏
try:
with open(data_path, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2) # ensure_ascii=False防止中文乱码
logging.info('已成功保存：%s', data_path)
except Exception as e:
logging.error('保存失败！文件路径：%s，错误：%s', data_path, e)
def main(page):
"""
单页主处理函数：串联整个单页流程
:param page: 页码
"""
index_html = scrape_index(page)
detail_urls = parse_index(index_html)
for detail_url in detail_urls:
detail_html = scrape_detail(detail_url)
data = parse_detail(detail_html)
if data:
logging.info('获取到电影数据：%s', data['name'])
save_data(data)
if __name__ == '__main__':
# 多进程并行爬取（单进程可以直接循环 main(page) for page in range(1, 11)）
pool = multiprocessing.Pool()
pages = range(1, TOTAL_PAGE + 1)
pool.map(main, pages)
pool.close() # 关闭进程池，不再接受新任务
pool.join() # 等待所有子进程完成
