🛡️ 21-反诈App更新项目 - API 清单

{
  "type": "phone",
  "content": "对方自称公安局民警，要求转账到安全账户",
  "evidence_urls": ["https://cdn.example.com/img1.jpg"],
  "reporter_phone": "13800138000",
  "is_anonymous": false
}

package handler

import (
    "context"
    "encoding/json"
    "net/http"
    "time"

    "github.com/gin-gonic/gin"
    "github.com/go-redis/redis/v8"
    "github.com/google/uuid"
    "gorm.io/gorm"
)

// ReportSubmitRequest 举报请求
type ReportSubmitRequest struct {
    Type         string   `json:"type" binding:"required"`
    Content      string   `json:"content" binding:"required"`
    EvidenceURLs []string `json:"evidence_urls"`
    ReporterPhone string  `json:"reporter_phone"`
    IsAnonymous  bool     `json:"is_anonymous"`
}

// ReportResponse 举报响应
type ReportResponse struct {
    ReportID string `json:"report_id"`
    SerialNo string `json:"serial_no"`
}

// ReportService 举报服务
type ReportService struct {
    db       *gorm.DB
    redis    *redis.Client
    lockTTL  time.Duration
    cacheTTL time.Duration
}

// NewReportService 创建服务
func NewReportService(db *gorm.DB, redis *redis.Client) *ReportService {
    return &ReportService{
        db:       db,
        redis:    redis,
        lockTTL:  10 * time.Second,  // 分布式锁超时
        cacheTTL: 24 * time.Hour,    // 幂等缓存24小时
    }
}

// SubmitReport 提交举报（高并发安全实现）
func (s *ReportService) SubmitReport(c *gin.Context, req *ReportSubmitRequest) (*ReportResponse, error) {
    ctx := c.Request.Context()

    // 1. 幂等性检查 - 基于X-Idempotency-Key
    idempotencyKey := c.GetHeader("X-Idempotency-Key")
    if idempotencyKey == "" {
        idempotencyKey = uuid.New().String() // 自动生成
    }

    // 1.1 检查幂等缓存（Redis）
    cacheKey := "idempotent:report:" + idempotencyKey
    cachedResult, err := s.redis.Get(ctx, cacheKey).Result()
    if err == nil && cachedResult != "" {
        var cachedResp ReportResponse
        if json.Unmarshal([]byte(cachedResult), &cachedResp) == nil {
            // 返回缓存结果（HTTP 200，但告知客户端）
            c.Header("X-Idempotency-Hit", "true")
            return &cachedResp, nil
        }
    }

    // 2. 分布式锁 - 防止并发提交
    lockKey := "lock:report:submit:" + req.ReporterPhone
    lockCtx, cancel := context.WithTimeout(ctx, s.lockTTL)
    defer cancel()

    // 使用Redis SETNX实现分布式锁
    lockAcquired := false
    for i := 0; i < 3; i++ { // 最多重试3次
        acquired, err := s.redis.SetNX(lockCtx, lockKey, uuid.New().String(), s.lockTTL).Result()
        if err == nil && acquired {
            lockAcquired = true
            break
        }
        time.Sleep(100 * time.Millisecond) // 100ms后重试
    }

    if !lockAcquired {
        return nil, fmt.Errorf("系统繁忙，请稍后重试")
    }
    defer s.redis.Del(ctx, lockKey) // 释放锁

    // 3. 业务逻辑处理
    reportID := generateReportID()
    serialNo := generateSerialNo()

    // 数据库事务
    err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
        report := map[string]interface{}{
            "id":             reportID,
            "serial_no":      serialNo,
            "user_id":        getUserID(c),
            "type":           req.Type,
            "content":        req.Content,
            "reporter_phone": req.ReporterPhone,
            "is_anonymous":   req.IsAnonymous,
            "status":         "pending",
            "create_time":    time.Now(),
        }

        if err := tx.Create(&report).Error; err != nil {
            return err
        }

        // 保存证据URL
        for _, url := range req.EvidenceURLs {
            evidence := map[string]interface{}{
                "report_id":     reportID,
                "evidence_type": "image",
                "file_url":      url,
                "create_time":   time.Now(),
            }
            if err := tx.Create(&evidence).Error; err != nil {
                return err
            }
        }
        return nil
    })

    if err != nil {
        return nil, err
    }

    // 4. 缓存结果（用于幂等）
    resp := &ReportResponse{ReportID: reportID, SerialNo: serialNo}
    respBytes, _ := json.Marshal(resp)
    s.redis.Set(ctx, cacheKey, respBytes, s.cacheTTL)

    // 5. 清除用户举报数缓存（缓存穿透防护）
    userCacheKey := fmt.Sprintf("user:%d:report_count", getUserID(c))
    s.redis.Del(ctx, userCacheKey)

    return resp, nil
}

// 辅助函数
func generateReportID() string { return "RPT" + time.Now().Format("20060102150405") }
func generateSerialNo() string { return "SN" + time.Now().Format("20060102150405") }
func getUserID(c *gin.Context) int64 { /* 从JWT获取 */ return 1 }

import asyncio
import hashlib
import json
import uuid
from datetime import datetime, timedelta
from typing import Optional, List

from fastapi import Header, HTTPException, Depends
from pydantic import BaseModel
from redis import asyncio as aioredis
from sqlalchemy import select, insert, update
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

# ============== 数据模型 ==============
class ReportSubmitRequest(BaseModel):
    type: str  # phone/sms/social/other
    content: str
    evidence_urls: Optional[List[str]] = []
    reporter_phone: Optional[str] = None
    is_anonymous: bool = False

class ReportResponse(BaseModel):
    report_id: str
    serial_no: str

# ============== 高并发服务 ==============
class ReportService:
    def __init__(self, db: AsyncSession, redis: aioredis.Redis):
        self.db = db
        self.redis = redis
        self.lock_ttl = 10  # 秒
        self.idempotency_ttl = 24 * 60 * 60  # 24小时

    async def submit_report(
        self,
        req: ReportSubmitRequest,
        idempotency_key: Optional[str] = Header(None),
        user_id: int = Depends(lambda: get_current_user())
    ) -> ReportResponse:
        # 1. 幂等性检查
        if not idempotency_key:
            idempotency_key = str(uuid.uuid4())

        cache_key = f"idempotent:report:{idempotency_key}"
        cached = await self.redis.get(cache_key)
        if cached:
            # 返回缓存结果（幂等命中）
            return ReportResponse.parse_raw(cached)

        # 2. 分布式锁（使用Redis SET NX EX）
        lock_key = f"lock:report:submit:{req.reporter_phone or user_id}"
        lock_acquired = False

        for _ in range(3):  # 重试3次
            acquired = await self.redis.set(
                lock_key, uuid.uuid4().hex,
                nx=True, ex=self.lock_ttl
            )
            if acquired:
                lock_acquired = True
                break
            await asyncio.sleep(0.1)  # 100ms

        if not lock_acquired:
            raise HTTPException(status_code=503, detail="系统繁忙，请稍后重试")

        try:
            # 3. 生成ID
            report_id = f"RPT{datetime.now().strftime('%Y%m%d%H%M%S')}"
            serial_no = f"SN{datetime.now().strftime('%Y%m%d%H%M%S')}"

            # 4. 数据库事务（异步）
            async with self.db.begin():
                # 插入举报记录
                report_data = {
                    "id": report_id,
                    "serial_no": serial_no,
                    "user_id": user_id,
                    "type": req.type,
                    "content": req.content,
                    "reporter_phone": req.reporter_phone,
                    "is_anonymous": req.is_anonymous,
                    "status": "pending",
                    "create_time": datetime.now()
                }
                await self.db.execute(insert(self.report_table).values(**report_data))

                # 批量插入证据
                for url in req.evidence_urls:
                    evidence_data = {
                        "report_id": report_id,
                        "evidence_type": "image",
                        "file_url": url,
                        "create_time": datetime.now()
                    }
                    await self.db.execute(insert(self.evidence_table).values(**evidence_data))

            # 5. 缓存结果
            resp = ReportResponse(report_id=report_id, serial_no=serial_no)
            await self.redis.set(
                cache_key,
                resp.json(),
                ex=self.idempotency_ttl
            )

            # 6. 清除用户统计缓存
            await self.redis.delete(f"user:{user_id}:report_count")

            return resp

        finally:
            # 释放锁
            await self.redis.delete(lock_key)

# ============== API路由 ==============
@router.post("/api/antifraud/report/submit")
async def submit_report(
    request: ReportSubmitRequest,
    x_idempotency_key: Optional[str] = Header(None),
    authorization: str = Header(...)
):
    """
    提交举报接口

    高并发安全机制：
    1. X-Idempotency-Key 幂等键，重复提交返回相同结果
    2. Redis分布式锁，防止并发写
    3. 多级缓存，加速响应
    """
    service = ReportService(db, redis)
    return await service.submit_report(request, x_idempotency_key)

{
  "code": 0,
  "msg": "success",
  "data": {
    "report_id": "RPT202603270001",
    "serial_no": "SN2026032700001"
  }
}

// CacheService 多级缓存服务
type CacheService struct {
    localCache *cc.Cache    // L1: 本地缓存 (github.com/bluele/gcache)
    redis      *redis.Client // L2: Redis
    versionKey string        // 版本号key
}

// ProgressCacheKey 缓存结构
type ProgressCache struct {
    Data      *ProgressResponse
    Version   int64
    CreatedAt time.Time
}

// GetProgressWithCache 带缓存的进度查询
func (s *CacheService) GetProgressWithCache(ctx context.Context, reportID string) (*ProgressResponse, bool) {
    cacheKey := fmt.Sprintf("progress:%s", reportID)

    // L1: 先查本地缓存
    if s.localCache != nil {
        if cached, err := s.localCache.Get(cacheKey); err == nil {
            if pc, ok := cached.(*ProgressCache); ok {
                return pc.Data, true
            }
        }
    }

    // L2: 查Redis（带版本号）
    cachedData, err := s.redis.Get(ctx, cacheKey).Result()
    if err == nil && cachedData != "" {
        var pc ProgressCache
        if json.Unmarshal([]byte(cachedData), &pc) == nil {
            // L1回填
            if s.localCache != nil {
                s.localCache.Set(cacheKey, &pc, 30*time.Second) // L1: 30秒
            }
            return pc.Data, true
        }
    }

    return nil, false
}

// InvalidateProgressCache 缓存失效（版本号+删除）
func (s *CacheService) InvalidateProgressCache(ctx context.Context, reportID string) error {
    cacheKey := fmt.Sprintf("progress:%s", reportID)

    // 1. 删除L1
    if s.localCache != nil {
        s.localCache.Remove(cacheKey)
    }

    // 2. 删除L2
    if err := s.redis.Del(ctx, cacheKey).Err(); err != nil {
        return err
    }

    // 3. 递增版本号（可选，用于监听）
    versionKey := fmt.Sprintf("progress:version:%s", reportID)
    s.redis.Incr(ctx, versionKey)

    return nil
}

// GetProgress 获取进度（先缓存后DB）
func (s *CacheService) GetProgress(ctx context.Context, reportID string) (*ProgressResponse, error) {
    // 先查缓存
    if data, hit := s.GetProgressWithCache(ctx, reportID); hit {
        return data, nil
    }

    // 查数据库
    data, err := s.loadProgressFromDB(ctx, reportID)
    if err != nil {
        return nil, err
    }

    // 回填缓存
    pc := &ProgressCache{
        Data:      data,
        Version:   time.Now().Unix(),
        CreatedAt: time.Now(),
    }
    pcBytes, _ := json.Marshal(pc)
    s.redis.Set(ctx, fmt.Sprintf("progress:%s", reportID), pcBytes, 5*time.Minute)

    return data, nil
}

import asyncio
import json
import hashlib
from datetime import datetime, timedelta
from typing import Optional, Dict, Any

import redis.asyncio as aioredis
from cachetools import LRUCache

class MultiLevelCache:
    """多级缓存：L1本地 + L2 Redis + 版本号校验"""

    def __init__(self, redis_url: str, max_l1_size: int = 1000, l1_ttl: int = 30):
        self.redis = aioredis.from_url(redis_url)
        self.l1_cache: LRUCache = LRUCache(maxsize=max_l1_size)  # L1本地缓存
        self.l1_ttl = l1_ttl  # L1: 30秒
        self.l2_ttl = 5 * 60   # L2: 5分钟

    async def get_progress(self, report_id: str) -> Optional[Dict[str, Any]]:
        cache_key = f"progress:{report_id}"

        # L1: 本地缓存
        if cache_key in self.l1_cache:
            return self.l1_cache[cache_key]

        # L2: Redis缓存
        cached = await self.redis.get(cache_key)
        if cached:
            data = json.loads(cached)
            # 回填L1
            self.l1_cache[cache_key] = data
            return data

        return None

    async def set_progress(self, report_id: str, data: Dict[str, Any]):
        cache_key = f"progress:{report_id}"

        # 写入L2（带版本号）
        cache_entry = {
            "data": data,
            "version": datetime.now().timestamp(),
            "cached_at": datetime.now().isoformat()
        }
        await self.redis.set(
            cache_key,
            json.dumps(cache_entry),
            ex=self.l2_ttl
        )

        # 回填L1
        self.l1_cache[cache_key] = data

    async def invalidate(self, report_id: str):
        """缓存失效"""
        cache_key = f"progress:{report_id}"

        # 删除L1
        self.l1_cache.pop(cache_key, None)

        # 删除L2
        await self.redis.delete(cache_key)

        # 递增版本号
        version_key = f"progress:version:{report_id}"
        await self.redis.incr(version_key)

// BankStopRequest 银行卡止付请求
type BankStopRequest struct {
    StopAccountType string  `json:"stop_account_type" binding:"required"`
    StopAccount      string  `json:"stop_account" binding:"required"`
    StopAccountName  string  `json:"stop_account_name" binding:"required"`
    BankName         string  `json:"bank_name" binding:"required"`
    TransferTime     string  `json:"transfer_time" binding:"required"`
    Currency         string  `json:"currency" binding:"required"`
    TransferAmount   float64 `json:"transfer_amount" binding:"required"`
    IsSuspect        bool    `json:"is_suspect"`
    FraudType        string  `json:"fraud_type" binding:"required"`
    StopReason       string  `json:"stop_reason" binding:"required"`
}

// BankStopResponse 止付响应
type BankStopResponse struct {
    StopID         string                 `json:"stop_id"`
    Status         string                 `json:"status"`
    GuofanResponse map[string]interface{} `json:"guofan_response,omitempty"`
}

// BankStopService 止付服务
type BankStopService struct {
    db    *gorm.DB
    redis *redis.Client
}

// SubmitBankStop 提交银行卡止付（高并发安全）
func (s *BankStopService) SubmitBankStop(c *gin.Context, req *BankStopRequest) (*BankStopResponse, error) {
    ctx := c.Request.Context()

    // 1. 生成幂等键（基于关键字段hash）
    idempotencyKey := generateIdempotencyKey(req.StopAccount, req.TransferTime, req.TransferAmount)

    // 1.1 检查幂等缓存
    cacheKey := "idempotent:bank_stop:" + idempotencyKey
    if cached, err := s.redis.Get(ctx, cacheKey).Result(); err == nil {
        var cachedResp BankStopResponse
        if json.Unmarshal([]byte(cached), &cachedResp) == nil {
            c.Header("X-Idempotency-Hit", "true")
            return &cachedResp, nil
        }
    }

    // 2. 分布式锁（账户维度）
    lockKey := "lock:bank_stop:" + req.StopAccount
    lockCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
    defer cancel()

    if !s.acquireLock(lockCtx, lockKey, 10*time.Second) {
        return nil, fmt.Errorf("该账户正在处理中，请稍后重试")
    }
    defer s.redis.Del(ctx, lockKey)

    // 3. 数据库行级锁（悲观锁）
    var existingStop map[string]interface{}
    err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
        // 使用 FOR UPDATE 行级锁
        result := tx.Raw(`
            SELECT id, status FROM antifraud_case_bank_stop
            WHERE stop_account = ? AND status IN ('pending', 'approved')
            FOR UPDATE
        `, req.StopAccount).Scan(&existingStop)

        if result.Error != nil {
            return result.Error
        }

        // 如果存在待处理或已批准的止付记录，拒绝重复提交
        if len(existingStop) > 0 {
            return fmt.Errorf("该账户存在进行中的止付申请")
        }

        // 4. 创建止付记录
        stopID := generateStopID()
        stopRecord := map[string]interface{}{
            "id":                stopID,
            "case_id":           getCaseID(c), // 从上下文获取
            "stop_account_type": req.StopAccountType,
            "stop_account":      req.StopAccount,
            "stop_account_name": req.StopAccountName,
            "bank_name":         req.BankName,
            "transfer_time":     req.TransferTime,
            "currency":          req.Currency,
            "transfer_amount":   req.TransferAmount,
            "is_suspect":        req.IsSuspect,
            "fraud_type":        req.FraudType,
            "stop_reason":       req.StopReason,
            "status":            "pending",
            "create_time":       time.Now(),
        }

        if err := tx.Create(&stopRecord).Error; err != nil {
            return err
        }

        // 5. 调用国反平台API（异步，可选）
        go s.callGuofanAPI(stopID, req)

        return nil
    })

    if err != nil {
        return nil, err
    }

    // 6. 缓存幂等结果
    resp := &BankStopResponse{
        StopID:  "STOP" + time.Now().Format("20060102150405"),
        Status:  "pending",
    }
    respBytes, _ := json.Marshal(resp)
    s.redis.Set(ctx, cacheKey, respBytes, 24*time.Hour)

    return resp, nil
}

// acquireLock 获取分布式锁
func (s *BankStopService) acquireLock(ctx context.Context, key string, ttl time.Duration) bool {
    for i := 0; i < 3; i++ {
        acquired, err := s.redis.SetNX(ctx, key, uuid.New().String(), ttl).Result()
        if err == nil && acquired {
            return true
        }
        time.Sleep(100 * time.Millisecond)
    }
    return false
}

// getIdempotencyCache 获取幂等缓存
func (s *BankStopService) getIdempotencyCache(key string) ([]byte, error) {
    ctx := context.Background()
    return s.redis.Get(ctx, "idempotent:bank_stop:"+key).Bytes()
}

// setIdempotencyCache 设置幂等缓存
func (s *BankStopService) setIdempotencyCache(key string, data []byte, ttl time.Duration) error {
    ctx := context.Background()
    return s.redis.Set(ctx, "idempotent:bank_stop:"+key, data, ttl).Err()
}

// callGuofanAPI 异步调用国反平台API
func (s *BankStopService) callGuofanAPI(stopID string, req *BankStopRequest) {
    // 异步调用国反平台止付接口
    // 具体实现根据国反平台接口文档
}