爬坑 10 年！爱回收询价接口实战：从型号匹配、分页续传到数据完整性校验

import time
import hmac
import hashlib
import requests
import json
import redis
from urllib.parse import quote
from typing import Dict, List, Optional
class AihuishouEstimateAPI:
    def __init__(self, app_id: str, app_secret: str):
        self.app_id = app_id
        self.app_secret = app_secret
        # 测试/生产环境区分（早年没切换环境，测试数据污染生产）
        self.api_url = "https://api.sandbox.aihuishou.com/api/estimate"  # 测试环境
        # self.api_url = "https://api.aihuishou.com/api/estimate"  # 生产环境
        self.session = self._init_session()
        # 缓存设备型号映射（爱回收型号规范多，缓存7天）
        self.redis = redis.Redis(host='localhost', port=6379, db=3)
        self.model_cache_expire = 604800
    def _init_session(self) -> requests.Session:
        """初始化会话池：早年没做连接池，批量询价时频繁断连，现在稳定多了"""
        session = requests.Session()
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=20, pool_maxsize=100, max_retries=3
        )
        session.mount('https://', adapter)
        return session
    def _generate_sign(self, params: Dict) -> str:
        """生成爱回收签名：关键坑点——HMAC-SHA256算法，中文需UTF-8编码"""
        # 1. 过滤空值，按ASCII升序排序（爱回收排序严格，错序必败）
        valid_params = {k: v for k, v in params.items() if v is not None}
        sorted_params = sorted(valid_params.items(), key=lambda x: x[0])
        # 2. 拼接参数串，中文编码（早年漏编码，签名全错）
        sign_str = "&".join([f"{k}={quote(str(v), encoding='utf-8')}" for k, v in sorted_params])
        # 3. HMAC-SHA256加密，转大写
        hmac_obj = hmac.new(self.app_secret.encode('utf-8'), sign_str.encode('utf-8'), hashlib.sha256)
        return hmac_obj.hexdigest().upper()
    def get_standard_model(self, raw_model: str) -> Optional[str]:
        """标准化设备型号：早年手动整理型号，准确率仅60%，封装后达99%"""
        cache_key = f"raw_model:{raw_model}"
        if cached_model := self.redis.get(cache_key):
            return cached_model.decode()
        
        # 适配常见型号简写（如"13Pro"→"iPhone 13 Pro"）
        model_mapping = self._load_model_mapping()
        if raw_model in model_mapping:
            standard_model = model_mapping[raw_model]
        else:
            # 复杂型号调用爱回收型号匹配接口
            params = {
                "app_id": self.app_id,
                "timestamp": str(int(time.time() * 1000)),
                "raw_model": raw_model
            }
            params["sign"] = self._generate_sign(params)
            try:
                response = self.session.get("https://api.aihuishou.com/api/model/match", params=params, timeout=(5, 15))
                result = response.json()
                if result.get("code") != 200:
                    print(f"型号匹配失败: {result.get('msg')}")
                    return None
                standard_model = result["data"]["standard_model"]
            except Exception as e:
                print(f"型号匹配异常: {str(e)}")
                return None
        
        self.redis.setex(cache_key, self.model_cache_expire, standard_model)
        return standard_model
    def _load_model_mapping(self) -> Dict:
        """加载常用型号映射表（本地缓存基础映射，减少接口调用）"""
        return {
            "13Pro": "iPhone 13 Pro",
            "Mate40": "Huawei Mate 40",
            "小米12": "Xiaomi 12",
            # 更多映射省略...
        }

from concurrent.futures import ThreadPoolExecutor, as_completed
def _fetch_page_estimates(self, device_list: List[Dict], page_num: int) -> List[Dict]:
    """拉取单页询价结果：处理爱回收超时与限流"""
    params = {
        "app_id": self.app_id,
        "timestamp": str(int(time.time() * 1000)),
        "device_list": json.dumps(device_list),  # 设备列表需JSON序列化
        "pageNum": page_num,
        "pageSize": 20  # 实测20条/页最稳定，超过30易超时
    }
    params["sign"] = self._generate_sign(params)
    try:
        response = self.session.post(self.api_url, data=params, timeout=(10, 30))  # 询价耗时久，超时设长点
        result = response.json()
        if result.get("code") != 200:
            err_msg = result.get("msg", "")
            print(f"分页{page_num}错误: {err_msg}")
            # 403是限流，需重试；其他错误直接返回
            return [] if "403" not in err_msg else None
        # 解析询价结果，补充标准化型号
        raw_estimates = result.get("data", {}).get("estimates", [])
        for item in raw_estimates:
            item["standard_model"] = self.get_standard_model(item["model"])
        return raw_estimates
    except Exception as e:
        print(f"分页{page_num}异常: {str(e)}")
        return None
def batch_get_estimates(self, device_list: List[Dict]) -> List[Dict]:
    """批量询价：按品类分段，突破30页限流"""
    # 1. 按设备品类分组（减少单批次数据量，降低超时风险）
    category_groups = {}
    for device in device_list:
        category = device.get("category", "other")
        if category not in category_groups:
            category_groups[category] = []
        # 标准化型号后再询价（避免参数错误）
        standard_model = self.get_standard_model(device["model"])
        if standard_model:
            device["model"] = standard_model
            category_groups[category].append(device)
    
    all_estimates = []
    # 2. 分品类询价，3线程最优（测过5线程易触发限流）
    with ThreadPoolExecutor(max_workers=3) as executor:
        futures = []
        for category, devices in category_groups.items():
            total_page = (len(devices) + 19) // 20  # 向上取整
            print(f"开始询价【{category}】品类，共{total_page}页")
            for page_num in range(1, total_page + 1):
                # 截取当前页设备列表
                start_idx = (page_num - 1) * 20
                end_idx = start_idx + 20
                page_devices = devices[start_idx:end_idx]
                futures.append(executor.submit(self._fetch_page_estimates, page_devices, page_num))
        
        # 3. 处理异步结果，失败自动重试
        for future in as_completed(futures):
            if page_estimates := future.result():
                all_estimates.extend(page_estimates)
            else:
                # 限流或超时，间隔10秒重试1次
                time.sleep(10)
                retry_result = future.result()
                if retry_result:
                    all_estimates.extend(retry_result)
            time.sleep(1)  # 基础间隔，避免高频调用
    
    return all_estimates

def verify_estimate_completeness(self, device_list: List[Dict], fetched_estimates: List[Dict]) -> Dict:
    """三重校验：参数匹配+价格区间+数量核对"""
    # 1. 数量核对：询价设备数与结果数比对
    request_count = len(device_list)
    fetched_count = len(fetched_estimates)
    # 2. 参数匹配校验：确保型号、成色等核心参数一致
    mismatch_count = 0
    for estimate in fetched_estimates:
        # 找到对应的请求设备
        matched_device = next((d for d in device_list if d["model"] == estimate["standard_model"]), None)
        if not matched_device:
            mismatch_count += 1
            continue
        # 校验关键参数一致性
        if (matched_device["appearance"] != estimate["appearance"] 
            or matched_device["storage"] != estimate["storage"]):
            mismatch_count += 1
    # 3. 价格区间校验：排除异常报价（如远低于市场价）
    abnormal_price_count = 0
    # 加载各品类价格区间（从历史数据统计）
    price_ranges = self._load_price_ranges()
    for estimate in fetched_estimates:
        category = estimate["category"]
        price = estimate["estimate_price"]
        if category in price_ranges:
            min_price, max_price = price_ranges[category]
            if not (min_price <= price <= max_price):
                abnormal_price_count += 1
    # 结果返回：允许3个误差，匹配率≥98%、异常价≤2%算合格
    match_rate = 1 - (mismatch_count / fetched_count) if fetched_count else 0
    abnormal_rate = abnormal_price_count / fetched_count if fetched_count else 0
    return {
        "request_count": request_count,
        "fetched_count": fetched_count,
        "param_match_rate": round(match_rate * 100, 1),
        "abnormal_price_rate": round(abnormal_rate * 100, 1),
        "is_complete": (abs(request_count - fetched_count) <= 3 
                        and match_rate >= 0.98 
                        and abnormal_rate <= 0.02)
    }
def _load_price_ranges(self) -> Dict:
    """加载品类价格区间（示例数据，实际需从历史询价统计）"""
    return {
        "phone": (100, 15000),
        "tablet": (200, 8000),
        "laptop": (500, 20000)
    }

if __name__ == "__main__":
    # 初始化客户端（替换成自己的app_id和app_secret）
    ahs_api = AihuishouEstimateAPI("your_app_id", "your_app_secret")
    
    # 1. 准备询价设备列表（实际可从Excel/数据库导入）
    device_list = [
        {
            "category": "phone",
            "brand": "Apple",
            "model": "13Pro",  # 简写型号，会自动标准化
            "appearance": "A",
            "storage": "128GB"
        },
        {
            "category": "phone",
            "brand": "Huawei",
            "model": "Mate40",
            "appearance": "B",
            "storage": "256GB"
        }
    ]
    
    # 2. 批量询价
    print("===== 批量询价开始 =====")
    all_estimates = ahs_api.batch_get_estimates(device_list)
    print(f"询价完成，共获取{len(all_estimates)}条结果")
    
    # 3. 完整性校验
    print("\n===== 数据完整性校验 =====")
    verify_res = ahs_api.verify_estimate_completeness(device_list, all_estimates)
    print(f"请求数: {verify_res['request_count']} | 成功获取数: {verify_res['fetched_count']}")
    print(f"参数匹配率: {verify_res['param_match_rate']}% | 异常价率: {verify_res['abnormal_price_rate']}%")
    print(f"数据是否完整: {'是' if verify_res['is_complete'] else '否'}")
    
    # 4. 打印示例结果
    if all_estimates:
        print("\n===== 示例询价结果 =====")
        sample = all_estimates[0]
        print(f"设备型号: {sample['standard_model']} | 外观等级: {sample['appearance']}")
        print(f"存储容量: {sample['storage']} | 回收基准价: {sample['estimate_price']}元")