从Bing图片搜索JSON API直接获取数据

import requests
import json
import time
import hashlib
import os
from urllib.parse import quote, urlencode
import logging
from typing import List, Dict, Optional
import asyncio
import aiohttp
from aiofiles import open as aio_open

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# 代理配置
proxyHost = "www.16yun.cn"
proxyPort = "5445"
proxyUser = "16QMSOML"
proxyPass = "280651"

class BingImageAPICrawler:
    """
    Bing 图片 API 爬虫类
    """
    
    def __init__(self, download_dir: str = "downloaded_images"):
        self.download_dir = download_dir
        self.session = requests.Session()
        
        # 设置请求头，模拟浏览器行为
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            'Accept-Encoding': 'gzip, deflate, br',
            'DNT': '1',
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1',
        }
        
        self.session.headers.update(self.headers)
        
        # 配置代理
        self.proxies = {
            'http': f'http://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}',
            'https': f'https://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}'
        }
        
        # 创建下载目录
        os.makedirs(download_dir, exist_ok=True)
    
    def search_images(self, keyword: str, page_count: int = 3) -> List[Dict]:
        """
        通过 API 搜索图片并返回结构化数据
        
        Args:
            keyword: 搜索关键词
            page_count: 要爬取的页数
            
        Returns:
            图片信息列表
        """
        all_images = []
        
        for page in range(page_count):
            try:
                # 计算偏移量
                offset = page * 35 + 1
                
                # 构造 API 请求参数
                params = {
                    'q': keyword,
                    'first': offset,
                    'count': 35,
                    'mmasync': 1,
                    'qft': '+filterui:photo-photo'  # 只获取照片类型的图片
                }
                
                api_url = f"https://www.bing.com/images/async?{urlencode(params)}"
                logger.info(f"正在获取第 {page + 1} 页数据: {api_url}")
                
                # 使用代理发送请求
                response = self.session.get(api_url, timeout=10, proxies=self.proxies)
                response.raise_for_status()
                
                # 解析返回的 HTML 片段中的图片数据
                page_images = self.parse_image_data(response.text, keyword)
                all_images.extend(page_images)
                
                logger.info(f"第 {page + 1} 页获取到 {len(page_images)} 张图片")
                
                # 礼貌性延迟，避免请求过快
                time.sleep(1)
                
            except requests.RequestException as e:
                logger.error(f"获取第 {page + 1} 页数据失败: {e}")
                continue
            except Exception as e:
                logger.error(f"解析第 {page + 1} 页数据时发生错误: {e}")
                continue
        
        logger.info(f"搜索完成，共获取到 {len(all_images)} 张图片的元数据")
        return all_images
    
    def parse_image_data(self, html_content: str, keyword: str) -> List[Dict]:
        """
        从 API 返回的 HTML 片段中解析图片数据
        
        Args:
            html_content: API 返回的 HTML 内容
            keyword: 搜索关键词
            
        Returns:
            图片信息字典列表
        """
        images = []
        
        # API 返回的是包含图片数据的 HTML 片段
        # 我们需要从中提取包含图片信息的 JSON 数据
        lines = html_content.split('\n')
        
        for line in lines:
            line = line.strip()
            if 'm=' in line and 'murl=' in line:
                try:
                    # 找到 JSON 数据的起始位置
                    start_idx = line.find('m="') + 3
                    end_idx = line.find('"', start_idx)
                    
                    if start_idx > 2 and end_idx > start_idx:
                        json_str = line[start_idx:end_idx]
                        # 处理 HTML 转义字符
                        json_str = json_str.replace('"', '"').replace('&', '&')
                        
                        # 解析 JSON 数据
                        img_data = json.loads(json_str)
                        
                        image_info = {
                            'keyword': keyword,
                            'title': img_data.get('t', ''),
                            'image_url': img_data.get('murl', ''),
                            'thumbnail_url': img_data.get('turl', ''),
                            'source_url': img_data.get('purl', ''),
                            'width': img_data.get('w', 0),
                            'height': img_data.get('h', 0),
                            'file_size': img_data.get('fs', 0),
                            'content_type': img_data.get('ity', 'jpg'),
                            'metadata': img_data
                        }
                        
                        # 验证必要字段
                        if image_info['image_url'] and image_info['image_url'].startswith('http'):
                            images.append(image_info)
                            
                except (json.JSONDecodeError, KeyError, ValueError) as e:
                    logger.debug(f"解析图片数据失败: {e}, 原始数据: {line[:100]}...")
                    continue
        
        return images
    
    def download_image(self, image_info: Dict) -> Optional[str]:
        """
        下载单张图片
        
        Args:
            image_info: 图片信息字典
            
        Returns:
            下载成功的图片路径，失败返回 None
        """
        try:
            image_url = image_info['image_url']
            # 使用代理下载图片
            response = self.session.get(image_url, timeout=15, proxies=self.proxies)
            response.raise_for_status()
            
            # 生成文件名：关键词_URL哈希值.扩展名
            url_hash = hashlib.md5(image_url.encode()).hexdigest()[:8]
            file_ext = image_info.get('content_type', 'jpg')
            if not file_ext or file_ext == 'jpeg':
                file_ext = 'jpg'
            
            filename = f"{image_info['keyword']}_{url_hash}.{file_ext}"
            filepath = os.path.join(self.download_dir, filename)
            
            # 保存图片
            with open(filepath, 'wb') as f:
                f.write(response.content)
            
            logger.info(f"图片下载成功: {filename}")
            return filepath
            
        except Exception as e:
            logger.error(f"下载图片失败 {image_info['image_url']}: {e}")
            return None
    
    async def download_image_async(self, session: aiohttp.ClientSession, image_info: Dict) -> Optional[str]:
        """
        异步下载单张图片
        
        Args:
            session: aiohttp 会话
            image_info: 图片信息字典
            
        Returns:
            下载成功的图片路径，失败返回 None
        """
        try:
            image_url = image_info['image_url']
            async with session.get(image_url) as response:
                if response.status == 200:
                    content = await response.read()
                    
                    # 生成文件名
                    url_hash = hashlib.md5(image_url.encode()).hexdigest()[:8]
                    file_ext = image_info.get('content_type', 'jpg')
                    if not file_ext or file_ext == 'jpeg':
                        file_ext = 'jpg'
                    
                    filename = f"{image_info['keyword']}_{url_hash}.{file_ext}"
                    filepath = os.path.join(self.download_dir, filename)
                    
                    # 异步保存图片
                    async with aio_open(filepath, 'wb') as f:
                        await f.write(content)
                    
                    logger.info(f"图片异步下载成功: {filename}")
                    return filepath
                else:
                    logger.error(f"下载失败，状态码: {response.status}, URL: {image_url}")
                    return None
                    
        except Exception as e:
            logger.error(f"异步下载图片失败 {image_info['image_url']}: {e}")
            return None
    
    async def download_images_async(self, images_info: List[Dict], concurrent_limit: int = 10):
        """
        异步批量下载图片
        
        Args:
            images_info: 图片信息字典列表
            concurrent_limit: 并发下载数量限制
        """
        # 配置代理连接器
        proxy_auth = aiohttp.BasicAuth(proxyUser, proxyPass)
        connector = aiohttp.TCPConnector(
            limit=concurrent_limit,
            proxy=f"http://{proxyHost}:{proxyPort}",
            proxy_auth=proxy_auth
        )
        timeout = aiohttp.ClientTimeout(total=30)
        
        async with aiohttp.ClientSession(
            connector=connector,
            timeout=timeout,
            headers=self.headers
        ) as session:
            
            tasks = []
            for image_info in images_info:
                task = self.download_image_async(session, image_info)
                tasks.append(task)
            
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            success_count = sum(1 for r in results if r is not None and not isinstance(r, Exception))
            logger.info(f"异步下载完成，成功: {success_count}/{len(images_info)}")
            
            return results

def main():
    """主函数示例"""
    # 创建爬虫实例
    crawler = BingImageAPICrawler()
    
    # 搜索关键词
    keyword = "自然风光"
    
    # 获取图片元数据
    logger.info(f"开始搜索关键词: {keyword}")
    images_data = crawler.search_images(keyword, page_count=2)
    
    if not images_data:
        logger.warning("未获取到任何图片数据")
        return
    
    # 保存元数据到 JSON 文件
    metadata_file = f"bing_{keyword}_metadata.json"
    with open(metadata_file, 'w', encoding='utf-8') as f:
        json.dump(images_data, f, ensure_ascii=False, indent=2)
    logger.info(f"图片元数据已保存到: {metadata_file}")
    
    # 方式1：同步下载（适合小批量）
    # for i, image_info in enumerate(images_data[:5]):  # 只下载前5张作为演示
    #     crawler.download_image(image_info)
    
    # 方式2：异步下载（适合大批量，推荐）
    async def async_download():
        await crawler.download_images_async(images_data[:10])  # 只下载前10张作为演示
    
    # 运行异步下载
    asyncio.run(async_download())
    
    logger.info("程序执行完成")

if __name__ == "__main__":
    main()