beige/backend/app/services/flywheel_service.py

"""
app/services/flywheel_service.py — 飞轮信号写入 + 偏好上下文聚合
preference_collector：三信号入口（选文案/选图/审核）写入 preference_events。
preference_aggregator：查最近50条 → 最常选角度 + 打回原因近3条原文拼 prompt。
飞轮不暴露独立埋点端点，只由业务接口内部调用（契约红线）。
"""

import logging
from typing import Any

from sqlalchemy import desc, func
from sqlalchemy.orm import Session

from app.constants.enums import SIGNAL_WEIGHTS, DataOwnership
from app.middleware.workspace_guard import CurrentUser
from app.models.flywheel import PreferenceEvent
from app.models.task import GenerationTask

logger = logging.getLogger(__name__)

# 实时聚合窗口：最近50条事件
_AGGREGATION_WINDOW = 50


def record_signal(
    db: Session,
    current_user: CurrentUser,
    task: GenerationTask,
    signal_type: str,
    candidate_id: int | None = None,
    angle_label: str | None = None,
    reason: str | None = None,
    signal_meta: str | None = None,
) -> None:
    """
    写入飞轮信号。
    workspace_id + product_id 都必须有（基石C + 按产品分开学）。
    signal_weight 用枚举默认值，北哥可校准。
    data_ownership 默认 client_data（选择行为归客户）。
    signal_meta：JSON 扩展（如选图存 strategy），不参与角度聚合主逻辑。
    """
    weight = SIGNAL_WEIGHTS.get(signal_type, 0)
    event = PreferenceEvent(
        workspace_id=current_user.workspace_id,
        product_id=task.product_id,
        task_id=task.id,
        user_id=current_user.user_id,
        signal_type=signal_type,
        signal_weight=weight,
        candidate_id=candidate_id,
        angle_label=angle_label,
        reason=reason,
        signal_meta=signal_meta,
        data_ownership=DataOwnership.CLIENT_DATA,
    )
    try:
        db.add(event)
        db.commit()
        logger.info(
            "Flywheel signal: type=%s weight=%s user=%s product=%s",
            signal_type, weight, current_user.user_id, task.product_id,
        )
    except Exception:
        db.rollback()
        logger.error(
            "Failed to write preference_event: type=%s user=%s",
            signal_type, current_user.user_id,
        )
        raise


def count_signals(db: Session, workspace_id: int, product_id: int) -> int:
    """该产品累计偏好信号总数（按 workspace+product，走联合索引）。
    供前端"飞轮已积累N条信号"累积感知用，体现越用越多。"""
    return (
        db.query(func.count(PreferenceEvent.id))
        .filter(
            PreferenceEvent.workspace_id == workspace_id,
            PreferenceEvent.product_id == product_id,
        )
        .scalar()
        or 0
    )


def get_preference_context(
    db: Session, workspace_id: int, product_id: int,
    product_dict: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """
    实时聚合偏好上下文（最近50条 events）供前端展示。
    R7断点3：统一口径——委托给生产链同一个 aggregate_preference_context（权重口径），
    消除"前端展示按次数/生成按权重"的口径分裂（说一套做一套）。
    按 workspace_id + product_id 严格过滤（不串数据，基石C）。
    """
    from app.services.ai_engine.preference_aggregator import aggregate_preference_context

    recent = (
        db.query(PreferenceEvent)
        .filter(
            PreferenceEvent.workspace_id == workspace_id,
            PreferenceEvent.product_id == product_id,
        )
        .order_by(desc(PreferenceEvent.created_at))
        .limit(_AGGREGATION_WINDOW)
        .all()
    )
    events_dicts = [
        {"signal_type": e.signal_type, "workspace_id": e.workspace_id,
         "product_id": e.product_id, "angle_label": e.angle_label or "",
         "signal_weight": e.signal_weight, "reason": e.reason or ""}
        for e in recent
    ]
    ctx = aggregate_preference_context(
        events_dicts, product_dict or {}, workspace_id, product_id
    )
    # 前端展示不需要 prompt_fragment（注入用），剥掉只回摘要+原因+计数
    return {
        "recent_preference": ctx.get("recent_preference", ""),
        "reject_reasons": ctx.get("reject_reasons", []),
        "injected_count": ctx.get("injected_count", 0),
        "signal_count": count_signals(db, workspace_id, product_id),
    }