Pine
101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
"""邀请归因 API：邀请码、小程序码、分享/扫码归因、邀请统计。"""
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
from ...infrastructure.repositories import Database
|
|
from ...services import invite_service
|
|
from ..dependencies import get_current_user, get_db
|
|
|
|
router = APIRouter(prefix="/invite", tags=["邀请归因"])
|
|
|
|
|
|
class ShareRequest(BaseModel):
|
|
channel: str = Field(default="share_app", description="share_app|share_timeline|qrcode|link")
|
|
scene: str = Field(default="", description="场景：test_result|profile|home等")
|
|
env: str = Field(default="trial", description="trial|release")
|
|
|
|
|
|
class ScanRequest(BaseModel):
|
|
inviter: str = Field(description="邀请人ID或邀请码")
|
|
scene: str = Field(default="", description="扫码场景值")
|
|
env: str = Field(default="trial", description="trial|release")
|
|
|
|
|
|
class QrcodeRequest(BaseModel):
|
|
scene: str = Field(default="home", description="场景：home|test_result|profile")
|
|
page: str = Field(default="pages/index/index", description="小程序页面路径")
|
|
env: str = Field(default="trial", description="trial|release")
|
|
width: int = Field(default=430, description="小程序码宽度")
|
|
|
|
|
|
@router.get("/me", summary="获取我的邀请码和统计")
|
|
async def get_my_invite(user: dict = Depends(get_current_user), db: Database = Depends(get_db)):
|
|
stats = await invite_service.get_invite_stats(db, user["id"])
|
|
return stats
|
|
|
|
|
|
@router.get("/qrcode", summary="生成带邀请参数的小程序码")
|
|
async def get_invite_qrcode(
|
|
scene: str = "home",
|
|
page: str = "pages/index/index",
|
|
env: str = "trial",
|
|
width: int = 430,
|
|
user: dict = Depends(get_current_user),
|
|
db: Database = Depends(get_db),
|
|
):
|
|
result = await invite_service.get_wxacode(db, user["id"], scene=scene, page=page, env=env, width=width)
|
|
if not result.get("success"):
|
|
raise HTTPException(status_code=500, detail=f"生成小程序码失败: {result.get('error')}")
|
|
# 返回 base64 图片
|
|
import base64
|
|
img_b64 = base64.b64encode(result["image"]).decode()
|
|
return {
|
|
"image": f"data:image/png;base64,{img_b64}",
|
|
"scene": result["scene"],
|
|
"invite_code": result["invite_code"],
|
|
}
|
|
|
|
|
|
@router.post("/share", summary="记录分享行为")
|
|
async def record_share(
|
|
req: ShareRequest,
|
|
request: Request,
|
|
user: dict = Depends(get_current_user),
|
|
db: Database = Depends(get_db),
|
|
):
|
|
ip = request.client.host if request.client else ""
|
|
await invite_service.record_share(db, user["id"], channel=req.channel, scene=req.scene, env=req.env, ip=ip)
|
|
return {"success": True}
|
|
|
|
|
|
@router.post("/scan", summary="记录扫码行为（无需登录）")
|
|
async def record_scan(req: ScanRequest, request: Request, db: Database = Depends(get_db)):
|
|
ip = request.client.host if request.client else ""
|
|
await invite_service.record_scan(db, req.inviter, channel="qrcode", scene=req.scene, env=req.env, ip=ip)
|
|
return {"success": True}
|
|
|
|
|
|
@router.get("/invitees", summary="获取我邀请的用户列表")
|
|
async def get_invitees(
|
|
user: dict = Depends(get_current_user),
|
|
db: Database = Depends(get_db),
|
|
):
|
|
invitees = await db.users.get_invitees(user["id"])
|
|
from ...infrastructure.oss import resolve_url
|
|
return {
|
|
"total": len(invitees),
|
|
"list": [
|
|
{
|
|
"id": u["id"],
|
|
"nickname": u.get("nickname", ""),
|
|
"avatar": resolve_url(u.get("avatar", "")),
|
|
"phone": u.get("phone", ""),
|
|
"invited_at": u.get("invited_at", ""),
|
|
"created_at": u.get("created_at", ""),
|
|
}
|
|
for u in invitees
|
|
],
|
|
}
|