云函数采集架构：Serverless模式下的动态IP与冷启动优化

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import requests
import json
import time

# ------------------------------
# 配置部分：代理、请求头等参数
# ------------------------------

# 代理IP配置（参考亿牛云爬虫代理信息 www.16yun.cn）
# 注意：以下信息仅为示例，请根据实际账号信息填写
PROXY_HOST = "proxy.16yun.cn"    # 亿牛云爬虫代理域名
PROXY_PORT = "8000"                   # 代理端口
PROXY_USER = "16YUN"          # 代理用户名
PROXY_PASS = "16IP"          # 代理密码

# 拼接代理认证 URL
proxy_auth = f"http://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}"
proxies = {
    "http": proxy_auth,
    "https": proxy_auth,
}

# 设置请求头：UserAgent 和 Cookie（这里的 Cookie 示例仅供参考）
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
                  'AppleWebKit/537.36 (KHTML, like Gecko) '
                  'Chrome/108.0.0.0 Safari/537.36',
    'Cookie': 'appver=2.0.2; os=pc; osver=Microsoft-Windows-10-Professional-build-19044-64bit;',
}

# 云函数预热策略：在入口函数调用前可以先执行一次请求，减少冷启动时间（示例）
def prewarm():
    try:
        # 简单发起一个 GET 请求用于预热
        requests.get("https://music.163.com", headers=headers, proxies=proxies, timeout=5)
        print("预热成功!")
    except Exception as e:
        print("预热异常:", e)

# ------------------------------
# 功能函数
# ------------------------------

def search_music(keyword, limit=30):
    """
    根据关键词搜索歌曲信息
    :param keyword: 搜索关键词
    :param limit: 返回结果数量限制
    :return: 搜索结果的 JSON 数据
    """
    url = "https://music.163.com/api/search/get/"
    # 构建搜索参数
    data = {
        "s": keyword,
        "type": 1,      # 1 代表单曲搜索
        "offset": 0,
        "total": "true",
        "limit": limit,
    }
    try:
        response = requests.post(url, headers=headers, data=data, proxies=proxies, timeout=10)
        response.encoding = 'utf-8'
        if response.status_code == 200:
            print("搜索请求成功!")
            return response.json()
        else:
            print("搜索请求失败，状态码:", response.status_code)
    except Exception as e:
        print("搜索请求异常:", e)
    return None

def get_lyric(song_id):
    """
    获取指定歌曲的歌词信息
    :param song_id: 歌曲ID
    :return: 歌词数据（JSON 格式）
    """
    url = f"https://music.163.com/api/song/lyric?os=pc&id={song_id}&lv=1&kv=1&tv=-1"
    try:
        response = requests.get(url, headers=headers, proxies=proxies, timeout=10)
        if response.status_code == 200:
            print(f"歌曲 {song_id} 的歌词请求成功!")
            return response.json()
        else:
            print(f"歌词请求失败，状态码: {response.status_code}")
    except Exception as e:
        print("歌词请求异常:", e)
    return None

def get_comments(song_id, limit=30):
    """
    获取指定歌曲的评论信息
    :param song_id: 歌曲ID
    :param limit: 返回评论数量限制
    :return: 评论数据（JSON 格式）
    """
    # 网易云音乐评论接口示例：R_SO_4_加歌曲ID
    url = f"https://music.163.com/api/v1/resource/comments/R_SO_4_{song_id}?limit={limit}"
    try:
        response = requests.get(url, headers=headers, proxies=proxies, timeout=10)
        if response.status_code == 200:
            print(f"歌曲 {song_id} 的评论请求成功!")
            return response.json()
        else:
            print(f"评论请求失败，状态码: {response.status_code}")
    except Exception as e:
        print("评论请求异常:", e)
    return None

# ------------------------------
# 云函数入口函数（示例）
# ------------------------------

def main_handler(event, context):
    """
    云函数入口
    :param event: 事件数据，包含搜索关键词等信息
    :param context: 云函数上下文信息
    :return: 任务执行结果
    """
    # 预热云函数，降低冷启动影响
    prewarm()
    
    # 从事件中获取关键词，若未传入则默认"流行"
    keyword = event.get("keyword", "流行")
    print("开始搜索关键词:", keyword)
    
    # 调用搜索接口获取歌曲数据
    search_result = search_music(keyword)
    if not search_result:
        return {"status": "error", "message": "搜索失败"}
    
    songs = search_result.get("result", {}).get("songs", [])
    result_data = []
    
    for song in songs:
        song_id = song.get("id")
        song_name = song.get("name")
        singer_list = [artist.get("name") for artist in song.get("artists", [])]
        
        print(f"处理歌曲：{song_name} (ID: {song_id})")
        # 获取歌词
        lyric_data = get_lyric(song_id)
        # 获取评论
        comment_data = get_comments(song_id)
        
        result_data.append({
            "song_id": song_id,
            "song_name": song_name,
            "singers": singer_list,
            "lyrics": lyric_data.get("lrc", {}).get("lyric") if lyric_data else "",
            "comments": comment_data.get("comments", []) if comment_data else []
        })
        # 控制请求频率，避免过快触发反爬策略
        time.sleep(1)
    
    # 最终返回采集的数据
    return {
        "status": "success",
        "data": result_data
    }

# ------------------------------
# 本地调试
# ------------------------------
if __name__ == "__main__":
    # 模拟云函数传入的事件数据
    test_event = {"keyword": "周杰伦"}
    result = main_handler(test_event, None)
    print(json.dumps(result, ensure_ascii=False, indent=4))