server-core/app/services/credit_engine.py

# -*- coding: utf-8 -*-
"""信用体系核心服务（累加积分制 · 无上限 · 初始 0 分）。

设计要点（v1.1）：
- 信用总分 = Σ(所有事件 delta)，无上限；新用户初始 0 分（L1 待起步）
- 六维为画像维度（0-100 归一化，供雷达图展示相对水平），不参与总分
- 事件驱动：先落 credit_ledger 流水 → 重算累计分+等级 → 徽章评估
- 每日全量重算兜底：full_recompute()（存量用户按历史数据回填）
"""
from __future__ import annotations

import logging
from typing import Any

from ..infrastructure.repositories import Database
from .badge_engine import evaluate_badges

log = logging.getLogger("credit.engine")

# 各事件计算规则使用的统计口径（避免魔法数字散落）
_ON_TIME_WINDOW_DAYS = 90          # 履约统计窗口
_POSITIVE_REVIEW_MIN = 3           # 好评最低星
_NEGATIVE_REVIEW_MAX = 2           # 差评最高星


async def _compute_delta(db: Database, rule: dict, payload: dict | None) -> int:
    """按规则计算事件分值（fixed / by_score / by_level / by_hours）。"""
    mode = rule.get("delta_mode", "fixed")
    base = int(rule.get("base_delta", 0))
    cfg = rule.get("delta_config", {}) or {}
    cap = abs(int(rule.get("cap_single", 0)))
    payload = payload or {}

    if mode == "fixed":
        delta = base
    elif mode == "by_score":
        # payload: {"score": 5} → cfg {"5":15,"4":10,...}
        score = payload.get("score")
        delta = int(cfg.get(str(score), base)) if score is not None else base
    elif mode == "by_level":
        # payload: {"level": "expert"} → cfg {"beginner":40,...}
        level = payload.get("level", "")
        delta = int(cfg.get(level, base)) if level else base
    elif mode == "by_hours":
        # payload: {"hours": 25} → cfg {"per":10,"monthly_cap":30}
        per = int(cfg.get("per", 10) or 10)
        hours = int(payload.get("hours", 0) or 0)
        delta = base * (hours // per) if per > 0 else 0
    else:
        delta = base

    # 单事件封顶（取绝对值比较，保留符号）
    if cap > 0:
        delta = max(-cap, min(cap, delta))
    return delta


async def on_credit_event(
    db: Database, *,
    user_id: str,
    event_code: str,
    ref_type: str = "",
    ref_id: str = "",
    payload: dict | None = None,
    reason: str = "",
    source: str = "system",
    operator_id: str = "",
    skip_unique_check: bool = False,
) -> dict:
    """信用事件入口：规则匹配 → 落流水 → 重算 → 徽章评估 → 返回结果。

    幂等性：
    - 规则 unique_per_user=1（认证类）：同一用户该事件只计一次（去重）
    - 规则 unique_per_user=0：按 ref_id 去重（同一任务/课程只计一次）
    """
    rule = await db.credit_rules.get(event_code)
    if rule is None or not rule.get("is_active"):
        log.info("credit event ignored: %s (rule missing/disabled)", event_code)
        return {"ok": False, "reason": "rule_disabled"}

    # 幂等去重
    if not skip_unique_check:
        if rule.get("unique_per_user"):
            if await db.credit_ledger.has_event(user_id, event_code):
                return {"ok": False, "reason": "already_counted", "event_code": event_code}
        elif ref_id:
            if await db.credit_ledger.has_event(user_id, event_code, ref_id):
                return {"ok": False, "reason": "already_counted", "event_code": event_code}

    delta = await _compute_delta(db, rule, payload)
    dimension = rule.get("dimension", "system")
    if not reason:
        reason = rule.get("name", event_code)

    entry = await db.credit_ledger.add(
        user_id=user_id, event_code=event_code, delta=delta, dimension=dimension,
        reason=reason, ref_type=ref_type, ref_id=ref_id,
        source=source, operator_id=operator_id,
    )

    # 重算累计分 + 等级（同步 opc_profiles.credit_score 兼容字段）
    recalc = await recalc_user(db, user_id)

    # 徽章评估（信用分变化可能触发等级徽章）
    granted = await evaluate_badges(db, user_id)

    # 实时通知：信用分变化（含原因与当前值；不阻塞主流程）
    try:
        from .notification_service import notify as _n

        total = recalc.get("total_score", 0)
        delta_txt = f"+{delta}" if delta >= 0 else str(delta)
        await _n(
            db, user_id, "credit", "信用分变化",
            f"信用分 {delta_txt}（{reason}），当前 {total} 分",
            event_code="credit.changed",
            level="success" if delta >= 0 else "warning",
            link="/opc/credit", ref_type=ref_type or "credit", ref_id=ref_id or "",
        )
        for bg in granted or []:
            await _n(
                db, user_id, "credit", "获得徽章",
                f"你获得了「{bg.get('name') or bg.get('badge_code') or '新徽章'}」徽章",
                event_code="credit.badge_granted", level="success",
                link="/opc/credit", ref_type="badge", ref_id=bg.get("badge_code") or "",
            )
    except Exception:  # noqa: BLE001
        pass

    return {"ok": True, "entry": entry, "granted_badges": granted}


async def recalc_user(db: Database, user_id: str) -> dict:
    """重算单个用户：累计分（流水汇总）+ 等级 + 六维画像。"""
    total = await db.credit_ledger.sum_delta(user_id)
    level_info = await db.credit_levels.level_for(total)
    dimensions = await _calc_dimensions(db, user_id, total)
    rating_stats = await _rating_stats(db, user_id)
    task_stats = await _task_stats(db, user_id)
    percentile = await _calc_percentile(db, total)

    await db.credit_scores.upsert(
        user_id, total_score=total, level=level_info["level"], dimensions=dimensions,
        rating_avg=rating_stats["avg"], rating_count=rating_stats["count"],
        task_count=task_stats["completed"], on_time_rate=task_stats["on_time_rate"],
        percentile=percentile,
    )
    # 同步旧字段（opc_profiles.credit_score），兼容 /me/credit 等旧接口
    await db.opc_profiles.upsert(user_id, credit_score=total)

    return {
        "user_id": user_id, "total_score": total,
        "level": level_info["level"], "level_name": level_info.get("name", ""),
        "dimensions": dimensions,
        "rating_avg": rating_stats["avg"], "rating_count": rating_stats["count"],
        "task_count": task_stats["completed"], "on_time_rate": task_stats["on_time_rate"],
        "percentile": percentile,
    }


async def _calc_dimensions(db: Database, user_id: str, total_score: int) -> list[dict]:
    """六维画像（0-100 归一化，相对水平展示；不参与总分）。

    口径（v1.1，可后续细化）：
    - capability 能力：按有效认证数 + 技能测试 + 培训完成换算
    - reputation 好评：贝叶斯修正均分 → 0-100
    - performance 履约：按时交付率 → 0-100
    - compliance 合规：负向扣分事件数 → 从 100 起扣
    - domain 领域：按领域认证/订单占比（简化为领域认证数）
    - activity 活跃：按活跃类事件加分累计换算
    """
    dims_cfg = {d["code"]: d for d in await db.credit_dimensions.list_active()}
    result: list[dict] = []

    # 能力：认证 + 培训 + 技能测试
    try:
        certs = await db.certifications.list(user_id=user_id, status="active")
        cert_count = len(certs)
    except Exception:  # noqa: BLE001
        cert_count = 0
    training_done = 0
    try:
        training_done = len(await db.course_certificates.list_by_user(user_id))
    except Exception:  # noqa: BLE001
        training_done = 0
    cap_score = min(100, 10 + cert_count * 15 + training_done * 5)

    # 好评：贝叶斯修正
    rating_stats = await _rating_stats(db, user_id)
    avg = rating_stats["avg"]
    n = rating_stats["count"]
    C, m = 5, 4.0
    bayes = (C * m + avg * n) / (C + n) if n else m
    rep_score = min(100, bayes / 5 * 100)

    # 履约：按时交付率（简化：用流水中的正负履约事件）
    per_score = await _performance_score(db, user_id)

    # 合规：从 100 起按负向扣
    comp_score = await _compliance_score(db, user_id)

    # 领域：领域认证数量
    try:
        certs_all = await db.certifications.list(user_id=user_id, status="active")
        domain_certs = sum(1 for c in certs_all if (c.get("cert_category") or c.get("cert_type", "")) == "domain")
    except Exception:  # noqa: BLE001
        domain_certs = 0
    domain_score = min(100, 10 + domain_certs * 20)

    # 活跃：按活跃事件加分
    act_score = await _activity_score(db, user_id)

    for code, name, weight, desc in [
        ("performance", "履约", 0.30, "按时交付率、一次验收通过率、完成率"),
        ("reputation", "好评", 0.20, "贝叶斯修正均分、好评率、评价数量"),
        ("capability", "能力", 0.20, "认证数量与等级、技能测试、培训完成"),
        ("domain", "领域", 0.15, "同领域深耕度、领域认证"),
        ("compliance", "合规", 0.10, "实名认证、违规记录、处罚情况"),
        ("activity", "活跃", 0.05, "登录频次、接单活跃度"),
    ]:
        score = {"performance": per_score, "reputation": rep_score, "capability": cap_score,
                 "domain": domain_score, "compliance": comp_score, "activity": act_score}[code]
        cfg = dims_cfg.get(code, {})
        result.append({
            "code": code,
            "name": cfg.get("name", name),
            "score": int(round(score)),
            "weight": cfg.get("weight", weight),
            "description": cfg.get("description", desc),
        })
    return result


async def _rating_stats(db: Database, user_id: str) -> dict:
    """好评维度统计：平均分 + 评价数（ratings 表，用户作为被评对象）。"""
    try:
        rows = await db.ratings.list_for(user_id)
        if not rows:
            return {"avg": 0.0, "count": 0}
        total = sum(float(r.get("score", 0)) for r in rows)
        return {"avg": round(total / len(rows), 2), "count": len(rows)}
    except Exception:  # noqa: BLE001
        return {"avg": 0.0, "count": 0}


async def _task_stats(db: Database, user_id: str) -> dict:
    """履约统计：已完成任务数 + 按时交付率（基于信用流水统计）。"""
    completed = 0
    on_time = 0
    total_done = 0
    try:
        rows = await db.credit_ledger.list_all(user_id=user_id, limit=1000)
        for r in rows:
            if r["event_code"] == "task.complete.on_time":
                completed += 1
                on_time += 1
                total_done += 1
            elif r["event_code"] == "task.complete.overdue":
                completed += 1
                total_done += 1
            elif r["event_code"] == "task.cancel.provider":
                total_done += 1
    except Exception:  # noqa: BLE001
        pass
    return {
        "completed": completed,
        "on_time_rate": round(on_time / total_done, 3) if total_done else 0.0,
    }


async def _performance_score(db: Database, user_id: str) -> int:
    """履约画像：按时交付 +10 / 逾期 -15 / 弃单 -20，100 为基线上限。"""
    on_time = 0
    overdue = 0
    cancel = 0
    try:
        rows = await db.credit_ledger.list_all(user_id=user_id, limit=1000)
        for r in rows:
            if r["event_code"] == "task.complete.on_time":
                on_time += 1
            elif r["event_code"] == "task.complete.overdue":
                overdue += 1
            elif r["event_code"] == "task.cancel.provider":
                cancel += 1
    except Exception:  # noqa: BLE001
        pass
    if on_time + overdue + cancel == 0:
        return 50
    score = on_time * 2 - overdue * 5 - cancel * 8
    return max(0, min(100, score + 50))


async def _compliance_score(db: Database, user_id: str) -> int:
    """合规画像：从 100 起按负向事件扣。"""
    penalty = 0
    try:
        rows = await db.credit_ledger.list_all(user_id=user_id, limit=1000)
        for r in rows:
            if r["event_code"] in ("violation.penalty", "violation.fraud", "violation.complaint",
                                   "dispute.lost", "deposit.deducted"):
                penalty += abs(r["delta"])
    except Exception:  # noqa: BLE001
        pass
    return max(0, min(100, 100 - penalty // 10))


async def _activity_score(db: Database, user_id: str) -> int:
    """活跃画像：按活跃类事件累计。"""
    act = 0
    try:
        rows = await db.credit_ledger.list_all(user_id=user_id, limit=1000)
        for r in rows:
            if r["event_code"] in ("behavior.daily", "task.accept", "service.publish"):
                act += abs(r["delta"])
    except Exception:  # noqa: BLE001
        pass
    return max(0, min(100, act * 2))


async def _calc_percentile(db: Database, total_score: int) -> int:
    """同平台百分位估算：以 credit_scores 表分布计算。"""
    try:
        rows = await db.credit_scores.list_top(limit=100000)
        if not rows:
            return 0
        below = sum(1 for r in rows if r["total_score"] < total_score)
        return int(below / len(rows) * 100)
    except Exception:  # noqa: BLE001
        return 0


async def full_recompute(db: Database) -> dict:
    """全量重算：所有用户按历史数据回填事件 → 重算总分/等级/徽章。

    回填策略（source='backfill'，可审计区分）：
    - 已有认证（certifications status active/approved）→ cert.* 事件（按认证类型）
    - 已有技能测试通过 → skill.test.passed
    - 已有培训结业（training_certificates）→ training.completed
    - 已有评价（ratings）→ task.review.positive/negative
    - 已有完成任务 → task.complete.on_time
    随后逐用户 recalc + 徽章评估。返回统计。
    """
    stats = {"users": 0, "backfilled_events": 0, "badges_granted": 0}

    # 1. 收集所有用户
    user_ids = [u.get("id") for u in await db.users.list() if u.get("id")]
    for uid in user_ids:
        stats["users"] += 1
        await _backfill_user(db, uid)
        await recalc_user(db, uid)
        granted = await evaluate_badges(db, uid)
        stats["badges_granted"] += len(granted)

    # 2. 更新等级徽章（等级变化回收/授予由 evaluate_badges 处理）
    log.info("full_recompute done: %s", stats)
    return stats


async def _backfill_user(db: Database, user_id: str) -> int:
    """按历史数据回填事件（幂等：已存在的流水不重复）。"""
    count = 0

    # 认证回填
    try:
        certs = await db.certifications.list(user_id=user_id, status="active")
        for c in certs:
            ctype = c.get("cert_category", "") or c.get("cert_type", "")
            event_map = {
                "identity": "cert.identity",
                "merchant": "cert.merchant",
                "enterprise": "cert.enterprise",
                "education": "cert.education",
                "qualification": "cert.qualification",
                "skill": "cert.skill",
                "opc": "cert.opc",
            }
            code = event_map.get(ctype, "")
            if not code:
                continue
            if await db.credit_ledger.has_event(user_id, code, c.get("id", "")):
                continue
            await on_credit_event(
                db, user_id=user_id, event_code=code,
                ref_type="certification", ref_id=c.get("id", ""),
                source="backfill", skip_unique_check=True,
            )
            count += 1
    except Exception as exc:  # noqa: BLE001
        log.warning("backfill cert failed for %s: %s", user_id, exc)

    # 培训回填
    try:
        if hasattr(db.course_certificates, "list_by_user"):
            certs2 = await db.course_certificates.list_by_user(user_id)
            for cc in certs2:
                cid = cc.get("id", "")
                if await db.credit_ledger.has_event(user_id, "training.completed", cid):
                    continue
                await on_credit_event(
                    db, user_id=user_id, event_code="training.completed",
                    ref_type="training_cert", ref_id=cid,
                    source="backfill", skip_unique_check=True,
                )
                count += 1
    except Exception as exc:  # noqa: BLE001
        log.warning("backfill training failed for %s: %s", user_id, exc)

    # 评价回填
    try:
        ratings = await db.ratings.list_for(user_id)
        for r in ratings:
            rid = r.get("id", "")
            score = float(r.get("score", 0) or 0)
            code = "task.review.positive" if score >= _POSITIVE_REVIEW_MIN else "task.review.negative"
            if await db.credit_ledger.has_event(user_id, code, rid):
                continue
            await on_credit_event(
                db, user_id=user_id, event_code=code,
                payload={"score": int(score)},
                ref_type="task", ref_id=r.get("task_id", "") or rid,
                reason=f"历史评价 {score} 星（回填）",
                source="backfill", skip_unique_check=True,
            )
            count += 1
    except Exception as exc:  # noqa: BLE001
        log.warning("backfill rating failed for %s: %s", user_id, exc)

    return count


async def ensure_user_profile(db: Database, user_id: str) -> dict:
    """确保用户有信用档案（新用户初始 0 分 = L1）。"""
    existing = await db.credit_scores.get(user_id)
    if existing is not None:
        return existing
    await recalc_user(db, user_id)
    return await db.credit_scores.get(user_id) or {
        "user_id": user_id, "total_score": 0, "level": "L1",
        "dimensions": [], "rating_avg": 0.0, "rating_count": 0,
        "task_count": 0, "on_time_rate": 0.0, "percentile": 0, "updated_at": "",
    }