Pine
0fd210ac1f
- profitsharing.py：服务商分账客户端（添加接收方/发起分账/查分账/退款/回调验签） - payment_bindings + settlement_logs：OPC 收款绑定（personal openid / merchant 子商户号）与分账流水镜像 - settlement_service.release_escrow：验收后优先发起微信分账（95%接单者/5%平台），未配置时降级台账结算（channel=manual） - handle_split_notify：分账回调确认 escrow released（幂等） - 路由：分账回调 + 绑定创建/列表/停用 - 迁移：alembic 0039（新表 + escrows 幂等加列）+ 兜底迁移脚本 + 冒烟测试
199 lines
9.7 KiB
Python
199 lines
9.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""业务层 · 结算/合同/争议/撮合服务。
|
||
|
||
资金相关规则（佣金、结算、争议联动）在业务层集中，接口层瘦身。
|
||
|
||
结算（release_escrow）· 验收通过后：
|
||
1) 优先走「微信服务商分账」（资金托管）：
|
||
接单者 active 收款绑定 + 订单 transaction_id + 服务商配置齐备
|
||
→ 分账 95% 给接单者 / 5% 佣金给平台服务商商户号 → 等分账回调确认后置 released
|
||
2) 降级（未接微信收单 / 无绑定 / 分账未配置）：
|
||
仅台账置 released（channel=manual），不阻塞业务，供后续对账补齐。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import secrets
|
||
import time
|
||
|
||
from fastapi import HTTPException
|
||
|
||
from ..infrastructure.repositories import Database
|
||
from ..pay import config as pay_config
|
||
from ..pay import profitsharing
|
||
from ..pay.repository import PaymentBindingRepository, SettlementLogRepository
|
||
|
||
# 平台佣金比例
|
||
COMMISSION_RATE = 0.05
|
||
|
||
logger = logging.getLogger("services.settlement")
|
||
|
||
|
||
class SettlementService:
|
||
"""结算托管、电子合同、争议、投资撮合。"""
|
||
|
||
def __init__(self, db: Database):
|
||
self.db = db
|
||
|
||
async def release_escrow(self, task_id: str) -> dict:
|
||
"""验收通过后结算：优先微信服务商分账，不满足条件时降级台账结算。"""
|
||
task = await self.db.tasks.get(task_id)
|
||
if task is None or task["status"] != "completed":
|
||
raise HTTPException(status_code=400, detail="任务未完成，不能结算")
|
||
escrows = await self.db.escrows.list()
|
||
esc = next((e for e in escrows if e["task_id"] == task_id), None)
|
||
if esc is None:
|
||
amount = max(task.get("budget_min", 0), task.get("budget_max", 0))
|
||
commission = int(amount * COMMISSION_RATE)
|
||
esc = await self.db.escrows.create(task_id, task["title"], amount, commission)
|
||
|
||
tx_id = esc.get("transaction_id") or ""
|
||
if not tx_id or not pay_config.profitsharing_enabled():
|
||
# 降级：未接微信收单 / 服务商分账未配置 → 台账结算
|
||
esc = await self.db.escrows.update_fields(
|
||
esc["id"], {"channel": "manual", "share_status": "manual"})
|
||
esc = await self.db.escrows.set_status(esc["id"], "released")
|
||
await self._log(esc, "manual", "未接入微信收单或分账未配置，台账结算", "shared",
|
||
f"no_transaction_id={not tx_id}")
|
||
return esc
|
||
|
||
receiver, reason = await self._resolve_receiver(task)
|
||
if receiver is None:
|
||
esc = await self.db.escrows.update_fields(
|
||
esc["id"], {"channel": "manual", "share_status": "manual"})
|
||
esc = await self.db.escrows.set_status(esc["id"], "released")
|
||
await self._log(esc, "manual", "接单者无收款绑定，台账结算", "shared", reason)
|
||
return esc
|
||
|
||
# 发起微信分账：金额（元→分）；接单者 = 总额 - 佣金，平台佣金单独分给服务商商户号
|
||
total_fen = int(esc["amount"]) * 100
|
||
commission_fen = int(esc["commission"]) * 100
|
||
opc_fen = total_fen - commission_fen
|
||
share_no = f"PS_{int(time.time())}_{secrets.token_hex(4).upper()}"
|
||
receivers = [
|
||
{
|
||
"type": "MERCHANT_ID" if receiver["bind_type"] == "merchant" else "PERSONAL_OPENID",
|
||
"account": receiver.get("sub_mchid") or receiver.get("openid"),
|
||
"amount": max(opc_fen, 0),
|
||
"description": "任务服务费",
|
||
},
|
||
{
|
||
"type": "MERCHANT_ID",
|
||
"account": pay_config.WECHATPAY_SP_MCHID,
|
||
"amount": commission_fen,
|
||
"description": "平台服务费",
|
||
},
|
||
]
|
||
try:
|
||
await profitsharing.create_split(
|
||
transaction_id=tx_id, out_order_no=share_no, receivers=receivers,
|
||
sub_mchid=receiver.get("sub_mchid") or "",
|
||
)
|
||
except RuntimeError as exc:
|
||
esc = await self.db.escrows.update_fields(esc["id"], {
|
||
"channel": "wx_split", "share_order_no": share_no,
|
||
"share_status": "failed", "receiver_binding_id": receiver["id"],
|
||
})
|
||
await self._log(esc, "wx_split", "分账发起失败", "failed", str(exc)[:300])
|
||
raise HTTPException(status_code=502, detail=f"分账发起失败：{exc}") from exc
|
||
|
||
esc = await self.db.escrows.update_fields(esc["id"], {
|
||
"channel": "wx_split", "share_order_no": share_no,
|
||
"share_status": "sharing", "receiver_binding_id": receiver["id"],
|
||
})
|
||
await self._log(esc, "wx_split", "分账已发起，等待微信回调确认", "initiated",
|
||
receivers_json=receivers)
|
||
logger.info("分账已发起 task=%s share=%s receiver=%s", task_id, share_no, receiver["id"])
|
||
return esc
|
||
|
||
async def handle_split_notify(self, result: dict) -> bool:
|
||
"""分账结果回调：微信确认分账成功 → escrow released + 流水 shared（幂等）。"""
|
||
share_no = result.get("out_order_no", "")
|
||
if not share_no:
|
||
logger.warning("分账回调缺少 out_order_no")
|
||
return True
|
||
repo = SettlementLogRepository(self.db.session)
|
||
log = await repo.get_by_share_order_no(share_no)
|
||
if log is None:
|
||
logger.warning("分账回调无对应流水: %s", share_no)
|
||
return True
|
||
if log["status"] == "shared":
|
||
return True # 已确认（回调重放）
|
||
esc = await self.db.escrows.get(log["escrow_id"])
|
||
if esc is not None:
|
||
await self.db.escrows.update_fields(esc["id"], {"share_status": "shared"})
|
||
await self.db.escrows.set_status(esc["id"], "released")
|
||
await repo.set_status(log["id"], "shared", "微信分账回调确认")
|
||
await self.db.audit.add(action="escrow.released_wx", resource="escrow",
|
||
resource_id=log["escrow_id"],
|
||
detail=f"share={share_no}", user_id="")
|
||
logger.info("分账回调确认 share=%s escrow=%s", share_no, log["escrow_id"])
|
||
return True
|
||
|
||
async def _resolve_receiver(self, task: dict) -> tuple[dict | None, str]:
|
||
"""接单者（claimed_by）的可用收款绑定；无则返回 (None, 原因)。"""
|
||
claimer = task.get("claimed_by")
|
||
if not claimer:
|
||
return None, "任务无接单者(claimed_by 为空)"
|
||
repo = PaymentBindingRepository(self.db.session)
|
||
binding = await repo.get_active(claimer)
|
||
if binding is None:
|
||
return None, f"接单者无 active 收款绑定(user={claimer})"
|
||
return binding, ""
|
||
|
||
async def _log(self, esc: dict, channel: str, note: str, status: str,
|
||
detail: str = "", receivers_json: list | None = None) -> dict:
|
||
"""写分账流水镜像（金额统一为分，与微信口径一致）。"""
|
||
repo = SettlementLogRepository(self.db.session)
|
||
return await repo.create({
|
||
"escrow_id": esc.get("id", ""), "task_id": esc.get("task_id", ""),
|
||
"share_order_no": esc.get("share_order_no", ""), "channel": channel,
|
||
"amount": int(esc.get("amount", 0)) * 100,
|
||
"commission": int(esc.get("commission", 0)) * 100,
|
||
"receivers_json": json.dumps(receivers_json or [], ensure_ascii=False),
|
||
"status": status, "detail": detail or note,
|
||
})
|
||
|
||
async def sign_contract(self, task_id: str, actor: dict, opc_id: str | None = None) -> dict:
|
||
"""签订电子合同（幂等：已签返回现有合同）。"""
|
||
task = await self.db.tasks.get(task_id)
|
||
if task is None:
|
||
raise HTTPException(status_code=404, detail="Task not found")
|
||
existing = await self.db.contracts.get_for_task(task_id)
|
||
if existing is not None:
|
||
return existing
|
||
resolved_opc = opc_id or (actor["id"] if actor.get("role") == "opc_member" else "")
|
||
return await self.db.contracts.create(task_id, task["title"], actor["id"], resolved_opc)
|
||
|
||
async def create_dispute(self, task_id: str, actor: dict, reason: str) -> dict:
|
||
"""发起争议并联动任务置为 disputed。"""
|
||
task = await self.db.tasks.get(task_id)
|
||
if task is None:
|
||
raise HTTPException(status_code=404, detail="Task not found")
|
||
dispute = await self.db.disputes.create(task_id, task["title"], actor["id"], reason)
|
||
await self.db.tasks.set_status(task_id, "disputed")
|
||
return dispute
|
||
|
||
async def resolve_dispute(self, dispute_id: str) -> dict:
|
||
dispute = await self.db.disputes.set_status(dispute_id, "resolved", resolution="平台调解结案")
|
||
if dispute is None:
|
||
raise HTTPException(status_code=404, detail="Dispute not found")
|
||
return dispute
|
||
|
||
async def investor_matches(self, user_id: str) -> dict:
|
||
"""投资撮合：按投资人偏好（行业/阶段关键词）打分排序。"""
|
||
pref = await self.db.investor_prefs.get(user_id) or {}
|
||
industries = set(pref.get("industries") or [])
|
||
data = await self.db.portal_pages.get("investor", "projects") or {"items": []}
|
||
scored = []
|
||
for it in data.get("items", []):
|
||
text = f"{it.get('title', '')} {it.get('meta', '')}"
|
||
score = sum(10 for ind in industries if ind and ind in text)
|
||
if pref.get("stage") and pref["stage"] in text:
|
||
score += 5
|
||
scored.append((score, it))
|
||
scored.sort(key=lambda x: x[0], reverse=True)
|
||
return {"items": [{"title": it["title"], "meta": it["meta"], "tag": it.get("tag"),
|
||
"match": min(100, s + 50)} for s, it in scored]}
|