OuYingJun1024
6f2731ee2d
- 验签: pangle.build_sign/verify_callback_sign 改为 GroMore 官方
sign=SHA256("{m-key}:{trans_id}")(普通 SHA256,只签 trans_id);
回调响应改成 GroMore 要求的 {is_verify, reason}
- 发奖: 按回调 reward_amount 发金币(rewards 带回退/夹紧),ad_reward 幂等 + 每日上限
- 分层: app/core/{pangle,wxpay}.py → app/integrations/(对齐 jiguang/meituan/sms);
更新 ad.py / crud/wallet.py / scripts/sim_pangle_callback.py / 测试的 import
(含 test_withdraw 的 monkeypatch 路径 app.core.wxpay→app.integrations.wxpay)
- 测试: test_ad_reward + test_withdraw 共 17 项通过
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
91 lines
3.5 KiB
Python
91 lines
3.5 KiB
Python
"""模拟穿山甲激励视频 S2S 发奖回调,本地验证「验签 → 幂等发币 → 余额到账」闭环。
|
|
|
|
真穿山甲回调需要后端公网可达 + 穿山甲后台配置"奖励回调 URL",本地内网拿不到。本脚本用
|
|
FastAPI TestClient 在进程内打**真实** endpoint(`/api/v1/ad/pangle-callback`)、写**真实** DB
|
|
(./data/app.db),绕开公网依赖,证明后端这半条链路 OK——发完进 App 收益明细/重进福利页即可看到余额。
|
|
|
|
用法(在 shaguabijia-app-server 目录下,用本项目 python 跑):
|
|
python scripts/sim_pangle_callback.py # 给最近注册用户发一次(随机 trans_id)
|
|
python scripts/sim_pangle_callback.py 2 # 指定 user_id=2
|
|
python scripts/sim_pangle_callback.py 2 --trans t1 # 指定交易号;同号再跑应 Δ0(验幂等)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
import uuid
|
|
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
# 必须在 import app.* 之前开 PANGLE 开关(settings 在 import 时构造并缓存);
|
|
# 这里 setdefault 不覆盖 .env / 真实环境已有的值。
|
|
os.environ.setdefault("PANGLE_CALLBACK_ENABLED", "true")
|
|
os.environ.setdefault("PANGLE_REWARD_SECRET", "dev-local-pangle-secret")
|
|
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
from sqlalchemy import select # noqa: E402
|
|
|
|
from app.integrations import pangle # noqa: E402
|
|
from app.core.config import settings # noqa: E402
|
|
from app.core.rewards import AD_REWARD_COIN # noqa: E402
|
|
from app.crud import wallet as crud_wallet # noqa: E402
|
|
from app.db.session import SessionLocal # noqa: E402
|
|
from app.main import app # noqa: E402
|
|
from app.models.user import User # noqa: E402
|
|
|
|
|
|
def _latest_user_id() -> int | None:
|
|
db = SessionLocal()
|
|
try:
|
|
return db.execute(select(User.id).order_by(User.id.desc())).scalars().first()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _coin_balance(user_id: int) -> int:
|
|
db = SessionLocal()
|
|
try:
|
|
return crud_wallet.get_or_create_account(db, user_id).coin_balance
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def main() -> None:
|
|
p = argparse.ArgumentParser(description="模拟穿山甲发奖回调")
|
|
p.add_argument("user_id", nargs="?", type=int, default=None, help="目标 user_id(默认最近注册)")
|
|
p.add_argument("--trans", default=None, help="交易号(默认随机);同号重跑验幂等")
|
|
args = p.parse_args()
|
|
|
|
if not settings.pangle_callback_configured:
|
|
raise SystemExit("PANGLE 未配置(PANGLE_CALLBACK_ENABLED / PANGLE_REWARD_SECRET)")
|
|
|
|
uid = args.user_id or _latest_user_id()
|
|
if uid is None:
|
|
raise SystemExit("DB 里没有用户,先在 App 登录一次再跑")
|
|
|
|
trans_id = args.trans or f"sim_{uuid.uuid4().hex[:12]}"
|
|
before = _coin_balance(uid)
|
|
|
|
params = {
|
|
"user_id": str(uid),
|
|
"trans_id": trans_id,
|
|
"reward_name": "金币",
|
|
# 镜像生产:后台代码位"奖励数量"应配成 AD_REWARD_COIN,回调据此发金币
|
|
"reward_amount": str(AD_REWARD_COIN),
|
|
}
|
|
params["sign"] = pangle.build_sign(trans_id, settings.PANGLE_REWARD_SECRET)
|
|
|
|
resp = TestClient(app).get("/api/v1/ad/pangle-callback", params=params)
|
|
after = _coin_balance(uid)
|
|
|
|
print(f"user_id={uid} trans_id={trans_id}")
|
|
print(f"HTTP {resp.status_code} {resp.json()}")
|
|
print(f"金币: {before} -> {after} (Δ{after - before}, 单次应发 {AD_REWARD_COIN})")
|
|
print("同 trans_id 再跑应 Δ0(幂等);进 App 收益明细/重进福利页刷新即可看到余额。")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|