Pine
442 lines
18 KiB
Python
442 lines
18 KiB
Python
# -*- 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": "",
|
||
}
|