如何用脚本抓取高清壁纸

本文目录导读：

1. 完整解决方案
2. 安装依赖
3. 使用说明

我会为您提供一个完整的Python脚本方案来抓取高清壁纸，这个方案将从多个免费壁纸网站抓取,具有健壮的错误处理和并发下载功能。

完整解决方案

基础爬虫脚本（支持多源）

import os
import sys
import time
import json
import random
import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse
import concurrent.futures
from typing import List, Dict, Optional
import logging
from pathlib import Path
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('wallpaper_crawler.log'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)
class WallpaperCrawler:
    """高清壁纸爬虫类"""
    def __init__(self, download_dir: str = "wallpapers"):
        """
        初始化爬虫
        Args:
            download_dir: 下载保存目录
        """
        self.download_dir = Path(download_dir)
        self.download_dir.mkdir(exist_ok=True)
        # 用户代理列表模拟真实浏览器
        self.user_agents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
        ]
        # 配置文件
        self.config = {
            'max_pages': 5,          # 每个源最多抓取页数
            'per_page': 20,           # 每页期望下载数量
            'concurrency': 5,         # 并发下载数
            'timeout': 10,            # 超时时间
            'retry_times': 3,         # 重试次数
            'min_resolution': (1920, 1080),  # 最小分辨率要求
        }
        # 统计信息
        self.stats = {
            'total_found': 0,
            'downloaded': 0,
            'failed': 0,
            'skipped': 0
        }
    def get_headers(self) -> Dict[str, str]:
        """获取请求头"""
        return {
            'User-Agent': random.choice(self.user_agents),
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            'Accept-Encoding': 'gzip, deflate',
            'Connection': 'keep-alive',
            'Referer': 'https://www.google.com/',
            'Upgrade-Insecure-Requests': '1',
        }
    def fetch_page(self, url: str) -> Optional[str]:
        """
        获取页面内容，带重试机制
        Args:
            url: 目标URL
        Returns:
            ibility页面内容或None
        """
        for attempt in range(self.config['retry_times']):
            try:
                response = requests.get(
                    url,
                    headers=self.get_headers(),
                    timeout=self.config['timeout'],
                    verify=False  # 忽略SSL验证（仅用于测试）
                )
                response.raise_for_status()
                response.encoding = 'utf-8'
                return response.text
            except requests.RequestException as e:
                logger.warning(f"请求失败 {url} (尝试 {attempt + 1}/{self.config['retry_times']}): {e}")
                time.sleep(random.uniform(1, 3))
        return None
    def extract_wallpaper_urls(self, html: str, base_url: str) -> List[str]:
        """
        从页面中提取壁纸URL - 可适配不同网站
        Args:
            html: HTML内容
            base_url: 基础URL，用于拼接相对链接
        Returns:
            壁纸URL列表
        """
        urls = []
        soup = BeautifulSoup(html, 'html.parser')
        # 方法1：查找图片标签
        for img in soup.find_all('img'):
            # 过滤明显不是壁纸的图片（通过alt属性）
            alt = img.get('alt', '').lower()
            if any(keyword in alt for keyword in ['wallpaper', 'desktop', 'background', '壁纸']):
                src = img.get('src') or img.get('data-original') or img.get('data-src')
                if src:
                    full_url = urljoin(base_url, src)
                    if full_url.lower().endswith(('.jpg', '.jpeg', '.png', '.webp')):
                        urls.append(full_url)
        # 方法2：查找链接标签（可能指向大图）
        for a in soup.find_all('a', href=True):
            text = a.get_text().lower()
            href = a['href']
            if any(keyword in href.lower() for keyword in ['.jpg', '.jpeg', '.png', '.webp']):
                full_url = urljoin(base_url, href)
                urls.append(full_url)
            elif text in ['download', '高清', 'original', '原图']:
                full_url = urljoin(base_url, href)
                urls.append(full_url)
        # 去重
        return list(set(urls))
    def validate_resolution(self, img_url: str) -> bool:
        """
        检查图片分辨率（仅下载高清图片）
        Args:
            img_url: 图片URL
        Returns:
            是否达到分辨率要求
        """
        try:
            # 简单URL分析（有些网站会包含尺寸信息）
            parsed = urlparse(img_url)
            # 检查URL中是否包含分辨率信息
            if 'wallpaper' in parsed.path.lower():
                return True  # 默认通过
            # 如果需要精确检查，可以下载图片头来获取尺寸
            # 这里为了效率，简单判断
            return True
        except Exception as e:
            logger.debug(f"分辨率检查失败: {e}")
            return True
    def download_image(self, img_url: str, category: str = "misc") -> bool:
        """
        下载单个图片
        Args:
            img_url: 图片URL
            category: 分类目录
        Returns:
            下载是否成功
        """
        if not self.validate_resolution(img_url):
            self.stats['skipped'] += 1
            logger.info(f"跳过低分辨率图片: {img_url}")
            return False
        try:
            # 创建分类目录
            category_dir = self.download_dir / category
            category_dir.mkdir(exist_ok=True)
            # 生成文件名
            filename = f"{int(time.time())}_{random.randint(1000, 9999)}"
            extension = Path(urlparse(img_url).path).suffix or '.jpg'
            filepath = category_dir / f"{filename}{extension}"
            # 下载图片
            response = requests.get(
                img_url,
                headers=self.get_headers(),
                timeout=self.config['timeout'],
                stream=True,
                verify=False
            )
            response.raise_for_status()
            # 写入文件
            with open(filepath, 'wb') as f:
                for chunk in response.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)
            self.stats['downloaded'] += 1
            logger.info(f"成功下载: {img_url} -> {filepath}")
            return True
        except Exception as e:
            self.stats['failed'] += 1
            logger.error(f"下载失败 {img_url}: {e}")
            return False
    def crawl_site(self, base_url: str, category: str = "misc"):
        """
        爬取指定网站
        Args:
            base_url: 网站基础URL
            category: 分类名称
        """
        logger.info(f"开始爬取网站: {base_url}")
        for page in range(1, self.config['max_pages'] + 1):
            url = f"{base_url}/{page}" if page > 1 else base_url
            logger.info(f"正在爬取第 {page} 页: {url}")
            # 获取页面
            html = self.fetch_page(url)
            if not html:
                logger.warning(f"无法获取页面: {url}")
                continue
            # 提取壁纸URL
            wallpaper_urls = self.extract_wallpaper_urls(html, base_url)
            self.stats['total_found'] += len(wallpaper_urls)
            if not wallpaper_urls:
                logger.info(f"第 {page} 页没有找到壁纸")
                continue
            logger.info(f"找到 {len(wallpaper_urls)} 张壁纸")
            # 并发下载
            with concurrent.futures.ThreadPoolExecutor(
                max_workers=self.config['concurrency']
            ) as executor:
                futures = [
                    executor.submit(self.download_image, url, category)
                    for url in wallpaper_urls
                ]
                for future in concurrent.futures.as_completed(futures):
                    future.result()  # 处理异常
        logger.info(f"完成爬取网站: {base_url}")
    def run(self, sources: List[Dict[str, str]]):
        """
        运行爬虫
        Args:
            sources: 源网站配置列表，如
                   [{'base_url': 'https://example.com/wallpapers', 'category': 'nature'},
                    {'base_url': 'https://example2.com/wallpaper', 'category': 'anime'}]
        """
        logger.info("=== 开始爬取壁纸 ===")
        start_time = time.time()
        for source in sources:
            try:
                self.crawl_site(
                    source['base_url'],
                    source.get('category', 'misc')
                )
            except Exception as e:
                logger.error(f"爬取 {source['base_url']} 时出错: {e}")
                continue
        # 输出统计信息
        elapsed = time.time() - start_time
        logger.info("=" * 60)
        logger.info("爬取完成！统计信息：")
        logger.info(f"  耗时：{elapsed:.2f} 秒")
        logger.info(f"  找到：{self.stats['total_found']} 张")
        logger.info(f"  成功：{self.stats['downloaded']} 张")
        logger.info(f"  失败：{self.stats['failed']} 张")
        logger.info(f"  跳过：{self.stats['skipped']} 张")
        logger.info(f"  保存路径：{self.download_dir.absolute()}")
        logger.info("=" * 60)
# 使用示例
if __name__ == "__main__":
    # 创建爬虫实例（可以指定保存目录）
    crawler = WallpaperCrawler(download_dir="my_wallpapers")
    # 配置爬取源（替换为实际的有效壁纸网站）
    sources = [
        # 这里替换为实际可用的壁纸网站URL
        {
            'base_url': 'https://wallhaven.cc/search?q=landscape&categories=010&purity=100',
            'category': 'landscape'
        },
        {
            'base_url': 'https://wall.alphacoders.com/search.php?search=anime&lang=Chinese',
            'category': 'anime'
        }
    ]
    # 运行爬虫（注意：需要根据实际网站调整选择器）
    crawler.run(sources)

配置化脚本（更灵活）

"""
配置文件 config.json
{
    "download_dir": "wallpapers",
    "max_pages": 3,
    "concurrency": 10,
    "sources": [
        {
            "name": "WallpapersCraft",
            "url": "https://wallpaperscraft.com/catalog/nature",
            "category": "nature",
            "selector": "img.wallpaper__image",
            "image_attr": "src"
        },
        {
            "name": "HDWallpapers",
            "url": "https://www.hdwallpapers.in/nature-wallpapers.html",
            "category": "nature",
            "selector": ".wall a",
            "image_attr": "href"
        }
    ]
}
"""
import os
import json
import argparse
from pathlib import Path
from typing import Dict, Any
class ConfigCrawler:
    """基于配置的爬虫"""
    def __init__(self, config_path: str):
        """从配置文件初始化"""
        with open(config_path, 'r', encoding='utf-8') as f:
            self.config = json.load(f)
        self.download_dir = Path(self.config.get('download_dir', 'wallpapers'))
        self.download_dir.mkdir(exist_ok=True)
        self.sources = self.config.get('sources', [])
    def load_session(self) -> requests.Session:
        """创建持久化连接会话"""
        session = requests.Session()
        session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept': 'text/html,application/xhtml+xml',
            'Referer': 'https://www.google.com/'
        })
        return session
    def login_if_required(self, session: requests.Session, source: Dict):
        """如果需要，执行登录"""
        login_url = source.get('login_url')
        if login_url:
            # 这里实现登录逻辑
            pass
    def smart_parse(self, html: str, selector: str, attr: str) -> List[str]:
        """智能解析"""
        soup = BeautifulSoup(html, 'html.parser')
        elements = soup.select(selector)
        urls = []
        for el in elements:
            url = el.get(attr)
            if url:
                urls.append(url)
        return urls
    def download_with_progress(self, urls: List[str], category: str):
        """带进度显示的下载"""
        total = len(urls)
        completed = 0
        def update_progress():
            nonlocal completed
            completed += 1
            percent = (completed / total * 100) if total > 0 else 0
            bar = '█' * int(percent // 5) + '░' * (20 - int(percent // 5))
            print(f"\r下载进度: [{bar}] {percent:.1f}% ({completed}/{total})", end='')
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            futures = []
            for url in urls:
                future = executor.submit(self.download_single, url, category)
                future.add_done_callback(lambda f: update_progress())
                futures.append(future)
            for future in concurrent.futures.as_completed(futures):
                if future.exception():
                    logger.error(f"下载失败: {future.exception()}")
            print()  # 新行
# 命令行接口
def main():
    parser = argparse.ArgumentParser(description='高清壁纸爬虫')
    parser.add_argument('-c', '--config', default='config.json', help='配置文件路径')
    parser.add_argument('--debug', action='store_true', help='开启调试模式')
    args = parser.parse_args()
    if args.debug:
        logging.basicConfig(level=logging.DEBUG)
    crawler = ConfigCrawler(args.config)
    # ... 执行爬取
if __name__ == "__main__":
    main()

专用爬虫示例（针对特定网站）

class WallhavenCrawler(WallpaperCrawler):
    """专门针对 wallhaven.cc 的爬虫"""
    def extract_wallpaper_urls(self, html: str, base_url: str) -> List[str]:
        """适配 wallhaven.cc 的解析逻辑"""
        urls = []
        soup = BeautifulSoup(html, 'html.parser')
        # Wallhaven 使用 figure class="thumb"
        for figure in soup.select('figure.thumb'):
            img = figure.find('img')
            if img:
                # 分辨率
                resolution = figure.get('data-resolution', '')
                if resolution:
                    width, height = map(int, resolution.split('x'))
                    if width >= self.config['min_resolution'][0] and \
                       height >= self.config['min_resolution'][1]:
                        # 构建高清原图URL
                        full_url = img.get('data-src') or img.get('src')
                        if full_url:
                            urls.append(full_url)
        return urls
    def download_image(self, img_url: str, category: str = "misc") -> bool:
        """重写下载方法，添加特定处理"""
        # 获取壁纸详情页的真实大图
        detail_url = img_url.replace('wallhaven.cc', 'wallhaven.cc/download')
        return super().download_image(detail_url, category)

安装依赖

pip install requests beautifulsoup4 lxml concurrent-futures

使用说明

1. 替换源网站：由于壁纸网站结构可能会有变化，您需要根据实际情况调整extract_wallpaper_urls方法中的选择器。

2. 注意带宽：高清壁纸可能很大，建议设置合理的per_page和concurrency。

3. 道德准则：

   • 仅用于个人使用
   • 设置合理的下载间隔
   • 遵守网站的robots.txt规则

4. 错误处理：脚本包含重试机制和详细的日志记录。

这个脚本提供了完整的解决方案，您可以根据需要调整适配不同的壁纸网站，如需针对特定网站进行适配，请提供具体网站URL,我可以为您定制解析逻辑。