Python爬虫在电商价格监控与竞品分析中的应用

import asyncio
import logging
import random
from typing import Dict, Any, Optional
import httpx

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

class RobustEcomCollector:
    # 具备动态退避与反爬状态检测的电商异步采集器
    
    def __init__(self, proxy_pool_url: Optional[str] = None):
        self.proxy_pool_url = proxy_pool_url
        self.headers_pool = [
            {
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36",
                "Accept": "application/json, text/plain, */*",
                "Accept-Language": "zh-CN,zh;q=0.9",
            },
            {
                "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
                "Accept": "application/json, text/plain, */*",
                "Accept-Language": "zh-CN,zh;q=0.8,en;q=0.6",
            }
        ]

    async def fetch_sku_detail(self, item_id: str, max_retries: int = 3) -> Optional[Dict[str, Any]]:
        # 请求指定商品的详细定价与促销数据，集成指数退避重试
        api_url = f"https://api.ecom-mock-target.com/detail/v2?sku_id={item_id}"
        
        for attempt in range(1, max_retries + 1):
            headers = random.choice(self.headers_pool)
            # 真实生产中通过代理池获取高匿名隧道代理
            proxy_mount = {"http://": "http://127.0.0.1:7890", "https://": "http://127.0.0.1:7890"} if self.proxy_pool_url else None
            
            try:
                async with httpx.AsyncClient(proxies=proxy_mount, timeout=8.0, follow_redirects=True) as client:
                    response = await client.get(api_url, headers=headers)
                    
                    # 1. 验证响应状态
                    if response.status_code == 200:
                        payload = response.json()
                        # 2. 检查应用层风控特征
                        if payload.get("code") == 0 and "data" in payload:
                            return payload["data"]
                        elif payload.get("code") in [403, 10002]:
                            logging.warning(f"触发反爬滑块拦截 (ID: {item_id})，准备重试")
                        else:
                            logging.error(f"业务接口返回异常: {payload}")
                    elif response.status_code in [403, 429]:
                        logging.warning(f"HTTP状态码 {response.status_code} 触发频控熔断")
                        
            except (httpx.RequestError, httpx.TimeoutException) as err:
                logging.warning(f"请求失败 [尝试 {attempt}/{max_retries}]: {err}")
                
            # 指数退避抖动等待: 2^(attempt-1) + 随机浮动
            backoff_delay = (2 ** (attempt - 1)) + random.uniform(0.5, 1.5)
            await asyncio.sleep(backoff_delay)
            
        logging.error(f"商品 {item_id} 经过 {max_retries} 次重试全部失败")
        return None

from dataclasses import dataclass, field
from typing import List, Dict, Any

@dataclass
class CouponRule:
    coupon_id: str
    coupon_type: str  # 'SHOP_DISCOUNT', 'PLATFORM_REDUCE', 'SUBSIDY'
    threshold: float  # 使用门槛 (0表示无门槛)
    discount: float   # 减免金额
    is_stackable: bool = True  # 是否可叠加

@dataclass
class SkuPriceCalculationResult:
    sku_id: str
    sku_name: str
    original_price: float
    promotion_base_price: float
    final_net_price: float
    total_savings: float
    applied_coupons: List[str] = field(default_factory=list)

class EcomPricingEngine:
    # 电商促销叠加规则与到手价演算引擎

    @staticmethod
    def compute_net_price(sku_data: Dict[str, Any]) -> SkuPriceCalculationResult:
        sku_id = sku_data["sku_id"]
        sku_name = sku_data.get("spec_name", "默认规格")
        original_price = float(sku_data.get("original_price", 0.0))
        # 1. 单品立减后的活动基准价
        direct_discount = float(sku_data.get("direct_discount", 0.0))
        current_price = max(0.0, original_price - direct_discount)
        
        # 2. 提取并按优先级排序优惠券 (店铺券 -> 品类券 -> 跨店津贴)
        raw_coupons = sku_data.get("available_coupons", [])
        coupons = [
            CouponRule(
                coupon_id=c["id"],
                coupon_type=c["type"],
                threshold=float(c.get("threshold", 0)),
                discount=float(c.get("discount", 0)),
                is_stackable=c.get("stackable", True)
            )
            for c in raw_coupons
        ]
        
        # 3. 模拟最优叠加演算
        applied = []
        # A. 店铺券处理
        shop_coupons = [c for c in coupons if c.coupon_type == 'SHOP_DISCOUNT' and current_price >= c.threshold]
        if shop_coupons:
            best_shop_coupon = max(shop_coupons, key=lambda x: x.discount)
            current_price -= best_shop_coupon.discount
            applied.append(f"店铺券:-¥{best_shop_coupon.discount}")
            
        # B. 平台补贴与跨店券处理
        platform_coupons = [c for c in coupons if c.coupon_type in ['PLATFORM_REDUCE', 'SUBSIDY'] and current_price >= c.threshold]
        if platform_coupons:
            best_plat_coupon = max(platform_coupons, key=lambda x: x.discount)
            current_price -= best_plat_coupon.discount
            applied.append(f"平台券:-¥{best_plat_coupon.discount}")

        final_net_price = round(max(0.01, current_price), 2)
        total_savings = round(original_price - final_net_price, 2)

        return SkuPriceCalculationResult(
            sku_id=sku_id,
            sku_name=sku_name,
            original_price=original_price,
            promotion_base_price=round(original_price - direct_discount, 2),
            final_net_price=final_net_price,
            total_savings=total_savings,
            applied_coupons=applied
        )

# 示例验证执行
sample_sku = {
    "sku_id": "SKU_889102",
    "spec_name": "沙漠金 256GB 全网通",
    "original_price": 8999.0,
    "direct_discount": 300.0,  # 划线降300
    "available_coupons": [
        {"id": "CP_01", "type": "SHOP_DISCOUNT", "threshold": 5000.0, "discount": 400.0, "stackable": True},
        {"id": "CP_02", "type": "PLATFORM_REDUCE", "threshold": 8000.0, "discount": 500.0, "stackable": True}
    ]
}

calc_res = EcomPricingEngine.compute_net_price(sample_sku)
# 演算结果: 标价8999 -> 直降300 -> 店铺券400 -> 平台券500 -> 最终到手价: 7799.0

import re
from typing import Dict, Tuple, Optional

class ProductMatcher:
    # 跨平台竞品同款属性提取与相似度评分器

    BRAND_MAP = {
        "apple": "BR_APPLE",
        "苹果": "BR_APPLE",
        "huawei": "BR_HUAWEI",
        "华为": "BR_HUAWEI",
        "xiaomi": "BR_XIAOMI",
        "小米": "BR_XIAOMI"
    }

    STOP_WORDS = [r"【.*?】", r"\[.*?\]", r"现货速发", r"官方正品", r"百亿补贴", r"限时立减", r"全网通5G", r"包邮"]

    @classmethod
    def clean_title(cls, title: str) -> str:
        # 剔除营销停用词与多余标点
        cleaned = title
        for pattern in cls.STOP_WORDS:
            cleaned = re.sub(pattern, " ", cleaned, flags=re.IGNORECASE)
        return " ".join(cleaned.split())

    @classmethod
    def extract_brand(cls, text: str) -> Optional[str]:
        for kw, brand_id in cls.BRAND_MAP.items():
            if re.search(rf"{kw}", text, re.IGNORECASE) or kw in text:
                return brand_id
        return None

    @classmethod
    def extract_specs(cls, text: str) -> Dict[str, str]:
        # 正则抽取容量与颜色规格
        specs = {}
        # 提取存储容量: 如 128G, 256GB, 1TB
        cap_match = re.search(r"(\d+(?:GB|G|TB))", text, re.IGNORECASE)
        if cap_match:
            cap = cap_match.group(1).upper()
            if cap.endswith("G") and not cap.endswith("GB"):
                cap = cap + "B"
            specs["capacity"] = cap
            
        # 提取颜色特征
        colors = ["沙漠金", "钛原色", "黑色钛金属", "白色", "暗夜黑"]
        for c in colors:
            if c in text:
                specs["color"] = c
                break
        return specs

    @classmethod
    def compute_match_score(cls, title_a: str, title_b: str) -> Tuple[float, bool]:
        # 计算两款商品是否为同款，返回 (置信得分, 是否判定为同款)
        clean_a = cls.clean_title(title_a)
        clean_b = cls.clean_title(title_b)

        # 1. 品牌强校验 (权重 0.3)
        brand_a = cls.extract_brand(clean_a)
        brand_b = cls.extract_brand(clean_b)
        if not brand_a or not brand_b or brand_a != brand_b:
            return 0.0, False
        brand_score = 1.0

        # 2. 型号匹配 (权重 0.4)
        model_pattern = r"iPhone\s*16\s*Pro"
        has_model_a = bool(re.search(model_pattern, clean_a, re.IGNORECASE))
        has_model_b = bool(re.search(model_pattern, clean_b, re.IGNORECASE))
        model_score = 1.0 if (has_model_a and has_model_b) else 0.0
        if model_score == 0.0:
            return 0.3 * brand_score, False

        # 3. 规格参数强校验 (权重 0.3)
        specs_a = cls.extract_specs(clean_a)
        specs_b = cls.extract_specs(clean_b)
        
        spec_matches = 0
        total_spec_keys = max(len(specs_a), 1)
        for k, v in specs_a.items():
            if specs_b.get(k) == v:
                spec_matches += 1
        spec_score = spec_matches / total_spec_keys

        # 加权置信度
        total_score = round(0.3 * brand_score + 0.4 * model_score + 0.3 * spec_score, 3)
        is_matched = total_score >= 0.85
        return total_score, is_matched

# 测试样例验证
t1 = "【官方现货】Apple iPhone 16 Pro 256GB 沙漠金 全网通5G手机"
t2 = "【百亿补贴/立减500】苹果 iPhone16Pro 256G 沙漠金 现货速发"
score, matched = ProductMatcher.compute_match_score(t1, t2)
# 输出: score: 1.0, matched: True (判定为同款竞品)

-- 1. 标准商品库与映射对齐表
CREATE TABLE IF NOT EXISTS standard_product_mapping (
    id SERIAL PRIMARY KEY,
    standard_sku_id VARCHAR(64) NOT NULL,
    channel_name VARCHAR(32) NOT NULL,      -- 'JD', 'TMALL', 'PDD'
    channel_sku_id VARCHAR(64) NOT NULL,
    product_title VARCHAR(255) NOT NULL,
    brand_code VARCHAR(32) NOT NULL,
    spec_attributes JSONB,                 -- {"capacity": "256GB", "color": "沙漠金"}
    match_confidence NUMERIC(4, 3),        -- 0.985
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    UNIQUE(channel_name, channel_sku_id)
);

-- 2. 价格时序快照流水表
CREATE TABLE IF NOT EXISTS sku_price_timeseries (
    time TIMESTAMP WITH TIME ZONE NOT NULL,
    channel_sku_id VARCHAR(64) NOT NULL,
    original_price NUMERIC(10, 2) NOT NULL,
    promotion_base_price NUMERIC(10, 2) NOT NULL,
    final_net_price NUMERIC(10, 2) NOT NULL,
    stock_status INT DEFAULT 1,             -- 1: 有货, 0: 无货
    applied_coupon_summary TEXT,
    PRIMARY KEY(time, channel_sku_id)
);

-- 3. 价格异动事件与告警记录表
CREATE TABLE IF NOT EXISTS price_alert_events (
    event_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    channel_sku_id VARCHAR(64) NOT NULL,
    alert_level VARCHAR(8) NOT NULL,        -- 'P0', 'P1', 'P2', 'P3'
    previous_price NUMERIC(10, 2) NOT NULL,
    current_price NUMERIC(10, 2) NOT NULL,
    price_change_rate NUMERIC(5, 4) NOT NULL, -- -0.1520
    is_handled BOOLEAN DEFAULT FALSE,
    notified_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);