"""结算服务：周期任务（推荐激活 / 视频创作收益 / 订单超时关单） 由 main.py lifespan 内的 asyncio 周期协程驱动（开发环境单进程够用）； 各结算函数均为幂等扫描，测试可直接调用。 """ import asyncio import logging from datetime import datetime, timedelta from decimal import Decimal import httpx from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.config import biz_config, settings from app.database import SessionFactory from app.models.billing import RechargeOrder, WatchRecord from app.models.incentive import Referral, VideoEarning from app.models.user import User from app.services import incentive_service, membership_service, wallet_service logger = logging.getLogger(__name__) # 单轮扫描批量上限 BATCH_SIZE = 500 # 周期间隔（秒） INTERVAL_REFERRAL = 600 # 推荐激活结算 10min INTERVAL_EARNINGS = 300 # 创作收益结算 5min INTERVAL_ORDER_CLOSE = 60 # 订单超时关单 1min INTERVAL_USDT_POLL = 15 # USDT 链上流水轮询 15s # USDT-TRC20 主网合约地址 USDT_CONTRACT = "TR7NHqjeKQxGTCi8q8ZY4pL8otSzgjLj6t" def _referral_cfg() -> dict: return biz_config.get("referral") or {} def _asset_cfg() -> dict: return biz_config.get("asset") or {} # ---------- 推荐激活结算 ---------- async def settle_referral_activations(db: AsyncSession) -> dict: """待激活推荐关系结算（《01-开发文档》2.8：72h 内观看≥N 且次日登录 → 有效；超期 → 无效）。 激活时点校验推荐人会员身份，会员才发放注册奖励积分（point biz_type=8）。 """ cfg = _referral_cfg() hours = int(cfg.get("activate_hours", 72)) watch_count = int(cfg.get("activate_watch_count", 3)) reward = int(cfg.get("register_reward_points", 10)) now = datetime.now() rows = (await db.execute( select(Referral).where(Referral.status == 0).order_by(Referral.id).limit(BATCH_SIZE) )).scalars().all() activated = expired = 0 for r in rows: if r.created_at + timedelta(hours=hours) < now: r.status = 2 # 超期未达标 expired += 1 continue watches = (await db.execute( select(func.count(WatchRecord.id)).where( WatchRecord.subject_type == 1, WatchRecord.subject_id == str(r.invitee_id) ) )).scalar() or 0 if watches < watch_count: continue invitee = await db.get(User, r.invitee_id) next_day_login = bool( invitee and invitee.last_login_at and invitee.last_login_at.date() > r.created_at.date() ) if not next_day_login: continue r.status = 1 r.activated_at = now activated += 1 if reward > 0 and await membership_service.is_member(db, r.inviter_id): await wallet_service.change_points( db, r.inviter_id, reward, wallet_service.BIZ_REFERRAL, biz_id=str(r.id), remark="推荐新用户激活奖励", ) await db.commit() return {"scanned": len(rows), "activated": activated, "expired": expired} # ---------- 视频创作收益结算 ---------- async def settle_video_earnings(db: AsyncSession) -> dict: """创作收益按增量结算：完整观看每满 1000 发 1 资产、点赞每满 100 发 1 资产（余数保留）。 结算时点校验作者会员身份：非会员只保留计数，不推进结算刻度。 """ cfg = _asset_cfg() views_per = int(cfg.get("complete_views_per_asset", 1000)) likes_per = int(cfg.get("likes_per_asset", 100)) rows = (await db.execute( select(VideoEarning).order_by(VideoEarning.video_id).limit(BATCH_SIZE) )).scalars().all() settled_videos = 0 granted = Decimal(0) for e in rows: if not await membership_service.is_member(db, e.author_id): continue view_assets = (e.complete_views - e.views_settled) // views_per if views_per > 0 else 0 like_assets = (e.likes_accrued - e.likes_settled) // likes_per if likes_per > 0 else 0 if view_assets <= 0 and like_assets <= 0: continue if view_assets > 0: await incentive_service.change_assets( db, e.author_id, Decimal(view_assets), incentive_service.ASSET_BIZ_COMPLETE, biz_id=str(e.video_id), remark="完整观看收益", ) e.views_settled += view_assets * views_per if like_assets > 0: await incentive_service.change_assets( db, e.author_id, Decimal(like_assets), incentive_service.ASSET_BIZ_LIKE, biz_id=str(e.video_id), remark="点赞收益", ) e.likes_settled += like_assets * likes_per settled_videos += 1 granted += view_assets + like_assets await db.commit() return {"scanned": len(rows), "settled_videos": settled_videos, "granted_assets": float(granted)} # ---------- 订单超时关单 ---------- async def close_expired_orders(db: AsyncSession) -> dict: """待支付订单超过 expire_at → 已关闭。 USDT 订单额外留 10 分钟宽限：用户在有效期末尾转账时，区块固化可能晚于 订单过期时间，若立即关单会导致已付款无法自动入账（只能人工补单）。 """ usdt_cfg = (biz_config.get("recharge") or {}).get("usdt") or {} grace_min = int(usdt_cfg.get("confirm_grace_min", 10)) now = datetime.now() rows = (await db.execute( select(RechargeOrder) .where(RechargeOrder.status == 0, RechargeOrder.expire_at < now) .order_by(RechargeOrder.id) .limit(BATCH_SIZE) )).scalars().all() closed = 0 for o in rows: if o.pay_method == 3 and o.expire_at > now - timedelta(minutes=grace_min): continue # USDT 待确认宽限期 o.status = 2 closed += 1 await db.commit() return {"closed": closed} # ---------- USDT 链上自动对账 ---------- async def poll_usdt_deposits(db: AsyncSession) -> dict: """USDT 自动对账：轮询收款地址的 TRC20 流水，四要素匹配后幂等入账。 匹配条件：to=收款地址 + 金额(含尾差)精确相等 + 转账时间在订单有效期内 + 交易已固化（only_confirmed，波场固化块即已被 ≥19 个 SR 确认）。 入账走 payment_service.settle_order（tx_hash 写 third_trade_no，唯一索引幂等）。 双 worker 会同时跑本任务：settle_order 的行锁保证只有一方入账成功。 """ from app.services import payment_service cfg = (biz_config.get("recharge") or {}).get("usdt") or {} address = cfg.get("address") or "" if not address: return {} expire_min = int(cfg.get("order_expire_min", 30)) now = datetime.now() # 有效期内的待支付 USDT 订单 orders = (await db.execute( select(RechargeOrder).where( RechargeOrder.pay_method == 3, RechargeOrder.status == 0, RechargeOrder.created_at >= now - timedelta(minutes=expire_min), ) )).scalars().all() if not orders: return {} min_ts = int(min(o.created_at for o in orders).timestamp() * 1000) headers = {"TRON-PRO-API-KEY": settings.tron_api_key} if settings.tron_api_key else {} async with httpx.AsyncClient(timeout=20) as client: resp = await client.get( f"https://api.trongrid.io/v1/accounts/{address}/transactions/trc20", params={ "only_to": "true", # 只要转入 "only_confirmed": "true", # 只要已固化（等价确认数≥19） "contract_address": USDT_CONTRACT, "min_timestamp": min_ts, "limit": 200, }, headers=headers, ) resp.raise_for_status() txs = resp.json().get("data") or [] matched = 0 used_txs: set[str] = set() # 同一笔流水只允许命中一个订单（防同金额订单重复入账） # 先快照订单关键字段：settle_order 失败时 rollback 会让 ORM 对象过期， # 继续遍历原对象会触发异步懒加载错误 snapshots = [ (o.order_no, o.usdt_amount, o.created_at) for o in orders if o.usdt_amount is not None ] for order_no, usdt_amount, created_at in snapshots: start_ms = created_at.timestamp() * 1000 end_ms = (created_at + timedelta(minutes=expire_min)).timestamp() * 1000 for tx in txs: tx_hash = tx.get("transaction_id") or "" if not tx_hash or tx_hash in used_txs: continue if tx.get("to") != address: continue # USDT 6 位小数：链上整数 → Decimal 精确比对（含随机尾差） if Decimal(tx.get("value") or "0") / Decimal(1_000_000) != usdt_amount: continue ts = int(tx.get("block_timestamp") or 0) if not (start_ms <= ts <= end_ms): continue try: await payment_service.settle_order(db, order_no, tx_hash) used_txs.add(tx_hash) matched += 1 logger.info("USDT 订单 %s 自动到账: tx=%s amount=%s", order_no, tx_hash, usdt_amount) except Exception: await db.rollback() logger.exception("USDT 订单 %s 入账失败", order_no) break return {"usdt_orders": len(snapshots), "matched": matched} if matched else {} # ---------- 周期协程（lifespan 挂载） ---------- async def _run_loop(name: str, fn, interval: int) -> None: while True: try: async with SessionFactory() as db: result = await fn(db) if result: logger.info("周期任务[%s]完成: %s", name, result) except asyncio.CancelledError: raise except Exception: logger.exception("周期任务[%s]执行失败", name) await asyncio.sleep(interval) def start_periodic_tasks() -> list[asyncio.Task]: """启动全部周期任务，返回 Task 列表（lifespan 退出时取消）""" return [ asyncio.create_task(_run_loop("推荐激活结算", settle_referral_activations, INTERVAL_REFERRAL)), asyncio.create_task(_run_loop("创作收益结算", settle_video_earnings, INTERVAL_EARNINGS)), asyncio.create_task(_run_loop("订单超时关单", close_expired_orders, INTERVAL_ORDER_CLOSE)), asyncio.create_task(_run_loop("USDT 链上对账", poll_usdt_deposits, INTERVAL_USDT_POLL)), ]