4.3 行为专线：推荐系统数据采集

4.3 行为专线：推荐系统数据采集

本节摘要：推荐系统吃的是行为序列——用户、物品、动作、时间戳四元组按时间排成的轨迹。本节讲从公开页面痕迹拼装行为数据的两条合法路径（榜单时间序列与评分聚合）、序列拼接的实现，以及行为数据特有的隐私脱敏与聚合化纪律。

行为数据的形状

推荐模型与 NLP、CV 模型吃的数据形状不同：不是一个文本或一张图，而是一条按时间排好的轨迹。最小四元组是用户、物品、动作、时间戳——"谁、对什么、做了什么、什么时候"。轨迹上的动作类型自带层级：浏览是弱信号、收藏与加购是中信号、成交是强信号，推荐侧的训练目标就架在这些信号强度上。

行为的采集天然带着红线：真实用户的行为日志属于平台与用户，绕过权限去抓等于闯入他人系统，1.5 节的三层边界在这里最窄。合法路径有两条：公开榜单的时间序列（热销榜、趋势榜按小时或天采样，得到"集体行为"的轨迹）与公开评分与评论的聚合（评分分布、评论量随时间的变化）。两条路径拿不到个体粒度，但推荐系统的冷启动、热度建模、品类趋势分析，用聚合粒度已经能做很多事。

路径一：榜单时间序列

榜单是集体行为的公开投影。每隔固定周期抓一次榜单页，叠加成时间序列：

import asyncio, json
from datetime import datetime
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, CacheMode
from crawl4ai import JsonCssExtractionStrategy

rank_schema = {
    "name": "热销榜条目",
    "baseSelector": "li.rank-item",
    "fields": [
        {"name": "rank",  "selector": "span.no",   "type": "text"},
        {"name": "sku",   "selector": "a",          "type": "attribute", "attribute": "data-sku"},
        {"name": "title", "selector": "span.title", "type": "text"},
    ],
}

async def snapshot_board(url: str) -> dict:
    """单次榜单快照：时间戳加全量名次"""
    cfg = CrawlerRunConfig(cache_mode=CacheMode.BYPASS,
                           extraction_strategy=JsonCssExtractionStrategy(schema=rank_schema))
    async with AsyncWebCrawler() as crawler:
        r = await crawler.arun(url, config=cfg)
        rows = json.loads(r.extracted_content or "[]")
        return {"captured_at": datetime.now().isoformat(timespec="seconds"),
                "board": rows}

snap = asyncio.run(snapshot_board("https://mall.example.com/rank"))
print(snap["captured_at"], "条目数:", len(snap["board"]))
# 输出：2026-08-26T15:00:12 条目数: 50
print(snap["board"][0])    # 输出：{'rank': '1', 'sku': '99012', 'title': '便携风扇 Pro'}
​

快照本身只是原料，价值在叠加后的序列分析：名次变动的速率是短期热度信号，榜单存续时长是长尾判断依据，新品入榜频率反映品类活跃度。快照按 3.4 节的日期分区落库，分析直接在分区上滑窗。

路径二：序列拼接与隐私聚合

评分与评论页给出的是"多少人、什么分布"，加上时间轴就是另一条行为轨迹。拼接时把个体字段即刻聚合，聚合数据本身就是脱敏：

import hashlib
from collections import defaultdict
from datetime import datetime

def anonymize_user(raw_id: str, salt: str = "2026q3") -> str:
    """个体标识先盐化哈希再截断：可区分不可还原"""
    return hashlib.sha256((salt + raw_id).encode()).hexdigest()[:10]

def aggregate_reviews(reviews: list[dict]) -> dict:
    """评论流聚合：只保留统计量，个体不留痕"""
    by_star = defaultdict(int)
    by_day = defaultdict(int)
    for rv in reviews:
        by_star[rv["rating"]] += 1
        day = rv["created_at"][:10]
        by_day[day] += 1
    return {"total": len(reviews),
            "star_dist": dict(sorted(by_star.items())),
            "daily_volume": dict(sorted(by_day.items()))}

reviews = [
    {"user": "buyer_8829", "rating": 5, "created_at": "2026-08-24 21:30", "text": "..."},
    {"user": "buyer_1207", "rating": 4, "created_at": "2026-08-25 09:12", "text": "..."},
    {"user": "buyer_8829", "rating": 5, "created_at": "2026-08-25 22:05", "text": "..."},
]
print(anonymize_user(reviews[0]["user"]))    # 输出：3b1f09c2a7（同一用户哈希稳定，可连不还原）
print(aggregate_reviews(reviews))
# 输出：{'total': 3, 'star_dist': {4: 1, 5: 2}, 'daily_volume': {'2026-08-24': 1, '2026-08-25': 2}}
​

两段函数体现了行为数据的两条纪律。哈希带盐：无盐哈希能被彩虹表反查，加盐后同一用户仍然可关联（序列分析需要）、身份不可还原（隐私要求满足）。聚合优先：分析目标是分布与趋势时，个体记录在管线里存在的时间越短越好——聚合结果直接落库，明细不落。

k 值下限是聚合纪律的量化版：任何对外提供或用于分析的统计单元，覆盖的原始个体数不低于 k（常用 5 或 10）。某天某星级只有一个评论者，这个统计格就该并入相邻格或置为缺失，否则聚合等于变相点名。

行为专线的案例：新品热度预警

一条完整链路走一遍。背景：品类运营需要在竞品出爆款的七十二小时内感知，人工盯屏不可持续。操作：对三个竞品的畅销榜与新品榜每小时采样（榜单页静态、robots 允许，采集成本极低），快照进分区仓储；分析任务计算"名次上升速度"与"评论量日环比"两个信号，任一超阈值即告警。结果：首月捕获十一次新品起量事件，中位感知时间三十一小时，比人工盯屏提前约两天。解读：这个案例没有碰任何个体数据，全部信号来自公开聚合——推荐侧的很多情报需求，在合规粒度上就能满足，越界往往只是懒。变式：把榜单换成价格曲线，同一套采样加告警就是价格战预警。

行为专线收工。最后一节回到没有明确下游的探索期需求：数据先攒着、任务以后定，怎么攒才不浪费——通用料件的数据集思维。