diff --git a/.gitignore b/.gitignore index f1b7204..b8455be 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ /.idea/* +/docs/* \ No newline at end of file diff --git a/config.yml b/config.yml index 604c07e..bba88ca 100644 --- a/config.yml +++ b/config.yml @@ -11,7 +11,7 @@ rate: # Database. database: - default: + cid: - type: "pgsql" host: "116.204.74.41" port: "15432" @@ -70,10 +70,11 @@ jaeger: #链路追踪 yidun: # 回调模式开关: true=使用回调模式（需要公网地址）, false=使用轮询模式 callback_mode: false + # 易盾回调 IP 白名单（逗号分隔），留空则不校验 + callback_allowed_ips: "" # 视频检测配置 video: - business_id: "YD00256761935486" secret_id: "f58a38341ca6227014df7c3bf0e6f16f" secret_key: "526aa631ba5d518aedeb70b5a3b67371" region: "cn-hangzhou" diff --git a/consts/public/collections.go b/consts/public/collections.go index 7c988f4..9fd24e5 100644 --- a/consts/public/collections.go +++ b/consts/public/collections.go @@ -1,10 +1 @@ package public - -// PostgreSQL表名常量 -const ( - PlatformTable = "cid_platform" // 平台管理表 - ApiInterfaceTable = "cid_api_interface" // 接口管理表 - DataFetchLogTable = "cid_data_fetch_log" // 数据获取日志表 - DataMappingTable = "cid_data_mapping" // 数据映射表 - ApplicationTable = "cid_application" // 应用管理表 -) diff --git a/controller/dataengine/material_verify_controller.go b/controller/dataengine/material_verify_controller.go index d4333f2..3616ef3 100644 --- a/controller/dataengine/material_verify_controller.go +++ b/controller/dataengine/material_verify_controller.go @@ -7,9 +7,12 @@ import ( serviceDataengine "cid/service/dataengine" "context" "fmt" + "strings" "time" + "gitea.redpowerfuture.com/red-future/common/beans" "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/net/ghttp" ) // MaterialVerifyController 素材校验控制器 @@ -104,6 +107,8 @@ type BatchVerifyReq struct { // ListImage 图片素材列表 func (c *MaterialVerifyController) ListImage(ctx context.Context, req *ImageListReq) (res *ImageListRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) + if req.Page == 0 { req.Page = 1 } @@ -132,6 +137,7 @@ func (c *MaterialVerifyController) ListImage(ctx context.Context, req *ImageList // StatsImage 图片素材统计 func (c *MaterialVerifyController) StatsImage(ctx context.Context, req *ImageListReq) (res *StatsRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) // 使用实体中定义的正确状态值：PENDING=待校验, VERIFIED=校验通过, REJECTED=校验不通过 pending, _ := dao.TencentImage.CountByStatus(ctx, entity.VerifyStatusPending) verified, _ := dao.TencentImage.CountByStatus(ctx, entity.VerifyStatusVerified) @@ -150,6 +156,8 @@ func (c *MaterialVerifyController) StatsImage(ctx context.Context, req *ImageLis // ListVideo 视频素材列表 func (c *MaterialVerifyController) ListVideo(ctx context.Context, req *VideoListReq) (res *VideoListRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) + if req.Page == 0 { req.Page = 1 } @@ -178,6 +186,7 @@ func (c *MaterialVerifyController) ListVideo(ctx context.Context, req *VideoList // StatsVideo 视频素材统计 func (c *MaterialVerifyController) StatsVideo(ctx context.Context, req *VideoListReq) (res *StatsRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) // 使用实体中定义的正确状态值：PENDING=待校验, VERIFIED=校验通过, REJECTED=校验不通过 pending, _ := dao.TencentVideo.CountByStatus(ctx, entity.VerifyStatusPending) verified, _ := dao.TencentVideo.CountByStatus(ctx, entity.VerifyStatusVerified) @@ -202,6 +211,8 @@ type ListLogRes struct { // ListLog 日志列表 func (c *MaterialVerifyController) ListLog(ctx context.Context, req *LogListReq) (res *ListLogRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) + if req.Page == 0 { req.Page = 1 } @@ -247,6 +258,8 @@ type GetLogDetailReq struct { // GetLogDetail 日志详情 func (c *MaterialVerifyController) GetLogDetail(ctx context.Context, req *GetLogDetailReq) (res *LogDetailRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) + log, err := serviceDataengine.MaterialVerify.GetLogByID(ctx, req.Id) if err != nil { return nil, err @@ -284,6 +297,7 @@ type StatsLogRes struct { // StatsLog 日志统计 func (c *MaterialVerifyController) StatsLog(ctx context.Context, req *LogListReq) (res *StatsLogRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) stats, err := serviceDataengine.MaterialVerify.GetStats(ctx) if err != nil { return nil, err @@ -310,6 +324,8 @@ type ManualVerifyImageRes struct { // ManualVerifyImage 手动校验图片 func (c *MaterialVerifyController) ManualVerifyImage(ctx context.Context, req *ManualVerifyReq) (res *ManualVerifyImageRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) + log, err := serviceDataengine.MaterialVerify.VerifyImageByID(ctx, req.MaterialID) if err != nil { return nil, err @@ -324,6 +340,8 @@ func (c *MaterialVerifyController) ManualVerifyImage(ctx context.Context, req *M // ManualVerifyVideo 手动校验视频 func (c *MaterialVerifyController) ManualVerifyVideo(ctx context.Context, req *ManualVerifyReq) (res *ManualVerifyImageRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) + log, err := serviceDataengine.MaterialVerify.VerifyVideoByID(ctx, req.MaterialID) if err != nil { return nil, err @@ -350,6 +368,8 @@ type BatchVerifyRes struct { // BatchVerifyImage 批量校验图片 func (c *MaterialVerifyController) BatchVerifyImage(ctx context.Context, req *BatchVerifyReq) (res *BatchVerifyRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) + if req.Limit <= 0 { req.Limit = 100 } @@ -363,37 +383,28 @@ func (c *MaterialVerifyController) BatchVerifyImage(ctx context.Context, req *Ba failCount := 0 for _, image := range images { - log, err := serviceDataengine.MaterialVerify.VerifyImageByID(ctx, image.ImageID) + _, err := serviceDataengine.MaterialVerify.VerifyImageByID(ctx, image.ImageID) if err != nil { failCount++ g.Log().Errorf(ctx, "图片校验失败: %s, error: %v", image.ImageID, err) } else { successCount++ - g.Log().Infof(ctx, "图片校验已提交: %s, logId: %d", image.ImageID, log.Id) } time.Sleep(100 * time.Millisecond) } - // 等待易盾处理，然后自动查询结果 - msg := fmt.Sprintf("批量校验完成，成功: %d，失败: %d", successCount, failCount) - if successCount > 0 { - g.Log().Infof(ctx, "提交完成，等待2秒后自动查询结果...") - time.Sleep(2 * time.Second) - pollSuccess, pollFail, _ := serviceDataengine.MaterialVerify.PollPendingResults(ctx) - msg = fmt.Sprintf("批量校验完成，提交成功: %d，提交失败: %d，自动查询成功: %d，未就绪: %d", - successCount, failCount, pollSuccess, pollFail) - } - return &BatchVerifyRes{ Success: successCount, Fail: failCount, Total: len(images), - Message: msg, + Message: fmt.Sprintf("批量提交完成，成功: %d，失败: %d。请通过轮询接口获取检测结果", successCount, failCount), }, nil } // BatchVerifyVideo 批量校验视频 func (c *MaterialVerifyController) BatchVerifyVideo(ctx context.Context, req *BatchVerifyReq) (res *BatchVerifyRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) + if req.Limit <= 0 { req.Limit = 100 } @@ -407,32 +418,21 @@ func (c *MaterialVerifyController) BatchVerifyVideo(ctx context.Context, req *Ba failCount := 0 for _, video := range videos { - log, err := serviceDataengine.MaterialVerify.VerifyVideoByID(ctx, video.VideoID) + _, err := serviceDataengine.MaterialVerify.VerifyVideoByID(ctx, video.VideoID) if err != nil { failCount++ g.Log().Errorf(ctx, "视频校验失败: %s, error: %v", video.VideoID, err) } else { successCount++ - g.Log().Infof(ctx, "视频校验已提交: %s, logId: %d", video.VideoID, log.Id) } time.Sleep(100 * time.Millisecond) } - // 等待易盾处理，然后自动查询结果 - msg := fmt.Sprintf("批量校验完成，成功: %d，失败: %d", successCount, failCount) - if successCount > 0 { - g.Log().Infof(ctx, "提交完成，等待2秒后自动查询结果...") - time.Sleep(2 * time.Second) - pollSuccess, pollFail, _ := serviceDataengine.MaterialVerify.PollPendingResults(ctx) - msg = fmt.Sprintf("批量校验完成，提交成功: %d，提交失败: %d，自动查询成功: %d，未就绪: %d", - successCount, failCount, pollSuccess, pollFail) - } - return &BatchVerifyRes{ Success: successCount, Fail: failCount, Total: len(videos), - Message: msg, + Message: fmt.Sprintf("批量提交完成，成功: %d，失败: %d。请通过轮询接口获取检测结果", successCount, failCount), }, nil } @@ -456,6 +456,7 @@ type ListAccountsRes struct { // ListAccounts 获取所有启用的广告账户列表（用于前端下拉筛选） func (c *MaterialVerifyController) ListAccounts(ctx context.Context, req *ListAccountsReq) (res *ListAccountsRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) accounts, err := dao.TencentAccountRelation.GetAll(ctx) if err != nil { return nil, err @@ -505,6 +506,8 @@ type ExportRejectedRes struct { // ExportRejected 导出不通过的图片/视频数据（含失败原因） func (c *MaterialVerifyController) ExportRejected(ctx context.Context, req *ExportRejectedReq) (res *ExportRejectedRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) + items, err := serviceDataengine.MaterialVerify.ExportRejectedData(ctx, req.MaterialType) if err != nil { return nil, err @@ -543,8 +546,32 @@ type CallbackRes struct { Msg string `json:"msg"` } +// checkIP 校验请求IP是否在白名单内（context模式） +func checkIP(ctx context.Context) bool { + r := ghttp.RequestFromCtx(ctx) + if r == nil { + return true + } + allowedIPs := g.Cfg().MustGet(ctx, "yidun.callback_allowed_ips", "").String() + if allowedIPs == "" { + return true + } + clientIP := r.GetClientIp() + for _, ip := range strings.Split(allowedIPs, ",") { + if strings.TrimSpace(ip) == clientIP { + return true + } + } + g.Log().Warningf(ctx, "回调IP不在白名单中, clientIP=%s", clientIP) + return false +} + // ImageCallback 图片校验回调 func (c *MaterialVerifyController) ImageCallback(ctx context.Context, req *ImageCallbackReq) (res *CallbackRes, err error) { + if !checkIP(ctx) { + return &CallbackRes{Code: 403, Msg: "IP not allowed"}, nil + } + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "yidun_callback", TenantId: 1}) if req.CallbackData == "" { return &CallbackRes{Code: 400, Msg: "callbackData不能为空"}, nil } @@ -559,6 +586,10 @@ func (c *MaterialVerifyController) ImageCallback(ctx context.Context, req *Image // VideoCallback 视频校验回调 func (c *MaterialVerifyController) VideoCallback(ctx context.Context, req *VideoCallbackReq) (res *CallbackRes, err error) { + if !checkIP(ctx) { + return &CallbackRes{Code: 403, Msg: "IP not allowed"}, nil + } + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "yidun_callback", TenantId: 1}) if req.CallbackData == "" { return &CallbackRes{Code: 400, Msg: "callbackData不能为空"}, nil } @@ -579,6 +610,7 @@ type ResultRes struct { // ImageResult 图片校验结果查询（轮询模式） func (c *MaterialVerifyController) ImageResult(ctx context.Context, req *TaskIDReq) (res *ResultRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) if req.TaskID == "" { return &ResultRes{Code: 400, Msg: "taskId不能为空"}, nil } @@ -593,6 +625,7 @@ func (c *MaterialVerifyController) ImageResult(ctx context.Context, req *TaskIDR // VideoResult 视频校验结果查询（轮询模式） func (c *MaterialVerifyController) VideoResult(ctx context.Context, req *TaskIDReq) (res *ResultRes, err error) { + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) if req.TaskID == "" { return &ResultRes{Code: 400, Msg: "taskId不能为空"}, nil } diff --git a/controller/yidun/content_check_controller.go b/controller/yidun/content_check_controller.go index 7fc74d0..2f0e234 100644 --- a/controller/yidun/content_check_controller.go +++ b/controller/yidun/content_check_controller.go @@ -4,6 +4,7 @@ import ( dto "cid/model/dto/yidun" serviceDataengine "cid/service/dataengine" "context" + "fmt" "gitea.redpowerfuture.com/red-future/common/beans" ) @@ -23,7 +24,7 @@ type StatusRes struct { // Start 启动送检服务 func (c *ContentCheckController) Start(ctx context.Context, req *dto.StartCheckReq) (res *beans.ResponseEmpty, err error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) if serviceDataengine.TencentContentCheck.IsRunning() { return &beans.ResponseEmpty{}, nil @@ -49,14 +50,14 @@ func (c *ContentCheckController) Start(ctx context.Context, req *dto.StartCheckR // Stop 停止送检服务 func (c *ContentCheckController) Stop(ctx context.Context, req *dto.EmptyReq) (res *beans.ResponseEmpty, err error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) serviceDataengine.TencentContentCheck.Stop(ctx) return } // Status 获取送检服务状态 func (c *ContentCheckController) Status(ctx context.Context, req *dto.EmptyReq) (res *StatusRes, err error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) res = &StatusRes{ Running: serviceDataengine.TencentContentCheck.IsRunning(), @@ -68,10 +69,10 @@ func (c *ContentCheckController) Status(ctx context.Context, req *dto.EmptyReq) // ProcessImageCallback 处理图片检测回调 func (c *ContentCheckController) ProcessImageCallback(ctx context.Context, req *dto.ProcessImageCallbackReq) (res *beans.ResponseEmpty, err error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) if req.CallbackData == "" { - return nil, err + return nil, fmt.Errorf("callbackData不能为空") } err = serviceDataengine.TencentContentCallback.ProcessImageCallback(ctx, req.CallbackData) @@ -80,10 +81,10 @@ func (c *ContentCheckController) ProcessImageCallback(ctx context.Context, req * // ProcessVideoCallback 处理视频检测回调 func (c *ContentCheckController) ProcessVideoCallback(ctx context.Context, req *dto.ProcessVideoCallbackReq) (res *beans.ResponseEmpty, err error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) if req.CallbackData == "" { - return nil, err + return nil, fmt.Errorf("callbackData不能为空") } err = serviceDataengine.TencentContentCallback.ProcessVideoCallback(ctx, req.CallbackData) @@ -92,10 +93,10 @@ func (c *ContentCheckController) ProcessVideoCallback(ctx context.Context, req * // ProcessImageResult 查询并处理图片检测结果（轮询模式） func (c *ContentCheckController) ProcessImageResult(ctx context.Context, req *dto.ProcessImageResultReq) (res *beans.ResponseEmpty, err error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) if req.TaskID == "" { - return nil, err + return nil, fmt.Errorf("taskId不能为空") } err = serviceDataengine.TencentContentCallback.ProcessImageResult(ctx, req.TaskID) @@ -104,10 +105,10 @@ func (c *ContentCheckController) ProcessImageResult(ctx context.Context, req *dt // ProcessVideoResult 查询并处理视频检测结果（轮询模式） func (c *ContentCheckController) ProcessVideoResult(ctx context.Context, req *dto.ProcessVideoResultReq) (res *beans.ResponseEmpty, err error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) if req.TaskID == "" { - return nil, err + return nil, fmt.Errorf("taskId不能为空") } err = serviceDataengine.TencentContentCallback.ProcessVideoResult(ctx, req.TaskID) @@ -116,7 +117,7 @@ func (c *ContentCheckController) ProcessVideoResult(ctx context.Context, req *dt // ManualSubmitImageByID 根据图片ID手动提交送检 func (c *ContentCheckController) ManualSubmitImageByID(ctx context.Context, req *dto.ManualSubmitImageByIDReq) (res *dto.ManualSubmitRes, err error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) result, err := serviceDataengine.TencentContentCheck.SubmitImageByID(ctx, req.ImageID) if err != nil { @@ -131,7 +132,7 @@ func (c *ContentCheckController) ManualSubmitImageByID(ctx context.Context, req // ManualSubmitVideoByID 根据视频ID手动提交送检 func (c *ContentCheckController) ManualSubmitVideoByID(ctx context.Context, req *dto.ManualSubmitVideoByIDReq) (res *dto.ManualSubmitRes, err error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) result, err := serviceDataengine.TencentContentCheck.SubmitVideoByID(ctx, req.VideoID) if err != nil { @@ -146,7 +147,7 @@ func (c *ContentCheckController) ManualSubmitVideoByID(ctx context.Context, req // GetImageCheckLogs 获取图片的送检日志 func (c *ContentCheckController) GetImageCheckLogs(ctx context.Context, req *dto.GetImageCheckLogsReq) (res *dto.GetCheckLogsRes, err error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) logs, err := serviceDataengine.TencentContentCallback.GetCheckLogsByImageID(ctx, req.ImageID) if err != nil { @@ -161,7 +162,7 @@ func (c *ContentCheckController) GetImageCheckLogs(ctx context.Context, req *dto // GetVideoCheckLogs 获取视频的送检日志 func (c *ContentCheckController) GetVideoCheckLogs(ctx context.Context, req *dto.GetVideoCheckLogsReq) (res *dto.GetCheckLogsRes, err error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) logs, err := serviceDataengine.TencentContentCallback.GetCheckLogsByVideoID(ctx, req.VideoID) if err != nil { diff --git a/controller/yidun/yidun_callback_controller.go b/controller/yidun/yidun_callback_controller.go index 9f7ecef..9cb5b9c 100644 --- a/controller/yidun/yidun_callback_controller.go +++ b/controller/yidun/yidun_callback_controller.go @@ -4,6 +4,7 @@ import ( dataengineService "cid/service/dataengine" "context" "fmt" + "strings" "gitea.redpowerfuture.com/red-future/common/beans" "github.com/gogf/gf/v2/frame/g" @@ -40,8 +41,14 @@ type PollResult struct { // 易盾回调格式: POST /yidun/callback/receiveImage // Body: callbackData={"antispam":{...}} func (c *YidunCallbackController) ReceiveImageCallback(r *ghttp.Request) { + // IP 白名单校验 + if !checkCallbackIP(r) { + r.Response.WriteJson(CallbackResult{Code: 403, Msg: "IP not allowed"}) + return + } + ctx := r.Context() - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "yidun_callback"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "yidun_callback", TenantId: 1}) // 易盾推送的数据在请求体中 var callbackData string @@ -86,8 +93,14 @@ func (c *YidunCallbackController) ReceiveImageCallback(r *ghttp.Request) { // 易盾回调格式: POST /yidun/callback/receiveVideo // Body: callbackData={"antispam":{...}} func (c *YidunCallbackController) ReceiveVideoCallback(r *ghttp.Request) { + // IP 白名单校验 + if !checkCallbackIP(r) { + r.Response.WriteJson(CallbackResult{Code: 403, Msg: "IP not allowed"}) + return + } + ctx := r.Context() - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "yidun_callback"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "yidun_callback", TenantId: 1}) // 易盾推送的数据在请求体中 var callbackData string @@ -136,7 +149,7 @@ func (c *YidunCallbackController) ReceiveVideoCallback(r *ghttp.Request) { // 格式: POST /yidun/callback/poll func (c *YidunCallbackController) PollAllResults(r *ghttp.Request) { ctx := r.Context() - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) g.Log().Info(ctx, "开始轮询所有待查询的检测结果...") @@ -172,7 +185,7 @@ func (c *YidunCallbackController) PollAllResults(r *ghttp.Request) { // 格式: POST /yidun/callback/pollImage func (c *YidunCallbackController) PollImageResults(r *ghttp.Request) { ctx := r.Context() - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) g.Log().Info(ctx, "开始轮询图片待查询的检测结果...") @@ -198,7 +211,7 @@ func (c *YidunCallbackController) PollImageResults(r *ghttp.Request) { // 格式: POST /yidun/callback/pollVideo func (c *YidunCallbackController) PollVideoResults(r *ghttp.Request) { ctx := r.Context() - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) g.Log().Info(ctx, "开始轮询视频待查询的检测结果...") @@ -224,7 +237,7 @@ func (c *YidunCallbackController) PollVideoResults(r *ghttp.Request) { // 格式: POST /yidun/callback/pollTask func (c *YidunCallbackController) PollByTaskID(r *ghttp.Request) { ctx := r.Context() - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) taskID := r.Get("taskId", "").String() taskType := r.Get("type", "").String() // image 或 video @@ -262,7 +275,7 @@ func (c *YidunCallbackController) PollByTaskID(r *ghttp.Request) { // 格式: GET /yidun/callback/pendingCount func (c *YidunCallbackController) GetPendingCount(r *ghttp.Request) { ctx := r.Context() - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) count, err := dataengineService.MaterialVerify.GetPendingResultsCount(ctx) if err != nil { @@ -287,7 +300,7 @@ func (c *YidunCallbackController) GetPendingCount(r *ghttp.Request) { // 格式: POST /yidun/callback/processImage func (c *YidunCallbackController) ProcessImageCallback(r *ghttp.Request) { ctx := r.Context() - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) var req struct { CallbackData string `json:"callbackData" v:"required#回调数据不能为空"` @@ -311,7 +324,7 @@ func (c *YidunCallbackController) ProcessImageCallback(r *ghttp.Request) { // 格式: POST /yidun/callback/processVideo func (c *YidunCallbackController) ProcessVideoCallback(r *ghttp.Request) { ctx := r.Context() - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) var req struct { CallbackData string `json:"callbackData" v:"required#回调数据不能为空"` @@ -338,3 +351,19 @@ func toString(v interface{}) string { } return "" } + +// checkCallbackIP 校验回调请求IP是否在白名单内 +func checkCallbackIP(r *ghttp.Request) bool { + allowedIPs := g.Cfg().MustGet(r.Context(), "yidun.callback_allowed_ips", "").String() + if allowedIPs == "" { + return true // 未配置白名单，跳过校验 + } + clientIP := r.GetClientIp() + for _, ip := range strings.Split(allowedIPs, ",") { + if strings.TrimSpace(ip) == clientIP { + return true + } + } + g.Log().Warningf(r.Context(), "回调IP不在白名单中, clientIP=%s, allowedIPs=%s", clientIP, allowedIPs) + return false +} diff --git a/controller/yidun/yidun_controller.go b/controller/yidun/yidun_controller.go index 851fff5..88a91b5 100644 --- a/controller/yidun/yidun_controller.go +++ b/controller/yidun/yidun_controller.go @@ -39,7 +39,7 @@ type DetectVideoReq struct { // DetectText 文本检测 func (c *yidunController) DetectText(ctx context.Context, req *DetectTextReq) (string, error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) businessId := g.Cfg().MustGet(ctx, "yidun.text.business_id").String() sdkReq := single.NewTextAsyncCheckRequest(businessId) @@ -57,13 +57,13 @@ func (c *yidunController) DetectText(ctx context.Context, req *DetectTextReq) (s // DetectImage 图片检测 func (c *yidunController) DetectImage(ctx context.Context, req *DetectImageReq) (*yidun.ImageSubmitResult, error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) return yidun.ImageDetection.DetectImage(ctx, req.ImageURL, req.DataID, req.CallbackURL) } // DetectVideo 视频检测 func (c *yidunController) DetectVideo(ctx context.Context, req *DetectVideoReq) (*yidun.VideoSubmitResult, error) { - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) return yidun.VideoDetection.DetectVideo(ctx, req.VideoURL, req.DataID, req.CallbackURL) } @@ -75,8 +75,14 @@ type ImageCallbackResult struct { // ReceiveImageCallback 接收图片检测结果推送 func (c *yidunController) ReceiveImageCallback(r *ghttp.Request) { + // IP 白名单校验 + if !checkCallbackIP(r) { + r.Response.WriteJson(ImageCallbackResult{Code: 403, Msg: "IP not allowed"}) + return + } + ctx := r.Context() - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) callbackData := r.GetForm("callbackData", "").String() if callbackData == "" { @@ -102,8 +108,14 @@ type VideoCallbackResult struct { // ReceiveVideoCallback 接收视频检测结果推送 func (c *yidunController) ReceiveVideoCallback(r *ghttp.Request) { + // IP 白名单校验 + if !checkCallbackIP(r) { + r.Response.WriteJson(VideoCallbackResult{Code: 403, Msg: "IP not allowed"}) + return + } + ctx := r.Context() - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) callbackData := r.GetForm("callbackData", "").String() if callbackData == "" { @@ -124,7 +136,7 @@ func (c *yidunController) ReceiveVideoCallback(r *ghttp.Request) { // GetVideoResult 获取视频检测结果 func (c *yidunController) GetVideoResult(r *ghttp.Request) { ctx := r.Context() - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) taskId := r.Get("taskId", "").String() if taskId == "" { @@ -145,7 +157,7 @@ func (c *yidunController) GetVideoResult(r *ghttp.Request) { // GetImageResult 获取图片检测结果 func (c *yidunController) GetImageResult(r *ghttp.Request) { ctx := r.Context() - ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin"}) + ctx = context.WithValue(ctx, "user", &beans.User{UserName: "admin", TenantId: 1}) taskId := r.Get("taskId", "").String() if taskId == "" { diff --git a/dao/dataengine/material_verify_log_dao.go b/dao/dataengine/material_verify_log_dao.go index 42cff75..8757b99 100644 --- a/dao/dataengine/material_verify_log_dao.go +++ b/dao/dataengine/material_verify_log_dao.go @@ -5,8 +5,9 @@ import ( daoEntity "cid/model/entity/dataengine" "context" + "gitea.redpowerfuture.com/red-future/common/db/gfdb" + "github.com/bwmarrin/snowflake" "github.com/gogf/gf/v2/frame/g" - "github.com/gogf/gf/v2/os/gtime" ) // MaterialVerifyLogDAO 素材校验日志数据访问层 @@ -19,9 +20,19 @@ var MaterialVerifyLog = new(MaterialVerifyLogDAO) const MaterialVerifyLogTable = "material_verify_log" // Create 创建校验日志 -func (d *MaterialVerifyLogDAO) Create(ctx context.Context, log *daoEntity.MaterialVerifyLog) (int64, error) { - // 构建插入数据，排除主键Id（让数据库自增） - data := g.Map{ +func (d *MaterialVerifyLogDAO) Create(ctx context.Context, log *daoEntity.MaterialVerifyLog) (id int64, err error) { + // GoFrame v2.10.0 pgsql 驱动不支持 RETURNING/LastInsertId + // 且 gfdb insertHook 会覆盖 id，无法从外部获取钩子生成的 ID + // 手动生成 Snowflake ID 并直接插入（绕过 gfdb 钩子） + node, err := snowflake.NewNode(1) + if err != nil { + g.Log().Errorf(ctx, "创建Snowflake节点失败: %v", err) + return 0, err + } + snowflakeID := node.Generate().Int64() + + _, err = g.DB("cid").Model(MaterialVerifyLogTable).Data(g.Map{ + "id": snowflakeID, "tenant_id": log.TenantID, "material_type": log.MaterialType, "material_id": log.MaterialID, @@ -29,21 +40,18 @@ func (d *MaterialVerifyLogDAO) Create(ctx context.Context, log *daoEntity.Materi "source_id": log.SourceID, "account_id": log.AccountID, "verify_status": log.VerifyStatus, - "created_at": gtime.Now(), - } - result, err := g.DB("default").Model(MaterialVerifyLogTable).Data(data).Insert() + }).Insert() if err != nil { g.Log().Errorf(ctx, "创建校验日志失败: %v", err) return 0, err } - id, _ := result.LastInsertId() - return id, nil + return snowflakeID, nil } // GetByID 根据ID获取日志 func (d *MaterialVerifyLogDAO) GetByID(ctx context.Context, id int64) (*daoEntity.MaterialVerifyLog, error) { var result daoEntity.MaterialVerifyLog - r, err := g.DB("default").Model(MaterialVerifyLogTable). + r, err := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable). Where(daoEntity.MaterialVerifyLogCols.Id, id). One() if err != nil { @@ -61,7 +69,7 @@ func (d *MaterialVerifyLogDAO) GetByID(ctx context.Context, id int64) (*daoEntit // GetByTaskID 根据任务ID获取日志 func (d *MaterialVerifyLogDAO) GetByTaskID(ctx context.Context, taskID string) (*daoEntity.MaterialVerifyLog, error) { var result daoEntity.MaterialVerifyLog - r, err := g.DB("default").Model(MaterialVerifyLogTable). + r, err := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable). Where(daoEntity.MaterialVerifyLogCols.TaskID, taskID). One() if err != nil { @@ -79,7 +87,7 @@ func (d *MaterialVerifyLogDAO) GetByTaskID(ctx context.Context, taskID string) ( // GetByMaterialID 根据素材ID获取日志列表 func (d *MaterialVerifyLogDAO) GetByMaterialID(ctx context.Context, materialID string) ([]daoEntity.MaterialVerifyLog, error) { var result []daoEntity.MaterialVerifyLog - r, err := g.DB("default").Model(MaterialVerifyLogTable). + r, err := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable). Where(daoEntity.MaterialVerifyLogCols.MaterialID, materialID). OrderDesc(daoEntity.MaterialVerifyLogCols.CreatedAt). All() @@ -95,7 +103,7 @@ func (d *MaterialVerifyLogDAO) GetByMaterialID(ctx context.Context, materialID s // GetBySource 根据来源获取日志 func (d *MaterialVerifyLogDAO) GetBySource(ctx context.Context, sourceTable string, sourceID int64) ([]daoEntity.MaterialVerifyLog, error) { var result []daoEntity.MaterialVerifyLog - r, err := g.DB("default").Model(MaterialVerifyLogTable). + r, err := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable). Where(daoEntity.MaterialVerifyLogCols.SourceTable, sourceTable). Where(daoEntity.MaterialVerifyLogCols.SourceID, sourceID). OrderDesc(daoEntity.MaterialVerifyLogCols.CreatedAt). @@ -111,9 +119,9 @@ func (d *MaterialVerifyLogDAO) GetBySource(ctx context.Context, sourceTable stri // UpdateVerifyResult 更新校验结果 func (d *MaterialVerifyLogDAO) UpdateVerifyResult(ctx context.Context, id int64, verifyStatus string, suggestion, label, resultType int, responseResult string, checkTime int64) error { - _, err := g.DB("default").Model(MaterialVerifyLogTable). + _, err := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable). Where(daoEntity.MaterialVerifyLogCols.Id, id). - Data(map[string]interface{}{ + Data(g.Map{ daoEntity.MaterialVerifyLogCols.VerifyStatus: verifyStatus, daoEntity.MaterialVerifyLogCols.Suggestion: suggestion, daoEntity.MaterialVerifyLogCols.Label: label, @@ -130,9 +138,9 @@ func (d *MaterialVerifyLogDAO) UpdateVerifyResult(ctx context.Context, id int64, // UpdateError 更新错误信息 func (d *MaterialVerifyLogDAO) UpdateError(ctx context.Context, id int64, verifyStatus string, errorMsg string) error { - _, err := g.DB("default").Model(MaterialVerifyLogTable). + _, err := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable). Where(daoEntity.MaterialVerifyLogCols.Id, id). - Data(map[string]interface{}{ + Data(g.Map{ daoEntity.MaterialVerifyLogCols.VerifyStatus: verifyStatus, daoEntity.MaterialVerifyLogCols.ErrorMsg: errorMsg, }).Update() @@ -145,9 +153,9 @@ func (d *MaterialVerifyLogDAO) UpdateError(ctx context.Context, id int64, verify // UpdateTaskID 更新任务ID func (d *MaterialVerifyLogDAO) UpdateTaskID(ctx context.Context, id int64, taskID string) error { - _, err := g.DB("default").Model(MaterialVerifyLogTable). + _, err := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable). Where(daoEntity.MaterialVerifyLogCols.Id, id). - Data(map[string]interface{}{ + Data(g.Map{ daoEntity.MaterialVerifyLogCols.TaskID: taskID, }).Update() if err != nil { @@ -158,9 +166,9 @@ func (d *MaterialVerifyLogDAO) UpdateTaskID(ctx context.Context, id int64, taskI // UpdateDuration 更新处理耗时 func (d *MaterialVerifyLogDAO) UpdateDuration(ctx context.Context, id int64, durationMs int64) error { - _, err := g.DB("default").Model(MaterialVerifyLogTable). + _, err := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable). Where(daoEntity.MaterialVerifyLogCols.Id, id). - Data(map[string]interface{}{ + Data(g.Map{ daoEntity.MaterialVerifyLogCols.DurationMs: durationMs, }).Update() if err != nil { @@ -171,9 +179,9 @@ func (d *MaterialVerifyLogDAO) UpdateDuration(ctx context.Context, id int64, dur // UpdateRequestParams 更新请求参数 func (d *MaterialVerifyLogDAO) UpdateRequestParams(ctx context.Context, id int64, requestParams string) error { - _, err := g.DB("default").Model(MaterialVerifyLogTable). + _, err := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable). Where(daoEntity.MaterialVerifyLogCols.Id, id). - Data(map[string]interface{}{ + Data(g.Map{ daoEntity.MaterialVerifyLogCols.RequestParams: requestParams, }).Update() if err != nil { @@ -185,18 +193,18 @@ func (d *MaterialVerifyLogDAO) UpdateRequestParams(ctx context.Context, id int64 // GetByCondition 根据条件分页查询 func (d *MaterialVerifyLogDAO) GetByCondition(ctx context.Context, condition map[string]interface{}, page, pageSize int) ([]daoEntity.MaterialVerifyLog, int, error) { var result []daoEntity.MaterialVerifyLog - model := g.DB("default").Model(MaterialVerifyLogTable) + m := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable) for k, v := range condition { - model = model.Where(k, v) + m.Where(k, v) } - total, err := model.Count() + total, err := m.Count() if err != nil { return nil, 0, err } - r, err := model. + r, err := m. OrderDesc(daoEntity.MaterialVerifyLogCols.CreatedAt). Page(page, pageSize). All() @@ -211,7 +219,7 @@ func (d *MaterialVerifyLogDAO) GetByCondition(ctx context.Context, condition map // CountByStatus 按状态统计 func (d *MaterialVerifyLogDAO) CountByStatus(ctx context.Context, verifyStatus string) (int, error) { - count, err := g.DB("default").Model(MaterialVerifyLogTable). + count, err := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable). Where(daoEntity.MaterialVerifyLogCols.VerifyStatus, verifyStatus). Count() if err != nil { @@ -224,7 +232,6 @@ func (d *MaterialVerifyLogDAO) CountByStatus(ctx context.Context, verifyStatus s func (d *MaterialVerifyLogDAO) GetStats(ctx context.Context) (map[string]int, error) { stats := make(map[string]int) - // 使用实体中定义的正确状态值：PENDING=待校验, VERIFIED=校验通过, REJECTED=校验不通过 statuses := []struct { statusKey string statusVal string @@ -244,18 +251,15 @@ func (d *MaterialVerifyLogDAO) GetStats(ctx context.Context) (map[string]int, er totalCount += count } - // 添加总计 stats["total"] = totalCount - return stats, nil } -// GetPendingResults 获取待查询结果的日志（状态为submitting且有taskID） +// GetPendingResults 获取待查询结果的日志（状态为pending且有taskID） func (d *MaterialVerifyLogDAO) GetPendingResults(ctx context.Context, limit int) ([]daoEntity.MaterialVerifyLog, error) { var result []daoEntity.MaterialVerifyLog - // 查询状态为 pending 且有 task_id 的记录 - r, err := g.DB("default").Model(MaterialVerifyLogTable). + r, err := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable). Where(daoEntity.MaterialVerifyLogCols.VerifyStatus, consts.CheckStatusPending). WhereNotNull(daoEntity.MaterialVerifyLogCols.TaskID). Where(daoEntity.MaterialVerifyLogCols.TaskID + " != ''"). @@ -276,7 +280,7 @@ func (d *MaterialVerifyLogDAO) GetPendingResults(ctx context.Context, limit int) // GetLastRejectedLogByMaterialID 根据素材ID获取最后一条失败的校验日志 func (d *MaterialVerifyLogDAO) GetLastRejectedLogByMaterialID(ctx context.Context, materialID string, verifyStatus string) (*daoEntity.MaterialVerifyLog, error) { var result daoEntity.MaterialVerifyLog - r, err := g.DB("default").Model(MaterialVerifyLogTable). + r, err := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable). Where(daoEntity.MaterialVerifyLogCols.MaterialID, materialID). Where(daoEntity.MaterialVerifyLogCols.VerifyStatus, verifyStatus). OrderDesc(daoEntity.MaterialVerifyLogCols.CreatedAt). @@ -295,7 +299,7 @@ func (d *MaterialVerifyLogDAO) GetLastRejectedLogByMaterialID(ctx context.Contex // CountPendingResults 统计待查询结果的数量 func (d *MaterialVerifyLogDAO) CountPendingResults(ctx context.Context) (int, error) { - count, err := g.DB("default").Model(MaterialVerifyLogTable). + count, err := gfdb.DB(ctx, "cid").Model(ctx, MaterialVerifyLogTable). Where(daoEntity.MaterialVerifyLogCols.VerifyStatus, consts.CheckStatusPending). WhereNotNull(daoEntity.MaterialVerifyLogCols.TaskID). Where(daoEntity.MaterialVerifyLogCols.TaskID + " != ''"). diff --git a/dao/dataengine/tencent_content_check_log_dao.go b/dao/dataengine/tencent_content_check_log_dao.go index 58c68ad..c394904 100644 --- a/dao/dataengine/tencent_content_check_log_dao.go +++ b/dao/dataengine/tencent_content_check_log_dao.go @@ -7,6 +7,8 @@ import ( "context" "encoding/json" + "gitea.redpowerfuture.com/red-future/common/db/gfdb" + "github.com/bwmarrin/snowflake" "github.com/gogf/gf/v2/frame/g" ) @@ -17,21 +19,43 @@ type TencentContentCheckLogDAO struct{} var TencentContentCheckLog = new(TencentContentCheckLogDAO) // Create 创建送检日志 -func (d *TencentContentCheckLogDAO) Create(ctx context.Context, log *entity.TencentContentCheckLog) (int64, error) { - r, err := g.DB("default").Model(consts.TencentContentCheckLogTable).Data(log).Insert() +func (d *TencentContentCheckLogDAO) Create(ctx context.Context, log *entity.TencentContentCheckLog) (id int64, err error) { + // GoFrame v2.10.0 pgsql 驱动不支持 RETURNING/LastInsertId + node, err := snowflake.NewNode(1) + if err != nil { + g.Log().Errorf(ctx, "创建Snowflake节点失败: %v", err) + return 0, err + } + snowflakeID := node.Generate().Int64() + + _, err = g.DB("cid").Model(consts.TencentContentCheckLogTable).Data(g.Map{ + "id": snowflakeID, + "source_table": log.SourceTable, + "source_id": log.SourceID, + "request_url": log.RequestURL, + "request_param": log.RequestParam, + "response_data": log.ResponseData, + "status": log.Status, + "check_time": log.CheckTime, + "fail_reason": log.FailReason, + "task_id": log.TaskID, + "suggestion": log.Suggestion, + "label": log.Label, + "result_type": log.ResultType, + "duration": log.Duration, + }).Insert() if err != nil { g.Log().Errorf(ctx, "创建送检日志失败: %v", err) return 0, err } - id, _ := r.LastInsertId() - return id, nil + return snowflakeID, nil } // UpdateStatus 更新送检状态 func (d *TencentContentCheckLogDAO) UpdateStatus(ctx context.Context, id int64, status string, responseData string, failReason string) error { - _, err := g.DB("default").Model(consts.TencentContentCheckLogTable). + _, err := gfdb.DB(ctx, "cid").Model(ctx, consts.TencentContentCheckLogTable). Where("id", id). - Data(map[string]interface{}{ + Data(g.Map{ "status": status, "response_data": responseData, "fail_reason": failReason, @@ -41,9 +65,9 @@ func (d *TencentContentCheckLogDAO) UpdateStatus(ctx context.Context, id int64, // UpdateCheckResult 更新检测结果 func (d *TencentContentCheckLogDAO) UpdateCheckResult(ctx context.Context, id int64, suggestion, label, resultType int, checkTime int64) error { - _, err := g.DB("default").Model(consts.TencentContentCheckLogTable). + _, err := gfdb.DB(ctx, "cid").Model(ctx, consts.TencentContentCheckLogTable). Where("id", id). - Data(map[string]interface{}{ + Data(g.Map{ "status": consts.CheckStatusCompleted, "suggestion": suggestion, "label": label, @@ -56,7 +80,7 @@ func (d *TencentContentCheckLogDAO) UpdateCheckResult(ctx context.Context, id in // GetByID 根据ID获取日志 func (d *TencentContentCheckLogDAO) GetByID(ctx context.Context, id int64) (*entity.TencentContentCheckLog, error) { var result entity.TencentContentCheckLog - r, err := g.DB("default").Model(consts.TencentContentCheckLogTable). + r, err := gfdb.DB(ctx, "cid").Model(ctx, consts.TencentContentCheckLogTable). Where("id", id). One() if err != nil { @@ -74,7 +98,7 @@ func (d *TencentContentCheckLogDAO) GetByID(ctx context.Context, id int64) (*ent // GetBySourceID 根据来源ID获取日志 func (d *TencentContentCheckLogDAO) GetBySourceID(ctx context.Context, sourceTable string, sourceID int64) ([]entity.TencentContentCheckLog, error) { var result []entity.TencentContentCheckLog - r, err := g.DB("default").Model(consts.TencentContentCheckLogTable). + r, err := gfdb.DB(ctx, "cid").Model(ctx, consts.TencentContentCheckLogTable). Where("source_table", sourceTable). Where("source_id", sourceID). OrderDesc("created_at"). @@ -91,7 +115,7 @@ func (d *TencentContentCheckLogDAO) GetBySourceID(ctx context.Context, sourceTab // GetByTaskID 根据任务ID获取日志 func (d *TencentContentCheckLogDAO) GetByTaskID(ctx context.Context, taskID string) (*entity.TencentContentCheckLog, error) { var result entity.TencentContentCheckLog - r, err := g.DB("default").Model(consts.TencentContentCheckLogTable). + r, err := gfdb.DB(ctx, "cid").Model(ctx, consts.TencentContentCheckLogTable). Where("task_id", taskID). One() if err != nil { @@ -109,18 +133,18 @@ func (d *TencentContentCheckLogDAO) GetByTaskID(ctx context.Context, taskID stri // ListByStatus 根据状态获取日志列表 func (d *TencentContentCheckLogDAO) ListByStatus(ctx context.Context, status string, page, pageSize int) ([]entity.TencentContentCheckLog, int, error) { var result []entity.TencentContentCheckLog - model := g.DB("default").Model(consts.TencentContentCheckLogTable) + m := gfdb.DB(ctx, "cid").Model(ctx, consts.TencentContentCheckLogTable) if status != "" { - model = model.Where("status", status) + m.Where("status", status) } - total, err := model.Count() + total, err := m.Count() if err != nil { return nil, 0, err } - r, err := model. + r, err := m. OrderDesc("created_at"). Page(page, pageSize). All() @@ -135,7 +159,7 @@ func (d *TencentContentCheckLogDAO) ListByStatus(ctx context.Context, status str // UpdateDuration 更新耗时 func (d *TencentContentCheckLogDAO) UpdateDuration(ctx context.Context, id int64, duration int64) error { - _, err := g.DB("default").Model(consts.TencentContentCheckLogTable). + _, err := gfdb.DB(ctx, "cid").Model(ctx, consts.TencentContentCheckLogTable). Where("id", id). Data("duration", duration). Update() @@ -144,14 +168,14 @@ func (d *TencentContentCheckLogDAO) UpdateDuration(ctx context.Context, id int64 // UpdateTaskID 更新任务ID func (d *TencentContentCheckLogDAO) UpdateTaskID(ctx context.Context, id int64, taskID string) error { - _, err := g.DB("default").Model(consts.TencentContentCheckLogTable). + _, err := gfdb.DB(ctx, "cid").Model(ctx, consts.TencentContentCheckLogTable). Where("id", id). Data("task_id", taskID). Update() return err } -// GetSubmitResult 获取图片提交结果 +// GetImageSubmitResult 获取图片提交结果 func (d *TencentContentCheckLogDAO) GetImageSubmitResult(ctx context.Context, id int64) (*yidunService.ImageSubmitResult, error) { log, err := d.GetByID(ctx, id) if err != nil || log == nil { diff --git a/go.mod b/go.mod index 939a71e..aa89c98 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.26.0 require ( gitea.redpowerfuture.com/red-future/common v0.0.29 + github.com/bwmarrin/snowflake v0.3.0 github.com/gogf/gf/contrib/drivers/pgsql/v2 v2.10.2 github.com/gogf/gf/contrib/nosql/redis/v2 v2.10.2 github.com/gogf/gf/v2 v2.10.2 diff --git a/go.sum b/go.sum index 969c904..a820c89 100644 --- a/go.sum +++ b/go.sum @@ -1,6 +1,5 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw= -gitea.redpowerfuture.com/red-future/common v0.0.23 h1:xieoA00iKOCDm5SO9iXn+cSyMKBAlZwI0fuEVPWrHLg= -gitea.redpowerfuture.com/red-future/common v0.0.23/go.mod h1:50U1Xi+Ie56z09S5LQbZvaken0Mxv3OeS9LgR7U/ZRY= +gitea.redpowerfuture.com/red-future/common v0.0.29 h1:5McaN5pSewvrLUHQzWMX6EaUvD+B5I5bMYoU+clHJk4= gitea.redpowerfuture.com/red-future/common v0.0.29/go.mod h1:50U1Xi+Ie56z09S5LQbZvaken0Mxv3OeS9LgR7U/ZRY= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/toml v1.6.0 h1:dRaEfpa2VI55EwlIW72hMRHdWouJeRF7TPYhI+AUQjk= @@ -24,6 +23,8 @@ github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/bwmarrin/snowflake v0.3.0 h1:xm67bEhkKh6ij1790JB83OujPR5CzNe8QuQqAgISZN0= +github.com/bwmarrin/snowflake v0.3.0/go.mod h1:NdZxfVWX+oR6y2K0o6qAYv6gIOP9rjG0/E9WsDpxqwE= github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= @@ -146,6 +147,7 @@ github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l8 github.com/hashicorp/consul/api v1.33.5 h1:Nn6q87zudRU1rLBTJEgaWxz9STCNadilLCD7B8OA5aI= github.com/hashicorp/consul/api v1.33.5/go.mod h1:pa6fJOSHKLOzNHpUVeqLDtxA5+J1D7NNzLasuk8eRXA= github.com/hashicorp/consul/sdk v0.17.3 h1:oZMMxzQGSsiT+ToOH50y3Qcs0nc9Ud+7L5lRx+EmMU0= +github.com/hashicorp/consul/sdk v0.17.3/go.mod h1:jnOmYjiNfVRpBaujQ1DFFVs0N6g3S1y6wygSjLTzYfc= github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I= github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= @@ -277,6 +279,7 @@ github.com/r3labs/diff/v2 v2.15.1/go.mod h1:I8noH9Fc2fjSaMxqF3G2lhDdC0b+JXCfyx85 github.com/redis/go-redis/v9 v9.21.0 h1:FPBE4hhbAke+TLmcY3WkpbDffJEomdqPn3HYiqAtL9E= github.com/redis/go-redis/v9 v9.21.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts= github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUtVbo7ada43DJhG55ua/hjS5I= github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc= @@ -314,6 +317,7 @@ github.com/yidun/yidun-golang-sdk v1.0.38/go.mod h1:+JGdWbkUvLi9uKTtHI+nrxajulfZ github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.mongodb.org/mongo-driver/v2 v2.6.0 h1:b9sJOYrkmt4l8bY43ZenFBcPlhYIjaOfYHLtbB/5qi8= go.mongodb.org/mongo-driver/v2 v2.6.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0= @@ -331,6 +335,7 @@ go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= go.opentelemetry.io/proto/otlp v1.7.1 h1:gTOMpGDb0WTBOP8JaO72iL3auEZhVmAQg4ipjOVAtj4= @@ -354,6 +359,7 @@ golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHl golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ= +golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0= golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= @@ -381,6 +387,7 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= +golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -423,6 +430,7 @@ golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtn golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.46.0 h1:7jTurBkPZu4moS/Uy4OQT1M+QBlsj3wejyZwsT8Z7rk= +golang.org/x/tools v0.46.0/go.mod h1:FrD85F8l+NWL+9XWBSyVSHO6Ne4jutsfIFba7AWQ5Ys= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/model/entity/dataengine/material_verify_log.go b/model/entity/dataengine/material_verify_log.go index 4f32681..de7aff1 100644 --- a/model/entity/dataengine/material_verify_log.go +++ b/model/entity/dataengine/material_verify_log.go @@ -80,6 +80,7 @@ const ( // 校验状态常量 const ( VerifyStatusPending = "PENDING" // 待校验 + VerifyStatusReview = "REVIEW" // 嫌疑，需人工复核 VerifyStatusVerified = "VERIFIED" // 校验通过 VerifyStatusRejected = "REJECTED" // 校验不通过 ) diff --git a/service/dataengine/material_verify_service.go b/service/dataengine/material_verify_service.go index f115c8f..9ff9f6a 100644 --- a/service/dataengine/material_verify_service.go +++ b/service/dataengine/material_verify_service.go @@ -10,6 +10,7 @@ import ( "fmt" "time" + "gitea.redpowerfuture.com/red-future/common/beans" "github.com/gogf/gf/v2/frame/g" ) @@ -41,7 +42,7 @@ func SuggestionToVerifyStatus(suggestion int) string { case consts.SuggestionPass: return entity.VerifyStatusVerified // 通过 case consts.SuggestionReview: - return entity.VerifyStatusPending // 嫌疑，需要人工审核，暂不更新状态 + return entity.VerifyStatusReview // 嫌疑，需人工复核 case consts.SuggestionBlock: return entity.VerifyStatusRejected // 不通过 default: @@ -55,22 +56,28 @@ func SuggestionToVerifyStatus(suggestion int) string { // VerifyImageByID 根据图片ID执行校验 func (s *MaterialVerifyService) VerifyImageByID(ctx context.Context, imageID string) (*entity.MaterialVerifyLog, error) { - // 1. 获取图片数据 image, err := dao.TencentImage.GetByImageID(ctx, imageID) if err != nil { - return nil, fmt.Errorf("查询图片数据失败: %w", err) + return nil, fmt.Errorf("unknown error: %w", err) } if image == nil { return nil, fmt.Errorf("未找到图片数据, imageID=%s", imageID) } - // 2. 创建校验日志 + // 幂等性检查：如果已在送检中，直接返回已有日志 + if image.VerifyStatus == consts.CheckStatusSubmitting { + logs, err := dao.MaterialVerifyLog.GetByMaterialID(ctx, imageID) + if err == nil && len(logs) > 0 { + g.Log().Infof(ctx, "图片已在送检中, imageID=%s, logId=%d", imageID, logs[0].Id) + return &logs[0], nil + } + } + log := s.createVerifyLog(ctx, entity.MaterialTypeImage, imageID, consts.SourceTableTencentImage, image.Id, image.AccountID) if log == nil { return nil, fmt.Errorf("创建校验日志失败") } - // 3. 执行校验 err = s.submitImageCheck(ctx, image, log) if err != nil { return nil, err @@ -83,10 +90,8 @@ func (s *MaterialVerifyService) VerifyImageByID(ctx context.Context, imageID str func (s *MaterialVerifyService) submitImageCheck(ctx context.Context, image *entity.TencentImage, log *entity.MaterialVerifyLog) error { startTime := time.Now() - // 获取回调模式开关 callbackMode := g.Cfg().MustGet(ctx, "yidun.callback_mode").Bool() - // 构建请求参数 requestParams := map[string]interface{}{ "imageURL": image.PreviewURL, "dataID": image.ImageID, @@ -99,7 +104,6 @@ func (s *MaterialVerifyService) submitImageCheck(ctx context.Context, image *ent ) if callbackMode { - // 回调模式：使用异步检测，易盾处理完成后会回调 callbackURL := g.Cfg().MustGet(ctx, "yidun.image.callback_url").String() requestParams["callbackURL"] = callbackURL @@ -109,35 +113,30 @@ func (s *MaterialVerifyService) submitImageCheck(ctx context.Context, image *ent dao.MaterialVerifyLog.UpdateError(ctx, log.Id, entity.VerifyStatusPending, err.Error()) dao.MaterialVerifyLog.UpdateDuration(ctx, log.Id, duration) g.Log().Warningf(ctx, "图片异步检测失败（保持待检验）, id=%d, imageId=%s, error=%v", image.Id, image.ImageID, err) - return fmt.Errorf("图片异步检测失败: %w", err) + return fmt.Errorf("unknown error: %w", err) } taskID = result.TaskID - // 保存任务ID和请求参数 dao.MaterialVerifyLog.UpdateTaskID(ctx, log.Id, taskID) dao.MaterialVerifyLog.UpdateRequestParams(ctx, log.Id, string(requestParamsJSON)) - // 更新原表状态为 submitting（等待回调） s.updateImageStatus(ctx, image.Id, StatusSubmitting) g.Log().Infof(ctx, "图片异步检测已提交, id=%d, imageId=%s, taskId=%s, duration=%dms", image.Id, image.ImageID, taskID, duration) } else { - // 轮询模式：使用同步检测，直接返回结果 syncResult, err := yidunService.ImageDetection.DetectImageSync(ctx, image.PreviewURL, image.ImageID) duration = time.Since(startTime).Milliseconds() if err != nil { dao.MaterialVerifyLog.UpdateError(ctx, log.Id, entity.VerifyStatusPending, err.Error()) dao.MaterialVerifyLog.UpdateDuration(ctx, log.Id, duration) g.Log().Warningf(ctx, "图片同步检测失败（保持待检验）, id=%d, imageId=%s, error=%v", image.Id, image.ImageID, err) - return fmt.Errorf("图片同步检测失败: %w", err) + return fmt.Errorf("unknown error: %w", err) } taskID = syncResult.TaskID - // 保存任务ID和请求参数 dao.MaterialVerifyLog.UpdateTaskID(ctx, log.Id, taskID) dao.MaterialVerifyLog.UpdateRequestParams(ctx, log.Id, string(requestParamsJSON)) - // 根据同步结果更新状态 verifyStatus := SuggestionToVerifyStatus(syncResult.Suggestion) responseJSON, _ := json.Marshal(syncResult) dao.MaterialVerifyLog.UpdateVerifyResult(ctx, log.Id, verifyStatus, @@ -157,22 +156,28 @@ func (s *MaterialVerifyService) submitImageCheck(ctx context.Context, image *ent // VerifyVideoByID 根据视频ID执行校验 func (s *MaterialVerifyService) VerifyVideoByID(ctx context.Context, videoID string) (*entity.MaterialVerifyLog, error) { - // 1. 获取视频数据 video, err := dao.TencentVideo.GetByVideoID(ctx, videoID) if err != nil { - return nil, fmt.Errorf("查询视频数据失败: %w", err) + return nil, fmt.Errorf("unknown error: %w", err) } if video == nil { return nil, fmt.Errorf("未找到视频数据, videoID=%s", videoID) } - // 2. 创建校验日志 + // 幂等性检查：如果已在送检中，直接返回已有日志 + if video.VerifyStatus == consts.CheckStatusSubmitting { + logs, err := dao.MaterialVerifyLog.GetByMaterialID(ctx, videoID) + if err == nil && len(logs) > 0 { + g.Log().Infof(ctx, "视频已在送检中, videoID=%s, logId=%d", videoID, logs[0].Id) + return &logs[0], nil + } + } + log := s.createVerifyLog(ctx, entity.MaterialTypeVideo, videoID, consts.SourceTableTencentVideo, video.Id, video.AccountID) if log == nil { return nil, fmt.Errorf("创建校验日志失败") } - // 3. 执行校验 err = s.submitVideoCheck(ctx, video, log) if err != nil { return nil, err @@ -185,16 +190,13 @@ func (s *MaterialVerifyService) VerifyVideoByID(ctx context.Context, videoID str func (s *MaterialVerifyService) submitVideoCheck(ctx context.Context, video *entity.TencentVideo, log *entity.MaterialVerifyLog) error { startTime := time.Now() - // 获取回调模式开关 callbackMode := g.Cfg().MustGet(ctx, "yidun.callback_mode").Bool() - // 根据开关决定回调地址 var callbackURL string if callbackMode { callbackURL = g.Cfg().MustGet(ctx, "yidun.video.callback_url").String() } - // 构建请求参数 requestParams := map[string]interface{}{ "videoURL": video.PreviewURL, "dataID": video.VideoID, @@ -202,30 +204,22 @@ func (s *MaterialVerifyService) submitVideoCheck(ctx context.Context, video *ent } requestParamsJSON, _ := json.Marshal(requestParams) - // 调用易盾视频检测 result, err := yidunService.VideoDetection.DetectVideo(ctx, video.PreviewURL, video.VideoID, callbackURL) duration := time.Since(startTime).Milliseconds() if err != nil { - // 调用易盾接口失败（如额度用光、网络错误、超时等），不更新状态，保持待检验 - // 只有易盾明确返回检测结果且suggestion=BLOCK时才标记为失败 dao.MaterialVerifyLog.UpdateError(ctx, log.Id, entity.VerifyStatusPending, err.Error()) dao.MaterialVerifyLog.UpdateDuration(ctx, log.Id, duration) g.Log().Warningf(ctx, "视频校验接口调用失败（保持待检验）, id=%d, videoId=%s, error=%v", video.Id, video.VideoID, err) - return fmt.Errorf("视频校验调用失败: %w", err) + return fmt.Errorf("unknown error: %w", err) } - // 保存任务ID和请求参数 dao.MaterialVerifyLog.UpdateTaskID(ctx, log.Id, result.TaskID) dao.MaterialVerifyLog.UpdateRequestParams(ctx, log.Id, string(requestParamsJSON)) - - // 更新原表状态为 submitting s.updateVideoStatus(ctx, video.Id, StatusSubmitting) - // 轮询模式（无回调）：提交后立即尝试查询检测结果 if !callbackMode { g.Log().Infof(ctx, "轮询模式：提交后立即查询结果, taskId=%s", result.TaskID) - // 等待500ms让易盾有时间处理 time.Sleep(500 * time.Millisecond) if err := s.ProcessVideoResultByTask(ctx, result.TaskID); err != nil { g.Log().Warningf(ctx, "提交后立即查询结果失败（不影响状态，后续轮询继续）, taskId=%s, error=%v", result.TaskID, err) @@ -249,7 +243,7 @@ func (s *MaterialVerifyService) ProcessImageCallback(ctx context.Context, callba var callback yidunService.ImageCallbackData if err := json.Unmarshal([]byte(callbackData), &callback); err != nil { g.Log().Errorf(ctx, "解析图片回调数据失败: %v", err) - return fmt.Errorf("解析回调数据失败: %w", err) + return fmt.Errorf("unknown error: %w", err) } if callback.Antispam == nil { @@ -260,30 +254,23 @@ func (s *MaterialVerifyService) ProcessImageCallback(ctx context.Context, callba g.Log().Infof(ctx, "处理图片校验结果 - taskId: %s, suggestion: %d, resultType: %d", antispam.TaskId, antispam.Suggestion, antispam.ResultType) - // 根据 taskId 查找校验日志 log, err := dao.MaterialVerifyLog.GetByTaskID(ctx, antispam.TaskId) if err != nil { - return fmt.Errorf("查找校验日志失败: %w", err) + return fmt.Errorf("unknown error: %w", err) } if log == nil { g.Log().Warningf(ctx, "未找到校验日志, taskId=%s", antispam.TaskId) return nil } - // 构建响应结果 - responseResult := callbackData - - // 根据 suggestion 确定校验状态 verifyStatus := SuggestionToVerifyStatus(antispam.Suggestion) - // 更新日志 err = dao.MaterialVerifyLog.UpdateVerifyResult(ctx, log.Id, verifyStatus, - antispam.Suggestion, antispam.Label, antispam.ResultType, responseResult, antispam.CensorTime) + antispam.Suggestion, antispam.Label, antispam.ResultType, callbackData, antispam.CensorTime) if err != nil { - return fmt.Errorf("更新校验日志失败: %w", err) + return fmt.Errorf("unknown error: %w", err) } - // 更新原表状态（图片回调只处理图片来源） if log.SourceTable == consts.SourceTableTencentImage { s.updateImageStatus(ctx, log.SourceID, verifyStatus) } @@ -301,7 +288,7 @@ func (s *MaterialVerifyService) ProcessVideoCallback(ctx context.Context, callba var callback yidunService.VideoCallbackData if err := json.Unmarshal([]byte(callbackData), &callback); err != nil { g.Log().Errorf(ctx, "解析视频回调数据失败: %v", err) - return fmt.Errorf("解析回调数据失败: %w", err) + return fmt.Errorf("unknown error: %w", err) } if callback.Antispam == nil { @@ -312,36 +299,28 @@ func (s *MaterialVerifyService) ProcessVideoCallback(ctx context.Context, callba g.Log().Infof(ctx, "处理视频校验结果 - taskId: %s, suggestion: %d, resultType: %d", antispam.TaskID, antispam.Suggestion, antispam.ResultType) - // 根据 taskId 查找校验日志 log, err := dao.MaterialVerifyLog.GetByTaskID(ctx, antispam.TaskID) if err != nil { - return fmt.Errorf("查找校验日志失败: %w", err) + return fmt.Errorf("unknown error: %w", err) } if log == nil { g.Log().Warningf(ctx, "未找到校验日志, taskId=%s", antispam.TaskID) return nil } - // 构建响应结果 - responseResult := callbackData - - // 根据 suggestion 确定校验状态 verifyStatus := SuggestionToVerifyStatus(antispam.Suggestion) - // 审核时间 checkTime := antispam.CensorTime if checkTime == 0 { checkTime = antispam.CheckTime } - // 更新日志 err = dao.MaterialVerifyLog.UpdateVerifyResult(ctx, log.Id, verifyStatus, - antispam.Suggestion, antispam.Label, antispam.ResultType, responseResult, checkTime) + antispam.Suggestion, antispam.Label, antispam.ResultType, callbackData, checkTime) if err != nil { - return fmt.Errorf("更新校验日志失败: %w", err) + return fmt.Errorf("unknown error: %w", err) } - // 更新原表状态（视频回调只处理视频来源） if log.SourceTable == consts.SourceTableTencentVideo { s.updateVideoStatus(ctx, log.SourceID, verifyStatus) } @@ -373,36 +352,27 @@ func (s *MaterialVerifyService) ProcessImageResultByTask(ctx context.Context, ta result, err := yidunService.ImageDetection.GetImageResult(ctx, taskID) if err != nil { - // 判断是否是未找到结果或仍在检测中的错误 if err == yidunService.ErrImageResultNotFound || err == yidunService.ErrImageStillProcessing { - // 未获取到结果（任务不存在或仍在处理），不更新状态，保持等待下次轮询 g.Log().Infof(ctx, "图片检测结果未就绪, taskId=%s, 保持pending状态, err=%v", taskID, err) return nil } - // 其他错误（如额度用光、网络错误、API错误等），不更新状态，保持待检验 - // 只有易盾明确返回suggestion=BLOCK时才标记为失败 dao.MaterialVerifyLog.UpdateError(ctx, log.Id, entity.VerifyStatusPending, err.Error()) g.Log().Warningf(ctx, "图片检测查询失败（保持待检验）, taskId=%s, error=%v", taskID, err) - return nil // 返回nil避免日志被反复处理，但保持pending状态 + return nil } - // 判断检测状态 if result.Status == YidunStatusProcessing || result.Status == YidunStatusNotStart { - // 检测仍在进行中，保持pending状态 g.Log().Infof(ctx, "图片检测仍在进行中, taskId=%s, status=%d, 保持pending状态", taskID, result.Status) return nil } if result.Status == YidunStatusFailed { - // 易盾检测失败（如额度用光、服务端错误等），不更新状态，保持待检验 - // 只有易盾明确返回suggestion=BLOCK时才标记为失败 - errMsg := fmt.Sprintf("易盾检测失败, status=%d", result.Status) - dao.MaterialVerifyLog.UpdateError(ctx, log.Id, entity.VerifyStatusPending, errMsg) + dao.MaterialVerifyLog.UpdateError(ctx, log.Id, entity.VerifyStatusPending, + fmt.Sprintf("易盾检测失败, status=%d", result.Status)) g.Log().Warningf(ctx, "图片检测失败（保持待检验）, taskId=%s, status=%d", taskID, result.Status) return nil } - // status == YidunStatusSuccess，检测成功，根据suggestion更新状态 verifyStatus := SuggestionToVerifyStatus(result.Suggestion) responseJSON, _ := json.Marshal(result) @@ -427,36 +397,27 @@ func (s *MaterialVerifyService) ProcessVideoResultByTask(ctx context.Context, ta result, err := yidunService.VideoDetection.GetVideoResult(ctx, taskID) if err != nil { - // 判断是否是未找到结果或仍在检测中的错误 if err == yidunService.ErrVideoResultNotFound || err == yidunService.ErrVideoStillProcessing { - // 未获取到结果（任务不存在或仍在处理），不更新状态，保持等待下次轮询 g.Log().Infof(ctx, "视频检测结果未就绪, taskId=%s, 保持pending状态, err=%v", taskID, err) return nil } - // 其他错误（如额度用光、网络错误、API错误等），不更新状态，保持待检验 - // 只有易盾明确返回suggestion=BLOCK时才标记为失败 dao.MaterialVerifyLog.UpdateError(ctx, log.Id, entity.VerifyStatusPending, err.Error()) g.Log().Warningf(ctx, "视频检测查询失败（保持待检验）, taskId=%s, error=%v", taskID, err) - return nil // 返回nil避免日志被反复处理，但保持pending状态 + return nil } - // 判断检测状态 if result.Status == YidunStatusProcessing || result.Status == YidunStatusNotStart { - // 检测仍在进行中，保持pending状态 g.Log().Infof(ctx, "视频检测仍在进行中, taskId=%s, status=%d, 保持pending状态", taskID, result.Status) return nil } if result.Status == YidunStatusFailed { - // 易盾检测失败（如额度用光、服务端错误等），不更新状态，保持待检验 - // 只有易盾明确返回suggestion=BLOCK时才标记为失败 - errMsg := fmt.Sprintf("易盾检测失败, status=%d", result.Status) - dao.MaterialVerifyLog.UpdateError(ctx, log.Id, entity.VerifyStatusPending, errMsg) + dao.MaterialVerifyLog.UpdateError(ctx, log.Id, entity.VerifyStatusPending, + fmt.Sprintf("易盾检测失败, status=%d", result.Status)) g.Log().Warningf(ctx, "视频检测失败（保持待检验）, taskId=%s, status=%d", taskID, result.Status) return nil } - // status == YidunStatusSuccess，检测成功，根据suggestion更新状态 verifyStatus := SuggestionToVerifyStatus(result.Suggestion) responseJSON, _ := json.Marshal(result) @@ -478,8 +439,16 @@ func (s *MaterialVerifyService) ProcessVideoResultByTask(ctx context.Context, ta // createVerifyLog 创建校验日志 func (s *MaterialVerifyService) createVerifyLog(ctx context.Context, materialType, materialID, sourceTable string, sourceID, accountID int64) *entity.MaterialVerifyLog { + // 从上下文提取租户ID + var tenantID int64 + if user := ctx.Value("user"); user != nil { + if u, ok := user.(*beans.User); ok { + tenantID = int64(u.TenantId) + } + } + log := &entity.MaterialVerifyLog{ - TenantID: 0, + TenantID: tenantID, MaterialType: materialType, MaterialID: materialID, SourceTable: sourceTable, @@ -498,7 +467,7 @@ func (s *MaterialVerifyService) createVerifyLog(ctx context.Context, materialTyp return log } -// updateImageStatus 更新图片状态 +// updateImageStatus 更新图片状态（已记录日志则同步更新，失败仅记录日志不影响主流程） func (s *MaterialVerifyService) updateImageStatus(ctx context.Context, imageID int64, verifyStatus string) { _, err := dao.TencentImage.UpdateStatus(ctx, imageID, verifyStatus) if err != nil { @@ -508,7 +477,7 @@ func (s *MaterialVerifyService) updateImageStatus(ctx context.Context, imageID i } } -// updateVideoStatus 更新视频状态 +// updateVideoStatus 更新视频状态（已记录日志则同步更新，失败仅记录日志不影响主流程） func (s *MaterialVerifyService) updateVideoStatus(ctx context.Context, videoID int64, verifyStatus string) { _, err := dao.TencentVideo.UpdateStatus(ctx, videoID, verifyStatus) if err != nil { @@ -706,11 +675,12 @@ func getFailureReason(log *entity.MaterialVerifyLog) string { return suggestionText } -// ExportRejectedData 导出不通过数据 +const exportBatchSize = 1000 + +// ExportRejectedData 导出不通过数据（分批加载，避免OOM） func (s *MaterialVerifyService) ExportRejectedData(ctx context.Context, materialType string) ([]ExportRejectedItem, error) { var items []ExportRejectedItem - // 加载账户名称映射 accountMap := make(map[int64]string) if accounts, err := dao.TencentAccountRelation.GetAll(ctx); err == nil { for _, acc := range accounts { @@ -720,70 +690,82 @@ func (s *MaterialVerifyService) ExportRejectedData(ctx context.Context, material } } - // 处理图片 if materialType == "" || materialType == entity.MaterialTypeImage { condition := map[string]interface{}{ entity.TencentImageCols.VerifyStatus: entity.VerifyStatusRejected, } - images, total, err := dao.TencentImage.GetByCondition(ctx, condition, 1, 100000) - if err != nil { - g.Log().Errorf(ctx, "查询不通过图片失败: %v", err) - return nil, fmt.Errorf("查询不通过图片失败: %w", err) - } - g.Log().Infof(ctx, "导出不通过图片: total=%d, got=%d", total, len(images)) - for _, img := range images { - // 查询最后一条失败的校验日志 - log, _ := dao.MaterialVerifyLog.GetLastRejectedLogByMaterialID(ctx, img.ImageID, entity.VerifyStatusRejected) - var createdAtStr string - if log != nil && log.CreatedAt != nil { - createdAtStr = log.CreatedAt.Format("Y-m-d H:i:s") + page := 1 + for { + images, total, err := dao.TencentImage.GetByCondition(ctx, condition, page, exportBatchSize) + if err != nil { + g.Log().Errorf(ctx, "查询不通过图片失败: %v", err) + return nil, fmt.Errorf("unknown error: %w", err) } - items = append(items, ExportRejectedItem{ - ID: img.Id, - MaterialID: img.ImageID, - AccountID: img.AccountID, - CorporationName: accountMap[img.AccountID], - PreviewURL: img.PreviewURL, - Description: img.Description, - ErrorMsg: getFailureReason(log), - MaterialType: entity.MaterialTypeImage, - ImageUsage: img.ImageUsage, - CreatedAt: createdAtStr, - }) + + for _, img := range images { + log, _ := dao.MaterialVerifyLog.GetLastRejectedLogByMaterialID(ctx, img.ImageID, entity.VerifyStatusRejected) + var createdAtStr string + if log != nil && log.CreatedAt != nil { + createdAtStr = log.CreatedAt.Format("2006-01-02 15:04:05") + } + items = append(items, ExportRejectedItem{ + ID: img.Id, + MaterialID: img.ImageID, + AccountID: img.AccountID, + CorporationName: accountMap[img.AccountID], + PreviewURL: img.PreviewURL, + Description: img.Description, + ErrorMsg: getFailureReason(log), + MaterialType: entity.MaterialTypeImage, + ImageUsage: img.ImageUsage, + CreatedAt: createdAtStr, + }) + } + + if page*exportBatchSize >= total { + break + } + page++ } } - // 处理视频 if materialType == "" || materialType == entity.MaterialTypeVideo { condition := map[string]interface{}{ entity.TencentVideoCols.VerifyStatus: entity.VerifyStatusRejected, } - videos, total, err := dao.TencentVideo.GetByCondition(ctx, condition, 1, 100000) - if err != nil { - g.Log().Errorf(ctx, "查询不通过视频失败: %v", err) - return nil, fmt.Errorf("查询不通过视频失败: %w", err) - } - g.Log().Infof(ctx, "导出不通过视频: total=%d, got=%d", total, len(videos)) - for _, vid := range videos { - // 查询最后一条失败的校验日志 - log, _ := dao.MaterialVerifyLog.GetLastRejectedLogByMaterialID(ctx, vid.VideoID, entity.VerifyStatusRejected) - var createdAtStr string - if log != nil && log.CreatedAt != nil { - createdAtStr = log.CreatedAt.Format("Y-m-d H:i:s") + page := 1 + for { + videos, total, err := dao.TencentVideo.GetByCondition(ctx, condition, page, exportBatchSize) + if err != nil { + g.Log().Errorf(ctx, "查询不通过视频失败: %v", err) + return nil, fmt.Errorf("unknown error: %w", err) } - items = append(items, ExportRejectedItem{ - ID: vid.Id, - MaterialID: vid.VideoID, - AccountID: vid.AccountID, - CorporationName: accountMap[vid.AccountID], - PreviewURL: vid.PreviewURL, - Description: vid.Description, - ErrorMsg: getFailureReason(log), - MaterialType: entity.MaterialTypeVideo, - CreatedAt: createdAtStr, - }) + + for _, vid := range videos { + log, _ := dao.MaterialVerifyLog.GetLastRejectedLogByMaterialID(ctx, vid.VideoID, entity.VerifyStatusRejected) + var createdAtStr string + if log != nil && log.CreatedAt != nil { + createdAtStr = log.CreatedAt.Format("2006-01-02 15:04:05") + } + items = append(items, ExportRejectedItem{ + ID: vid.Id, + MaterialID: vid.VideoID, + AccountID: vid.AccountID, + CorporationName: accountMap[vid.AccountID], + PreviewURL: vid.PreviewURL, + Description: vid.Description, + ErrorMsg: getFailureReason(log), + MaterialType: entity.MaterialTypeVideo, + CreatedAt: createdAtStr, + }) + } + + if page*exportBatchSize >= total { + break + } + page++ } } diff --git a/service/dataengine/tencent_content_callback_service.go b/service/dataengine/tencent_content_callback_service.go index 0a7a0c9..cb2dd14 100644 --- a/service/dataengine/tencent_content_callback_service.go +++ b/service/dataengine/tencent_content_callback_service.go @@ -36,26 +36,21 @@ func (s *TencentContentCallbackService) ProcessImageCallback(ctx context.Context g.Log().Infof(ctx, "处理图片检测结果 - taskId: %s, suggestion: %d, resultType: %d", antispam.TaskId, antispam.Suggestion, antispam.ResultType) - // 根据 taskId 查找送检日志 log, err := dao.TencentContentCheckLog.GetByTaskID(ctx, antispam.TaskId) if err != nil { g.Log().Errorf(ctx, "查找送检日志失败, taskId=%s: %v", antispam.TaskId, err) return fmt.Errorf("查找送检日志失败: %w", err) } - if log == nil { g.Log().Warningf(ctx, "未找到送检日志, taskId=%s", antispam.TaskId) return nil } - // 更新送检日志 - checkTime := antispam.CensorTime - err = dao.TencentContentCheckLog.UpdateCheckResult(ctx, log.Id, - antispam.Suggestion, antispam.Label, antispam.ResultType, checkTime) + antispam.Suggestion, antispam.Label, antispam.ResultType, antispam.CensorTime) if err != nil { g.Log().Errorf(ctx, "更新送检日志检测结果失败: %v", err) - return err + return fmt.Errorf("更新送检日志检测结果失败: %w", err) } g.Log().Infof(ctx, "图片检测回调处理完成, taskId=%s, suggestion=%d", antispam.TaskId, antispam.Suggestion) @@ -80,19 +75,16 @@ func (s *TencentContentCallbackService) ProcessVideoCallback(ctx context.Context g.Log().Infof(ctx, "处理视频检测结果 - taskId: %s, suggestion: %d, resultType: %d, censorSource: %d", antispam.TaskID, antispam.Suggestion, antispam.ResultType, antispam.CensorSource) - // 根据 taskId 查找送检日志 log, err := dao.TencentContentCheckLog.GetByTaskID(ctx, antispam.TaskID) if err != nil { g.Log().Errorf(ctx, "查找送检日志失败, taskId=%s: %v", antispam.TaskID, err) return fmt.Errorf("查找送检日志失败: %w", err) } - if log == nil { g.Log().Warningf(ctx, "未找到送检日志, taskId=%s", antispam.TaskID) return nil } - // 更新送检日志 checkTime := antispam.CensorTime if checkTime == 0 { checkTime = antispam.CheckTime @@ -102,7 +94,7 @@ func (s *TencentContentCallbackService) ProcessVideoCallback(ctx context.Context antispam.Suggestion, antispam.Label, antispam.ResultType, checkTime) if err != nil { g.Log().Errorf(ctx, "更新送检日志检测结果失败: %v", err) - return err + return fmt.Errorf("更新送检日志检测结果失败: %w", err) } g.Log().Infof(ctx, "视频检测回调处理完成, taskId=%s, suggestion=%d", antispam.TaskID, antispam.Suggestion) @@ -113,25 +105,22 @@ func (s *TencentContentCallbackService) ProcessVideoCallback(ctx context.Context func (s *TencentContentCallbackService) ProcessImageResult(ctx context.Context, taskID string) error { g.Log().Infof(ctx, "查询图片检测结果, taskId: %s", taskID) - // 查找送检日志 log, err := dao.TencentContentCheckLog.GetByTaskID(ctx, taskID) if err != nil || log == nil { return fmt.Errorf("未找到送检日志, taskId=%s", taskID) } - // 调用易盾查询结果 result, err := yidunService.ImageDetection.GetImageResult(ctx, taskID) if err != nil { g.Log().Errorf(ctx, "查询图片检测结果失败: %v", err) - return err + return fmt.Errorf("查询图片检测结果失败: %w", err) } - // 更新日志 err = dao.TencentContentCheckLog.UpdateCheckResult(ctx, log.Id, result.Suggestion, result.Label, result.ResultType, result.CensorTime) if err != nil { g.Log().Errorf(ctx, "更新送检日志检测结果失败: %v", err) - return err + return fmt.Errorf("更新送检日志检测结果失败: %w", err) } g.Log().Infof(ctx, "图片检测结果处理完成, taskId=%s, suggestion=%d", taskID, result.Suggestion) @@ -142,25 +131,22 @@ func (s *TencentContentCallbackService) ProcessImageResult(ctx context.Context, func (s *TencentContentCallbackService) ProcessVideoResult(ctx context.Context, taskID string) error { g.Log().Infof(ctx, "查询视频检测结果, taskId: %s", taskID) - // 查找送检日志 log, err := dao.TencentContentCheckLog.GetByTaskID(ctx, taskID) if err != nil || log == nil { return fmt.Errorf("未找到送检日志, taskId=%s", taskID) } - // 调用易盾查询结果 result, err := yidunService.VideoDetection.GetVideoResult(ctx, taskID) if err != nil { g.Log().Errorf(ctx, "查询视频检测结果失败: %v", err) - return err + return fmt.Errorf("查询视频检测结果失败: %w", err) } - // 更新日志 err = dao.TencentContentCheckLog.UpdateCheckResult(ctx, log.Id, result.Suggestion, result.Label, result.ResultType, result.CensorTime) if err != nil { g.Log().Errorf(ctx, "更新送检日志检测结果失败: %v", err) - return err + return fmt.Errorf("更新送检日志检测结果失败: %w", err) } g.Log().Infof(ctx, "视频检测结果处理完成, taskId=%s, suggestion=%d", taskID, result.Suggestion) @@ -169,7 +155,6 @@ func (s *TencentContentCallbackService) ProcessVideoResult(ctx context.Context, // GetCheckLogsByImageID 根据图片ID获取送检日志 func (s *TencentContentCallbackService) GetCheckLogsByImageID(ctx context.Context, imageID string) ([]entity.TencentContentCheckLog, error) { - // 先获取图片数据 image, err := dao.TencentImage.GetByImageID(ctx, imageID) if err != nil || image == nil { return nil, fmt.Errorf("未找到图片数据") @@ -180,7 +165,6 @@ func (s *TencentContentCallbackService) GetCheckLogsByImageID(ctx context.Contex // GetCheckLogsByVideoID 根据视频ID获取送检日志 func (s *TencentContentCallbackService) GetCheckLogsByVideoID(ctx context.Context, videoID string) ([]entity.TencentContentCheckLog, error) { - // 先获取视频数据 video, err := dao.TencentVideo.GetByVideoID(ctx, videoID) if err != nil || video == nil { return nil, fmt.Errorf("未找到视频数据") diff --git a/service/dataengine/tencent_content_check_service.go b/service/dataengine/tencent_content_check_service.go index be04af8..4f3fe4c 100644 --- a/service/dataengine/tencent_content_check_service.go +++ b/service/dataengine/tencent_content_check_service.go @@ -7,7 +7,6 @@ import ( yidunService "cid/service/yidun" "context" "encoding/json" - "fmt" "time" "gitea.redpowerfuture.com/red-future/common/beans" @@ -16,14 +15,10 @@ import ( // ContentCheckConfig 送检配置 type ContentCheckConfig struct { - // 每批处理数量 - BatchSize int `json:"batch_size"` - // 图片检测启用 - ImageEnabled bool `json:"image_enabled"` - // 视频检测启用 - VideoEnabled bool `json:"video_enabled"` - // 定时任务间隔（秒） - IntervalSeconds int `json:"interval_seconds"` + BatchSize int `json:"batch_size"` + ImageEnabled bool `json:"image_enabled"` + VideoEnabled bool `json:"video_enabled"` + IntervalSeconds int `json:"interval_seconds"` } // DefaultConfig 默认配置 @@ -61,7 +56,11 @@ func (s *TencentContentCheckService) Start(ctx context.Context) error { g.Log().Infof(ctx, "启动内容送检服务，配置: batch_size=%d, interval=%ds, image=%v, video=%v", s.config.BatchSize, s.config.IntervalSeconds, s.config.ImageEnabled, s.config.VideoEnabled) - go s.runScheduler(ctx) + schedCtx := context.Background() + if user := ctx.Value("user"); user != nil { + schedCtx = context.WithValue(schedCtx, "user", user) + } + go s.runScheduler(schedCtx) return nil } @@ -76,23 +75,18 @@ func (s *TencentContentCheckService) runScheduler(ctx context.Context) { ticker := time.NewTicker(time.Duration(s.config.IntervalSeconds) * time.Second) defer ticker.Stop() - // 启动时先执行一次 s.processAll(ctx) - for s.isRunning { - select { - case <-ticker.C: - s.processAll(ctx) - case <-ctx.Done(): - s.isRunning = false + for range ticker.C { + if !s.isRunning { return } + s.processAll(ctx) } } // processAll 处理所有待送检数据 func (s *TencentContentCheckService) processAll(ctx context.Context) { - // 添加系统用户上下文，绕过gfdb租户验证 ctx = context.WithValue(ctx, "user", &beans.User{UserName: "system", TenantId: 1}) startTime := time.Now() @@ -100,7 +94,6 @@ func (s *TencentContentCheckService) processAll(ctx context.Context) { var totalProcessed int - // 处理图片 if s.config.ImageEnabled { imageCount, _ := dao.TencentImage.CountPending(ctx) if imageCount > 0 { @@ -109,7 +102,6 @@ func (s *TencentContentCheckService) processAll(ctx context.Context) { } } - // 处理视频 if s.config.VideoEnabled { videoCount, _ := dao.TencentVideo.CountPending(ctx) if videoCount > 0 { @@ -122,9 +114,8 @@ func (s *TencentContentCheckService) processAll(ctx context.Context) { g.Log().Infof(ctx, "处理完成，共处理 %d 条数据，耗时 %dms", totalProcessed, duration) } -// processImages 处理图片送检 +// processImages 处理图片送检（统一走 MaterialVerify 系统） func (s *TencentContentCheckService) processImages(ctx context.Context) (int, error) { - // 获取待送检图片 images, err := dao.TencentImage.GetPendingList(ctx, s.config.BatchSize) if err != nil { g.Log().Errorf(ctx, "获取待送检图片失败: %v", err) @@ -141,22 +132,17 @@ func (s *TencentContentCheckService) processImages(ctx context.Context) (int, er failedCount := 0 for _, img := range images { - // 创建送检日志 - log := s.createCheckLog(ctx, consts.SourceTableTencentImage, img.Id, img.ImageID, img.PreviewURL) - - // 提交送检 - err := s.submitImageCheck(ctx, &img, log) + // 统一走 MaterialVerify 系统提交（处理完整校验流程：日志→提交→状态反写） + mLog, err := MaterialVerify.VerifyImageByID(ctx, img.ImageID) if err != nil { failedCount++ - // 更新日志为失败 - if log != nil { - dao.TencentContentCheckLog.UpdateStatus(ctx, log.Id, consts.CheckStatusFailed, "", err.Error()) - } + g.Log().Errorf(ctx, "图片送检失败, imageId=%s, error=%v", img.ImageID, err) } else { successCount++ + // 审计日志：同步写入 tencent_content_check_log + s.writeAuditLog(ctx, consts.SourceTableTencentImage, img.Id, img.ImageID, img.PreviewURL, mLog.TaskID) } - // 避免请求过快 time.Sleep(100 * time.Millisecond) } @@ -164,9 +150,8 @@ func (s *TencentContentCheckService) processImages(ctx context.Context) (int, er return len(images), nil } -// processVideos 处理视频送检 +// processVideos 处理视频送检（统一走 MaterialVerify 系统） func (s *TencentContentCheckService) processVideos(ctx context.Context) (int, error) { - // 获取待送检视频 videos, err := dao.TencentVideo.GetPendingList(ctx, s.config.BatchSize) if err != nil { g.Log().Errorf(ctx, "获取待送检视频失败: %v", err) @@ -183,22 +168,15 @@ func (s *TencentContentCheckService) processVideos(ctx context.Context) (int, er failedCount := 0 for _, video := range videos { - // 创建送检日志 - log := s.createCheckLog(ctx, consts.SourceTableTencentVideo, video.Id, video.VideoID, video.PreviewURL) - - // 提交送检 - err := s.submitVideoCheck(ctx, &video, log) + mLog, err := MaterialVerify.VerifyVideoByID(ctx, video.VideoID) if err != nil { failedCount++ - // 更新日志为失败 - if log != nil { - dao.TencentContentCheckLog.UpdateStatus(ctx, log.Id, consts.CheckStatusFailed, "", err.Error()) - } + g.Log().Errorf(ctx, "视频送检失败, videoId=%s, error=%v", video.VideoID, err) } else { successCount++ + s.writeAuditLog(ctx, consts.SourceTableTencentVideo, video.Id, video.VideoID, video.PreviewURL, mLog.TaskID) } - // 避免请求过快 time.Sleep(100 * time.Millisecond) } @@ -206,8 +184,8 @@ func (s *TencentContentCheckService) processVideos(ctx context.Context) (int, er return len(videos), nil } -// createCheckLog 创建送检日志 -func (s *TencentContentCheckService) createCheckLog(ctx context.Context, sourceTable string, sourceID int64, mediaID string, mediaURL string) *entity.TencentContentCheckLog { +// writeAuditLog 写入审计日志（tencent_content_check_log） +func (s *TencentContentCheckService) writeAuditLog(ctx context.Context, sourceTable string, sourceID int64, mediaID string, mediaURL string, taskID string) { requestParam := map[string]interface{}{ "media_id": mediaID, "url": mediaURL, @@ -219,147 +197,52 @@ func (s *TencentContentCheckService) createCheckLog(ctx context.Context, sourceT SourceID: sourceID, RequestURL: "易盾内容安全检测接口", RequestParam: string(requestParamJSON), - Status: consts.CheckStatusPending, + Status: consts.CheckStatusSuccess, CheckTime: time.Now().UnixMilli(), + TaskID: taskID, } id, err := dao.TencentContentCheckLog.Create(ctx, log) if err != nil { - g.Log().Errorf(ctx, "创建送检日志失败: %v", err) - return nil + g.Log().Errorf(ctx, "创建送检审计日志失败: %v", err) + return } - log.Id = id - g.Log().Debugf(ctx, "创建送检日志成功, id=%d, sourceTable=%s, sourceID=%d", id, sourceTable, sourceID) - return log + g.Log().Debugf(ctx, "创建送检审计日志成功, id=%d, sourceTable=%s, sourceID=%d, taskId=%s", id, sourceTable, sourceID, taskID) } -// submitImageCheck 提交图片送检 -func (s *TencentContentCheckService) submitImageCheck(ctx context.Context, image *entity.TencentImage, log *entity.TencentContentCheckLog) error { - startTime := time.Now() - - // 更新日志状态为送检中 - if log != nil { - dao.TencentContentCheckLog.UpdateStatus(ctx, log.Id, consts.CheckStatusSubmitting, "", "") - } - - // 获取回调地址 - callbackURL := g.Cfg().MustGet(ctx, "yidun.image.callback_url").String() - - // 调用易盾图片检测 - result, err := yidunService.ImageDetection.DetectImage(ctx, image.PreviewURL, image.ImageID, callbackURL) - - duration := time.Since(startTime).Milliseconds() - - // 更新日志 - if log != nil { - if err != nil { - dao.TencentContentCheckLog.UpdateDuration(ctx, log.Id, duration) - dao.TencentContentCheckLog.UpdateStatus(ctx, log.Id, consts.CheckStatusFailed, "", err.Error()) - g.Log().Errorf(ctx, "图片送检失败, id=%d, url=%s, error=%v", image.Id, image.PreviewURL, err) - return err - } - - // 更新日志和图片状态 - responseData, _ := json.Marshal(result) - dao.TencentContentCheckLog.UpdateStatus(ctx, log.Id, consts.CheckStatusSuccess, string(responseData), "") - dao.TencentContentCheckLog.UpdateTaskID(ctx, log.Id, result.TaskID) - dao.TencentContentCheckLog.UpdateDuration(ctx, log.Id, duration) - } - - g.Log().Infof(ctx, "图片送检成功, id=%d, imageId=%s, taskId=%s", image.Id, image.ImageID, result.TaskID) - return nil -} - -// submitVideoCheck 提交视频送检 -func (s *TencentContentCheckService) submitVideoCheck(ctx context.Context, video *entity.TencentVideo, log *entity.TencentContentCheckLog) error { - startTime := time.Now() - - // 更新日志状态为送检中 - if log != nil { - dao.TencentContentCheckLog.UpdateStatus(ctx, log.Id, consts.CheckStatusSubmitting, "", "") - } - - // 获取回调地址 - callbackURL := g.Cfg().MustGet(ctx, "yidun.video.callback_url").String() - - // 调用易盾视频检测 - result, err := yidunService.VideoDetection.DetectVideo(ctx, video.PreviewURL, video.VideoID, callbackURL) - - duration := time.Since(startTime).Milliseconds() - - // 更新日志 - if log != nil { - if err != nil { - dao.TencentContentCheckLog.UpdateDuration(ctx, log.Id, duration) - dao.TencentContentCheckLog.UpdateStatus(ctx, log.Id, consts.CheckStatusFailed, "", err.Error()) - g.Log().Errorf(ctx, "视频送检失败, id=%d, url=%s, error=%v", video.Id, video.PreviewURL, err) - return err - } - - // 更新日志和视频状态 - responseData, _ := json.Marshal(result) - dao.TencentContentCheckLog.UpdateStatus(ctx, log.Id, consts.CheckStatusSuccess, string(responseData), "") - dao.TencentContentCheckLog.UpdateTaskID(ctx, log.Id, result.TaskID) - dao.TencentContentCheckLog.UpdateDuration(ctx, log.Id, duration) - } - - g.Log().Infof(ctx, "视频送检成功, id=%d, videoId=%s, taskId=%s", video.Id, video.VideoID, result.TaskID) - return nil -} - -// SubmitImageByID 根据图片ID手动提交送检 +// SubmitImageByID 根据图片ID手动提交送检（统一走 MaterialVerify 系统） func (s *TencentContentCheckService) SubmitImageByID(ctx context.Context, imageID string) (*yidunService.ImageSubmitResult, error) { - // 根据图片ID获取数据 - image, err := dao.TencentImage.GetByImageID(ctx, imageID) - if err != nil { - return nil, fmt.Errorf("查询图片数据失败: %w", err) - } - if image == nil { - return nil, fmt.Errorf("未找到图片数据, imageID=%s", imageID) - } - - // 创建送检日志 - log := s.createCheckLog(ctx, consts.SourceTableTencentImage, image.Id, image.ImageID, image.PreviewURL) - if log == nil { - return nil, fmt.Errorf("创建送检日志失败") - } - - // 提交送检 - err = s.submitImageCheck(ctx, image, log) + mLog, err := MaterialVerify.VerifyImageByID(ctx, imageID) if err != nil { return nil, err } - // 获取送检结果 - return dao.TencentContentCheckLog.GetImageSubmitResult(ctx, log.Id) + image, err := dao.TencentImage.GetByImageID(ctx, imageID) + if err == nil && image != nil { + s.writeAuditLog(ctx, consts.SourceTableTencentImage, image.Id, imageID, image.PreviewURL, mLog.TaskID) + } + + return &yidunService.ImageSubmitResult{ + TaskID: mLog.TaskID, + }, nil } -// SubmitVideoByID 根据视频ID手动提交送检 +// SubmitVideoByID 根据视频ID手动提交送检（统一走 MaterialVerify 系统） func (s *TencentContentCheckService) SubmitVideoByID(ctx context.Context, videoID string) (*yidunService.VideoSubmitResult, error) { - // 根据视频ID获取数据 - video, err := dao.TencentVideo.GetByVideoID(ctx, videoID) - if err != nil { - return nil, fmt.Errorf("查询视频数据失败: %w", err) - } - if video == nil { - return nil, fmt.Errorf("未找到视频数据, videoID=%s", videoID) - } - - // 创建送检日志 - log := s.createCheckLog(ctx, consts.SourceTableTencentVideo, video.Id, video.VideoID, video.PreviewURL) - if log == nil { - return nil, fmt.Errorf("创建送检日志失败") - } - - // 提交送检 - err = s.submitVideoCheck(ctx, video, log) + mLog, err := MaterialVerify.VerifyVideoByID(ctx, videoID) if err != nil { return nil, err } - // 获取送检结果 - return dao.TencentContentCheckLog.GetVideoSubmitResult(ctx, log.Id) + video, err := dao.TencentVideo.GetByVideoID(ctx, videoID) + if err == nil && video != nil { + s.writeAuditLog(ctx, consts.SourceTableTencentVideo, video.Id, videoID, video.PreviewURL, mLog.TaskID) + } + + return &yidunService.VideoSubmitResult{ + TaskID: mLog.TaskID, + }, nil } // GetPendingStats 获取待送检统计 diff --git a/sql/alter_material_verify_log.sql b/sql/alter_material_verify_log.sql new file mode 100644 index 0000000..933ae90 --- /dev/null +++ b/sql/alter_material_verify_log.sql @@ -0,0 +1,6 @@ +-- 补充 material_verify_log 表缺失的 SQLBaseDO 列 +ALTER TABLE material_verify_log + ADD COLUMN IF NOT EXISTS creator VARCHAR(100) DEFAULT '' NOT NULL, + ADD COLUMN IF NOT EXISTS updater VARCHAR(100) DEFAULT '' NOT NULL, + ADD COLUMN IF NOT EXISTS updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, + ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMP WITH TIME ZONE DEFAULT NULL; diff --git a/sql/alter_tencent_content_check_log.sql b/sql/alter_tencent_content_check_log.sql new file mode 100644 index 0000000..c2d007a --- /dev/null +++ b/sql/alter_tencent_content_check_log.sql @@ -0,0 +1,5 @@ +-- 补充 tencent_content_check_log 表缺失的 SQLBaseDO 列 +ALTER TABLE tencent_content_check_log + ADD COLUMN IF NOT EXISTS tenant_id BIGINT DEFAULT 0 NOT NULL, + ADD COLUMN IF NOT EXISTS creator VARCHAR(100) DEFAULT '' NOT NULL, + ADD COLUMN IF NOT EXISTS updater VARCHAR(100) DEFAULT '' NOT NULL;