突破反爬：使用代理IP和User-Agent轮询爬取音乐数据

import requests
import time
import random
from typing import List, Dict, Optional
from fake_useragent import UserAgent
from concurrent.futures import ThreadPoolExecutor, as_completed

class MusicDataCrawler:
    """
    音乐数据爬虫类
    具备代理IP和User-Agent轮询功能
    """
    
    def __init__(self):
        self.session = requests.Session()
        self.ua_generator = UserAgent()
        
        # 设置固定代理信息
        self.proxyHost = "www.16yun.cn"
        self.proxyPort = "5445"
        self.proxyUser = "16QMSOML"
        self.proxyPass = "280651"
        
        # 初始化代理IP池（包含付费代理和免费代理）
        self.proxy_pool = self._init_proxy_pool()
        
        # 请求统计
        self.request_count = 0
        self.success_count = 0
        
        # 爬虫配置
        self.max_retries = 3
        self.timeout = 10
        self.request_delay = (1, 3)  # 请求延迟范围（秒）
    
    def _init_proxy_pool(self) -> List[Dict]:
        """
        初始化代理IP池，包含付费代理和免费代理
        """
        # 构建认证代理
        auth_proxy = self._build_auth_proxy()
        
        # 代理池包含付费代理和免费代理
        proxies = [
            auth_proxy,  # 付费认证代理
            {'http': 'http://103.156.144.121:80', 'https': 'http://103.156.144.121:80'},
            {'http': 'http://45.65.132.180:8080', 'https': 'http://45.65.132.180:8080'},
            # 可以添加更多代理...
        ]
        return proxies
    
    def _build_auth_proxy(self) -> Dict:
        """
        构建带认证的代理配置
        """
        proxy_url = f"http://{self.proxyUser}:{self.proxyPass}@{self.proxyHost}:{self.proxyPort}"
        return {
            'http': proxy_url,
            'https': proxy_url
        }
    
    def _get_auth_proxy(self) -> Dict:
        """
        获取带认证的代理（优先使用）
        """
        return self._build_auth_proxy()
    
    def _get_random_user_agent(self) -> str:
        """
        获取随机User-Agent
        """
        return self.ua_generator.random
    
    def _get_random_proxy(self) -> Optional[Dict]:
        """
        从代理池中随机选择一个代理
        增加权重，让付费代理有更高使用概率
        """
        if not self.proxy_pool:
            return None
        
        # 给付费代理更高权重（60%概率使用付费代理）
        if random.random() < 0.6:
            return self._get_auth_proxy()
        else:
            return random.choice(self.proxy_pool)
    
    def _make_request(self, url: str, params: Dict = None, retry_count: int = 0) -> Optional[requests.Response]:
        """
        执行单次请求，包含代理和User-Agent轮询
        """
        try:
            # 随机延迟，模拟人类行为
            delay = random.uniform(*self.request_delay)
            time.sleep(delay)
            
            # 准备请求头
            headers = {
                'User-Agent': self._get_random_user_agent(),
                'Accept': 'application/json, text/plain, */*',
                'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
                'Accept-Encoding': 'gzip, deflate, br',
                'Connection': 'keep-alive',
            }
            
            # 获取代理
            proxies = self._get_random_proxy()
            
            self.request_count += 1
            print(f"请求 #{self.request_count}: {url}")
            
            # 安全地显示代理信息（隐藏密码）
            if proxies and 'http' in proxies:
                proxy_display = proxies['http'].replace(self.proxyPass, '***')
                print(f"使用代理: {proxy_display}")
            else:
                print(f"使用代理: {proxies}")
                
            print(f"使用User-Agent: {headers['User-Agent'][:50]}...")
            
            response = self.session.get(
                url,
                params=params,
                headers=headers,
                proxies=proxies,
                timeout=self.timeout,
                verify=False  # 注意：这里为了演示关闭了SSL验证，生产环境应谨慎使用
            )
            
            # 检查响应状态
            if response.status_code == 200:
                self.success_count += 1
                print(f"请求成功! 成功率: {self.success_count}/{self.request_count} "
                      f"({self.success_count/self.request_count*100:.1f}%)")
                return response
            elif response.status_code in [403, 429]:
                # 遇到访问限制，可能是代理或User-Agent失效
                print(f"遇到访问限制: {response.status_code}")
                if retry_count < self.max_retries:
                    print(f"进行重试 ({retry_count + 1}/{self.max_retries})")
                    # 重试时更换代理
                    return self._make_request(url, params, retry_count + 1)
                else:
                    print("达到最大重试次数，放弃请求")
                    return None
            else:
                print(f"请求失败，状态码: {response.status_code}")
                return None
                
        except requests.exceptions.RequestException as e:
            print(f"请求异常: {e}")
            if retry_count < self.max_retries:
                print(f"进行重试 ({retry_count + 1}/{self.max_retries})")
                # 重试时更换代理
                return self._make_request(url, params, retry_count + 1)
            return None
    
    def crawl_music_list(self, search_keyword: str, page_count: int = 3) -> List[Dict]:
        """
        爬取音乐列表数据
        注意：这里使用模拟的API端点，实际应用中需要替换为目标网站的API
        """
        music_data = []
        
        for page in range(1, page_count + 1):
            print(f"\n开始爬取第 {page} 页数据...")
            
            # 模拟音乐API请求URL（请替换为实际目标网站的API）
            # 这里使用一个示例URL结构
            api_url = "https://api.example-music-site.com/search"
            params = {
                'keyword': search_keyword,
                'page': page,
                'limit': 20
            }
            
            response = self._make_request(api_url, params)
            
            if response:
                try:
                    # 解析JSON响应
                    data = response.json()
                    
                    # 模拟数据提取（根据实际API响应结构调整）
                    if data.get('success'):
                        songs = data.get('data', {}).get('songs', [])
                        
                        for song in songs:
                            music_info = {
                                'id': song.get('id'),
                                'name': song.get('name'),
                                'artist': song.get('artist', {}).get('name'),
                                'album': song.get('album', {}).get('name'),
                                'duration': song.get('duration'),
                                'play_url': song.get('play_url')
                            }
                            music_data.append(music_info)
                            print(f"获取歌曲: {music_info['name']} - {music_info['artist']}")
                    
                    print(f"第 {page} 页爬取完成，获得 {len(songs)} 首歌曲")
                    
                except ValueError as e:
                    print(f"JSON解析错误: {e}")
            else:
                print(f"第 {page} 页爬取失败")
            
            # 页面间延迟
            time.sleep(random.uniform(2, 5))
        
        return music_data
    
    def batch_crawl(self, keywords: List[str], max_workers: int = 3) -> Dict[str, List[Dict]]:
        """
        批量爬取多个关键词的音乐数据
        """
        results = {}
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            # 提交任务
            future_to_keyword = {
                executor.submit(self.crawl_music_list, keyword, 2): keyword 
                for keyword in keywords
            }
            
            # 收集结果
            for future in as_completed(future_to_keyword):
                keyword = future_to_keyword[future]
                try:
                    results[keyword] = future.result()
                except Exception as e:
                    print(f"关键词 '{keyword}' 爬取失败: {e}")
                    results[keyword] = []
        
        return results

    def get_proxy_status(self) -> Dict:
        """
        获取代理使用状态统计
        """
        return {
            'total_requests': self.request_count,
            'successful_requests': self.success_count,
            'success_rate': self.success_count / self.request_count * 100 if self.request_count > 0 else 0,
            'proxy_count': len(self.proxy_pool),
            'primary_proxy': f"{self.proxyUser}@{self.proxyHost}:{self.proxyPort}"
        }

def main():
    """
    主函数：演示爬虫的使用
    """
    # 初始化爬虫
    crawler = MusicDataCrawler()
    
    # 显示代理状态
    status = crawler.get_proxy_status()
    print(f"代理状态: {status}")
    
    # 单个关键词爬取示例
    print("=== 开始单关键词爬取 ===")
    songs = crawler.crawl_music_list("周杰伦", page_count=2)
    print(f"\n爬取完成! 共获得 {len(songs)} 首歌曲")
    
    # 显示前几首歌曲信息
    for i, song in enumerate(songs[:5]):
        print(f"{i+1}. {song['name']} - {song['artist']}")
    
    # 批量爬取示例
    print("\n=== 开始批量爬取 ===")
    keywords = ["流行", "摇滚", "古典"]
    batch_results = crawler.batch_crawl(keywords)
    
    for keyword, songs in batch_results.items():
        print(f"关键词 '{keyword}': 获得 {len(songs)} 首歌曲")
    
    # 最终统计
    final_status = crawler.get_proxy_status()
    print(f"\n最终统计:")
    print(f"总请求数: {final_status['total_requests']}")
    print(f"成功请求: {final_status['successful_requests']}")
    print(f"成功率: {final_status['success_rate']:.1f}%")

if __name__ == "__main__":
    main()