Python爬虫实战：利用代理IP获取招聘网站信息

import requests
from fake_useragent import UserAgent

class ProxyHandler:
    def __init__(self):
        self.proxy_list = self.load_proxies()  # 从站大爷API获取代理
        self.ua = UserAgent()
        
    def load_proxies(self):
        # 实际开发中替换为你的代理API
        return ["http://user:pass@ip:port", ...]
    
    def get_proxy(self):
        if not self.proxy_list:
            raise Exception("代理池已耗尽")
        return {"http": self.proxy_list.pop(0)}
    
    def get_headers(self):
        return {"User-Agent": self.ua.random}

import time
import random
from bs4 import BeautifulSoup
from proxy_handler import ProxyHandler

class JobSpider:
    def __init__(self):
        self.proxy = ProxyHandler()
        self.base_url = "https://www.example.com/jobs"
        self.keyword = "python"
        self.results = []
        
    def fetch_page(self, page_num):
        url = f"{self.base_url}?page={page_num}&keyword={self.keyword}"
        headers = self.proxy.get_headers()
        proxy = self.proxy.get_proxy()
        
        try:
            response = requests.get(
                url, 
                headers=headers, 
                proxies=proxy,
                timeout=10
            )
            response.raise_for_status()
            return response.text
        except Exception as e:
            print(f"请求失败: {e}")
            # 失败后自动更换代理重试
            return self.fetch_page(page_num)
    
    def parse_job(self, html):
        soup = BeautifulSoup(html, 'html.parser')
        jobs = soup.select('.job-card')  # 根据实际页面调整选择器
        
        for job in jobs:
            data = {
                "title": job.select_one('.job-title').text.strip(),
                "company": job.select_one('.company-name').text.strip(),
                "salary": job.select_one('.salary').text.strip(),
                "location": job.select_one('.job-location').text.strip()
            }
            self.results.append(data)
    
    def run(self, max_pages=5):
        for page in range(1, max_pages + 1):
            print(f"正在采集第{page}页...")
            html = self.fetch_page(page)
            self.parse_job(html)
            
            # 随机延迟1-3秒
            time.sleep(random.uniform(1, 3))
            
            # 每3页更换一次User-Agent
            if page % 3 == 0:
                self.proxy.ua = UserAgent()

if __name__ == "__main__":
    spider = JobSpider()
    spider.run(max_pages=10)
    
    # 保存结果到CSV
    import pandas as pd
    df = pd.DataFrame(spider.results)
    df.to_csv("jobs_data.csv", index=False, encoding='utf-8-sig')

from selenium import webdriver
from selenium.webdriver.chrome.options import Options

def selenium_crawl(url):
    options = Options()
    options.add_argument('--headless')
    options.add_argument(f'user-agent={proxy.ua.random}')
    
    driver = webdriver.Chrome(options=options)
    driver.get(url)
    
    # 模拟鼠标移动
    from selenium.webdriver.common.action_chains import ActionChains
    actions = ActionChains(driver)
    actions.move_by_offset(100, 100).perform()
    
    html = driver.page_source
    driver.quit()
    return html

class SmartProxyPool:
    def __init__(self):
        self.primary_proxies = [...]  # 优质代理
        self.backup_proxies = [...]   # 备用代理
        
    def get_proxy(self):
        if self.primary_proxies:
            return {"http": self.primary_proxies.pop(0)}
        elif self.backup_proxies:
            print("警告：使用备用代理")
            return {"http": self.backup_proxies.pop(0)}
        else:
            raise Exception("所有代理已耗尽")

import sqlite3

def save_to_db(data):
    conn = sqlite3.connect('jobs.db')
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS jobs (
            title TEXT,
            company TEXT,
            salary TEXT,
            location TEXT,
            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    
    for job in data:
        cursor.execute('''
            INSERT INTO jobs VALUES (?,?,?,?)
        ''', (job['title'], job['company'], job['salary'], job['location']))
    
    conn.commit()
    conn.close()

def clean_salary(salary_str):
    try:
        if "k" in salary_str.lower():
            return float(salary_str.replace("k", "").replace("K", "")) * 1000
        return float(salary_str)
    except:
        return None

# 应用清洗
df['salary_min'] = df['salary'].apply(
    lambda x: clean_salary(x.split('-')[0]) if '-' in str(x) else clean_salary(x)
)

def test_proxy(proxy):
    test_url = "https://httpbin.org/ip"
    try:
        response = requests.get(test_url, proxies=proxy, timeout=5)
        return "origin" in response.json()
    except:
        return False

Q3：招聘网站参数加密如何处理？