red-future/cid
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

cid适配ppgo_job

Browse Source
This commit is contained in:
lmk
2026-07-18 10:10:23 +08:00
parent d930266fbf
commit 2fd858c97e
20 changed files with 810 additions and 884 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service/dataengine/material_verify_service.go
+165 -96
View File
@@ -20,12 +20,6 @@ const (
PollBatchSize = 20
)
// 状态常量
const (
// 原表状态 - 与 tencent_image/tencent_video 表的 status 字段对应
StatusSubmitting = consts.CheckStatusSubmitting // 送检中
)
// MaterialVerifyService 素材校验服务
type MaterialVerifyService struct{}
@@ -55,31 +49,39 @@ func SuggestionToVerifyStatus(suggestion int) string {
// =============================================================================
// VerifyImageByID 根据图片ID执行校验
// 使用原子 CAS 防止并发重复送检
func (s *MaterialVerifyService) VerifyImageByID(ctx context.Context, imageID string) (*entity.MaterialVerifyLog, error) {
image, err := dao.TencentImage.GetByImageID(ctx, imageID)
if err != nil {
return nil, fmt.Errorf("unknown error: %w", err)
return nil, fmt.Errorf("查询图片数据失败, imageID=%s: %w", imageID, err)
}
if image == nil {
return nil, fmt.Errorf("未找到图片数据, imageID=%s", imageID)
}
// 幂等性检查：如果已在送检中，直接返回已有日志
if image.VerifyStatus == consts.CheckStatusSubmitting {
// 原子 CAS：PENDING → SUBMITTING，确保只有第一个调用者能抢到处理权
claimed, err := dao.TencentImage.ClaimPending(ctx, image.Id)
if err != nil {
return nil, fmt.Errorf("认领图片送检失败, imageID=%s: %w", imageID, err)
}
if !claimed {
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)
g.Log().Infof(ctx, "图片已被其他进程送检, imageID=%s, logId=%d", imageID, logs[0].Id)
return &logs[0], nil
}
return nil, fmt.Errorf("图片正在送检中且无校验日志, imageID=%s", imageID)
}
log := s.createVerifyLog(ctx, entity.MaterialTypeImage, imageID, consts.SourceTableTencentImage, image.Id, image.AccountID)
if log == nil {
dao.TencentImage.UpdateStatus(ctx, image.Id, entity.VerifyStatusPending)
return nil, fmt.Errorf("创建校验日志失败")
}
err = s.submitImageCheck(ctx, image, log)
if err != nil {
dao.TencentImage.UpdateStatus(ctx, image.Id, entity.VerifyStatusPending)
return nil, err
}
@@ -113,13 +115,14 @@ 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("unknown error: %w", err)
return fmt.Errorf("图片异步检测提交失败, imageId=%s: %w", image.ImageID, err)
}
taskID = result.TaskID
dao.MaterialVerifyLog.UpdateTaskID(ctx, log.Id, taskID)
dao.MaterialVerifyLog.UpdateRequestParams(ctx, log.Id, string(requestParamsJSON))
s.updateImageStatus(ctx, image.Id, StatusSubmitting)
TencentContentCheck.writeAuditLog(ctx, consts.SourceTableTencentImage, image.Id, image.ImageID, image.PreviewURL, taskID, -1, 0, 0, "", duration)
g.Log().Infof(ctx, "图片异步检测已提交, id=%d, imageId=%s, taskId=%s, duration=%dms",
image.Id, image.ImageID, taskID, duration)
@@ -130,7 +133,7 @@ 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("unknown error: %w", err)
return fmt.Errorf("图片同步检测失败, imageId=%s: %w", image.ImageID, err)
}
taskID = syncResult.TaskID
@@ -143,6 +146,8 @@ func (s *MaterialVerifyService) submitImageCheck(ctx context.Context, image *ent
syncResult.Suggestion, syncResult.Label, syncResult.ResultType, string(responseJSON), syncResult.CensorTime)
s.updateImageStatus(ctx, image.Id, verifyStatus)
TencentContentCheck.writeAuditLog(ctx, consts.SourceTableTencentImage, image.Id, image.ImageID, image.PreviewURL, taskID, syncResult.Suggestion, syncResult.Label, syncResult.ResultType, string(responseJSON), duration)
g.Log().Infof(ctx, "图片同步检测完成, id=%d, imageId=%s, taskId=%s, suggestion=%d, verifyStatus=%s, duration=%dms",
image.Id, image.ImageID, taskID, syncResult.Suggestion, verifyStatus, duration)
}
@@ -155,31 +160,38 @@ func (s *MaterialVerifyService) submitImageCheck(ctx context.Context, image *ent
// =============================================================================
// VerifyVideoByID 根据视频ID执行校验
// 使用原子 CAS 防止并发重复送检
func (s *MaterialVerifyService) VerifyVideoByID(ctx context.Context, videoID string) (*entity.MaterialVerifyLog, error) {
video, err := dao.TencentVideo.GetByVideoID(ctx, videoID)
if err != nil {
return nil, fmt.Errorf("unknown error: %w", err)
return nil, fmt.Errorf("查询视频数据失败, videoID=%s: %w", videoID, err)
}
if video == nil {
return nil, fmt.Errorf("未找到视频数据, videoID=%s", videoID)
}
// 幂等性检查：如果已在送检中，直接返回已有日志
if video.VerifyStatus == consts.CheckStatusSubmitting {
claimed, err := dao.TencentVideo.ClaimPending(ctx, video.Id)
if err != nil {
return nil, fmt.Errorf("认领视频送检失败, videoID=%s: %w", videoID, err)
}
if !claimed {
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)
g.Log().Infof(ctx, "视频已被其他进程送检, videoID=%s, logId=%d", videoID, logs[0].Id)
return &logs[0], nil
}
return nil, fmt.Errorf("视频正在送检中且无校验日志, videoID=%s", videoID)
}
log := s.createVerifyLog(ctx, entity.MaterialTypeVideo, videoID, consts.SourceTableTencentVideo, video.Id, video.AccountID)
if log == nil {
dao.TencentVideo.UpdateStatus(ctx, video.Id, entity.VerifyStatusPending)
return nil, fmt.Errorf("创建校验日志失败")
}
err = s.submitVideoCheck(ctx, video, log)
if err != nil {
dao.TencentVideo.UpdateStatus(ctx, video.Id, entity.VerifyStatusPending)
return nil, err
}
@@ -211,21 +223,18 @@ func (s *MaterialVerifyService) submitVideoCheck(ctx context.Context, video *ent
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("unknown error: %w", err)
return fmt.Errorf("视频检测提交失败, videoId=%s: %w", video.VideoID, err)
}
dao.MaterialVerifyLog.UpdateTaskID(ctx, log.Id, result.TaskID)
dao.MaterialVerifyLog.UpdateRequestParams(ctx, log.Id, string(requestParamsJSON))
s.updateVideoStatus(ctx, video.Id, StatusSubmitting)
if !callbackMode {
g.Log().Infof(ctx, "轮询模式：提交后立即查询结果, taskId=%s", result.TaskID)
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)
}
g.Log().Infof(ctx, "轮询模式：视频检测已提交, taskId=%s, 请通过轮询接口获取结果", result.TaskID)
}
TencentContentCheck.writeAuditLog(ctx, consts.SourceTableTencentVideo, video.Id, video.VideoID, video.PreviewURL, result.TaskID, -1, 0, 0, "", duration)
g.Log().Infof(ctx, "视频校验已提交, id=%d, videoId=%s, taskId=%s, duration=%dms",
video.Id, video.VideoID, result.TaskID, duration)
@@ -243,7 +252,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("unknown error: %w", err)
return fmt.Errorf("解析图片回调数据失败: %w", err)
}
if callback.Antispam == nil {
@@ -256,7 +265,7 @@ func (s *MaterialVerifyService) ProcessImageCallback(ctx context.Context, callba
log, err := dao.MaterialVerifyLog.GetByTaskID(ctx, antispam.TaskId)
if err != nil {
return fmt.Errorf("unknown error: %w", err)
return fmt.Errorf("查询图片校验日志失败, taskId=%s: %w", antispam.TaskId, err)
}
if log == nil {
g.Log().Warningf(ctx, "未找到校验日志, taskId=%s", antispam.TaskId)
@@ -268,13 +277,20 @@ func (s *MaterialVerifyService) ProcessImageCallback(ctx context.Context, callba
err = dao.MaterialVerifyLog.UpdateVerifyResult(ctx, log.Id, verifyStatus,
antispam.Suggestion, antispam.Label, antispam.ResultType, callbackData, antispam.CensorTime)
if err != nil {
return fmt.Errorf("unknown error: %w", err)
return fmt.Errorf("更新图片校验日志结果失败, logId=%d: %w", log.Id, err)
}
if log.SourceTable == consts.SourceTableTencentImage {
s.updateImageStatus(ctx, log.SourceID, verifyStatus)
}
// 更新送检审计日志
s.updateCheckLogResult(ctx, antispam.TaskId, antispam.Suggestion, antispam.Label, antispam.ResultType, callbackData)
// 提取风险描述
if antispam.RiskDescription != "" {
_ = dao.MaterialVerifyLog.UpdateRiskDescription(ctx, log.Id, antispam.RiskDescription)
}
g.Log().Infof(ctx, "图片校验回调处理完成, taskId=%s, verifyStatus=%s, suggestion=%d",
antispam.TaskId, verifyStatus, antispam.Suggestion)
@@ -288,7 +304,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("unknown error: %w", err)
return fmt.Errorf("解析视频回调数据失败: %w", err)
}
if callback.Antispam == nil {
@@ -301,7 +317,7 @@ func (s *MaterialVerifyService) ProcessVideoCallback(ctx context.Context, callba
log, err := dao.MaterialVerifyLog.GetByTaskID(ctx, antispam.TaskID)
if err != nil {
return fmt.Errorf("unknown error: %w", err)
return fmt.Errorf("查询视频校验日志失败, taskId=%s: %w", antispam.TaskID, err)
}
if log == nil {
g.Log().Warningf(ctx, "未找到校验日志, taskId=%s", antispam.TaskID)
@@ -318,23 +334,47 @@ func (s *MaterialVerifyService) ProcessVideoCallback(ctx context.Context, callba
err = dao.MaterialVerifyLog.UpdateVerifyResult(ctx, log.Id, verifyStatus,
antispam.Suggestion, antispam.Label, antispam.ResultType, callbackData, checkTime)
if err != nil {
return fmt.Errorf("unknown error: %w", err)
return fmt.Errorf("更新视频校验日志结果失败, logId=%d: %w", log.Id, err)
}
if log.SourceTable == consts.SourceTableTencentVideo {
s.updateVideoStatus(ctx, log.SourceID, verifyStatus)
}
// 更新送检审计日志
s.updateCheckLogResult(ctx, antispam.TaskID, antispam.Suggestion, antispam.Label, antispam.ResultType, callbackData)
// 提取风险描述
if antispam.RiskDescription != "" {
_ = dao.MaterialVerifyLog.UpdateRiskDescription(ctx, log.Id, antispam.RiskDescription)
}
g.Log().Infof(ctx, "视频校验回调处理完成, taskId=%s, verifyStatus=%s, suggestion=%d",
antispam.TaskID, verifyStatus, antispam.Suggestion)
return nil
}
// updateCheckLogResult 更新 tencent_content_check_log 的检测结果
func (s *MaterialVerifyService) updateCheckLogResult(ctx context.Context, taskID string, suggestion, label, resultType int, responseData string) {
checkLog, err := dao.TencentContentCheckLog.GetByTaskID(ctx, taskID)
if err != nil {
g.Log().Warningf(ctx, "查询送检审计日志失败, taskId=%s: %v", taskID, err)
return
}
if checkLog == nil {
g.Log().Debugf(ctx, "送检审计日志不存在, taskId=%s（可能是通过 API 直接提交的）", taskID)
return
}
_ = dao.TencentContentCheckLog.UpdateCheckResult(ctx, checkLog.Id, suggestion, label, resultType, time.Now().UnixMilli())
}
// =============================================================================
// 轮询模式处理
// =============================================================================
// ErrResultPending 表示检测结果尚未就绪，非错误状态
var ErrResultPending = fmt.Errorf("检测结果尚未就绪")
// 易盾检测状态常量
const (
YidunStatusNotStart = 0 // 未开始
@@ -344,9 +384,13 @@ const (
)
// ProcessImageResultByTask 根据任务ID处理图片结果（轮询模式）
// 返回 nil 表示结果已处理完成，返回 ErrResultPending 表示仍未就绪
func (s *MaterialVerifyService) ProcessImageResultByTask(ctx context.Context, taskID string) error {
log, err := dao.MaterialVerifyLog.GetByTaskID(ctx, taskID)
if err != nil || log == nil {
if err != nil {
return fmt.Errorf("查询校验日志失败, taskId=%s: %w", taskID, err)
}
if log == nil {
return fmt.Errorf("未找到校验日志, taskId=%s", taskID)
}
@@ -354,23 +398,23 @@ func (s *MaterialVerifyService) ProcessImageResultByTask(ctx context.Context, ta
if err != nil {
if err == yidunService.ErrImageResultNotFound || err == yidunService.ErrImageStillProcessing {
g.Log().Infof(ctx, "图片检测结果未就绪, taskId=%s, 保持pending状态, err=%v", taskID, err)
return nil
return ErrResultPending
}
dao.MaterialVerifyLog.UpdateError(ctx, log.Id, entity.VerifyStatusPending, err.Error())
g.Log().Warningf(ctx, "图片检测查询失败（保持待检验）, taskId=%s, error=%v", taskID, err)
return nil
return ErrResultPending
}
if result.Status == YidunStatusProcessing || result.Status == YidunStatusNotStart {
g.Log().Infof(ctx, "图片检测仍在进行中, taskId=%s, status=%d, 保持pending状态", taskID, result.Status)
return nil
return ErrResultPending
}
if result.Status == YidunStatusFailed {
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
return ErrResultPending
}
verifyStatus := SuggestionToVerifyStatus(result.Suggestion)
@@ -383,6 +427,14 @@ func (s *MaterialVerifyService) ProcessImageResultByTask(ctx context.Context, ta
s.updateImageStatus(ctx, log.SourceID, verifyStatus)
}
// 提取风险描述
if result.Antispam != nil && result.Antispam.RiskDescription != nil {
_ = dao.MaterialVerifyLog.UpdateRiskDescription(ctx, log.Id, *result.Antispam.RiskDescription)
}
// 更新送检审计日志
s.updateCheckLogResult(ctx, taskID, result.Suggestion, result.Label, result.ResultType, string(responseJSON))
g.Log().Infof(ctx, "图片检测结果更新成功, taskId=%s, status=%d, suggestion=%d, verifyStatus=%s",
taskID, result.Status, result.Suggestion, verifyStatus)
return nil
@@ -391,7 +443,10 @@ func (s *MaterialVerifyService) ProcessImageResultByTask(ctx context.Context, ta
// ProcessVideoResultByTask 根据任务ID处理视频结果（轮询模式）
func (s *MaterialVerifyService) ProcessVideoResultByTask(ctx context.Context, taskID string) error {
log, err := dao.MaterialVerifyLog.GetByTaskID(ctx, taskID)
if err != nil || log == nil {
if err != nil {
return fmt.Errorf("查询校验日志失败, taskId=%s: %w", taskID, err)
}
if log == nil {
return fmt.Errorf("未找到校验日志, taskId=%s", taskID)
}
@@ -399,23 +454,23 @@ func (s *MaterialVerifyService) ProcessVideoResultByTask(ctx context.Context, ta
if err != nil {
if err == yidunService.ErrVideoResultNotFound || err == yidunService.ErrVideoStillProcessing {
g.Log().Infof(ctx, "视频检测结果未就绪, taskId=%s, 保持pending状态, err=%v", taskID, err)
return nil
return ErrResultPending
}
dao.MaterialVerifyLog.UpdateError(ctx, log.Id, entity.VerifyStatusPending, err.Error())
g.Log().Warningf(ctx, "视频检测查询失败（保持待检验）, taskId=%s, error=%v", taskID, err)
return nil
return ErrResultPending
}
if result.Status == YidunStatusProcessing || result.Status == YidunStatusNotStart {
g.Log().Infof(ctx, "视频检测仍在进行中, taskId=%s, status=%d, 保持pending状态", taskID, result.Status)
return nil
return ErrResultPending
}
if result.Status == YidunStatusFailed {
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
return ErrResultPending
}
verifyStatus := SuggestionToVerifyStatus(result.Suggestion)
@@ -428,6 +483,14 @@ func (s *MaterialVerifyService) ProcessVideoResultByTask(ctx context.Context, ta
s.updateVideoStatus(ctx, log.SourceID, verifyStatus)
}
// 提取风险描述
if result.Antispam != nil && result.Antispam.RiskDescription != nil {
_ = dao.MaterialVerifyLog.UpdateRiskDescription(ctx, log.Id, *result.Antispam.RiskDescription)
}
// 更新送检审计日志
s.updateCheckLogResult(ctx, taskID, result.Suggestion, result.Label, result.ResultType, string(responseJSON))
g.Log().Infof(ctx, "视频检测结果更新成功, taskId=%s, status=%d, suggestion=%d, verifyStatus=%s",
taskID, result.Status, result.Suggestion, verifyStatus)
return nil
@@ -439,7 +502,6 @@ 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 {
@@ -467,7 +529,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 {
@@ -477,7 +539,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 {
@@ -515,10 +577,8 @@ func (s *MaterialVerifyService) GetStats(ctx context.Context) (map[string]int, e
// 轮询模式 - 批量查询检测结果
// =============================================================================
// PollPendingResults 轮询所有待查询结果的日志（手动触发）
// 返回处理成功的数量和错误信息
// PollPendingResults 轮询所有待查询结果的日志
func (s *MaterialVerifyService) PollPendingResults(ctx context.Context) (int, int, error) {
// 获取待查询的日志
logs, err := dao.MaterialVerifyLog.GetPendingResults(ctx, PollBatchSize)
if err != nil {
return 0, 0, err
@@ -538,7 +598,6 @@ func (s *MaterialVerifyService) PollPendingResults(ctx context.Context) (int, in
for _, log := range logs {
var err error
// 根据来源表判断调用哪个接口
if log.SourceTable == consts.SourceTableTencentImage {
err = s.ProcessImageResultByTask(ctx, log.TaskID)
} else if log.SourceTable == consts.SourceTableTencentVideo {
@@ -548,7 +607,9 @@ func (s *MaterialVerifyService) PollPendingResults(ctx context.Context) (int, in
continue
}
if err != nil {
if err == ErrResultPending {
g.Log().Infof(ctx, "结果未就绪, logId=%d, taskId=%s", log.Id, log.TaskID)
} else if err != nil {
failCount++
lastErr = err
g.Log().Warningf(ctx, "处理结果失败, logId=%d, taskId=%s, error=%v", log.Id, log.TaskID, err)
@@ -557,23 +618,20 @@ func (s *MaterialVerifyService) PollPendingResults(ctx context.Context) (int, in
g.Log().Infof(ctx, "处理结果成功, logId=%d, taskId=%s", log.Id, log.TaskID)
}
// 避免请求过快
time.Sleep(100 * time.Millisecond)
}
g.Log().Infof(ctx, "轮询完成, 成功=%d, 失败=%d", successCount, failCount)
g.Log().Infof(ctx, "轮询完成, 成功=%d, 失败=%d, 未就绪=%d", successCount, failCount, len(logs)-successCount-failCount)
return successCount, failCount, lastErr
}
// PollPendingResultsByType 按类型轮询待查询结果的日志
func (s *MaterialVerifyService) PollPendingResultsByType(ctx context.Context, sourceTable string) (int, int, error) {
// 获取待查询的日志
logs, err := dao.MaterialVerifyLog.GetPendingResults(ctx, PollBatchSize)
if err != nil {
return 0, 0, err
}
// 过滤指定类型
var filteredLogs []entity.MaterialVerifyLog
for _, log := range logs {
if log.SourceTable == sourceTable {
@@ -586,8 +644,6 @@ func (s *MaterialVerifyService) PollPendingResultsByType(ctx context.Context, so
return 0, 0, nil
}
g.Log().Infof(ctx, "开始轮询 %d 条待处理结果, sourceTable=%s", len(filteredLogs), sourceTable)
successCount := 0
failCount := 0
var lastErr error
@@ -611,7 +667,6 @@ func (s *MaterialVerifyService) PollPendingResultsByType(ctx context.Context, so
time.Sleep(100 * time.Millisecond)
}
g.Log().Infof(ctx, "轮询完成, sourceTable=%s, 成功=%d, 失败=%d", sourceTable, successCount, failCount)
return successCount, failCount, lastErr
}
@@ -631,16 +686,16 @@ func (s *MaterialVerifyService) PollPendingVideoResults(ctx context.Context) (in
// ExportRejectedItem 导出的不通过数据项
type ExportRejectedItem struct {
ID int64 `json:"id"` // 素材表主键ID
MaterialID string `json:"materialId"` // 素材ID（imageId/videoId）
AccountID int64 `json:"accountId"` // 账户ID
CorporationName string `json:"corporationName"` // 公司名称
PreviewURL string `json:"previewUrl"` // 预览URL
Description string `json:"description"` // 描述
ErrorMsg string `json:"errorMsg"` // 失败原因（最后一条失败日志的error_msg）
MaterialType string `json:"materialType"` // 素材类型 IMAGE/VIDEO
ImageUsage string `json:"imageUsage"` // 图片用途（仅图片）
CreatedAt string `json:"createdAt"` // 检测时间（日志创建时间）
ID int64 `json:"id"`
MaterialID string `json:"materialId"`
AccountID int64 `json:"accountId"`
CorporationName string `json:"corporationName"`
PreviewURL string `json:"previewUrl"`
Description string `json:"description"`
ErrorMsg string `json:"errorMsg"`
MaterialType string `json:"materialType"`
ImageUsage string `json:"imageUsage,omitempty"`
CreatedAt string `json:"createdAt"`
}
// getFailureReason 获取失败原因
@@ -648,11 +703,9 @@ func getFailureReason(log *entity.MaterialVerifyLog) string {
if log == nil {
return "无校验日志"
}
// 优先使用 error_msg
if log.ErrorMsg != "" {
return log.ErrorMsg
}
// 根据 suggestion 和 label 生成原因
reasonMap := map[int]string{
0: "内容检测通过",
1: "内容嫌疑（需人工审核）",
@@ -662,7 +715,6 @@ func getFailureReason(log *entity.MaterialVerifyLog) string {
if suggestionText == "" {
suggestionText = fmt.Sprintf("未知(suggestion=%d)", log.Suggestion)
}
// 如果有 response_result，尝试提取更多信息
if log.ResponseResult != "" {
var resultMap map[string]interface{}
if err := json.Unmarshal([]byte(log.ResponseResult), &resultMap); err == nil {
@@ -677,7 +729,7 @@ func getFailureReason(log *entity.MaterialVerifyLog) string {
const exportBatchSize = 1000
// ExportRejectedData 导出不通过数据（分批加载，避免OOM）
// ExportRejectedData 导出不通过数据
func (s *MaterialVerifyService) ExportRejectedData(ctx context.Context, materialType string) ([]ExportRejectedItem, error) {
var items []ExportRejectedItem
@@ -694,15 +746,13 @@ func (s *MaterialVerifyService) ExportRejectedData(ctx context.Context, material
condition := map[string]interface{}{
entity.TencentImageCols.VerifyStatus: entity.VerifyStatusRejected,
}
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)
return nil, fmt.Errorf("查询不通过图片失败: %w", err)
}
for _, img := range images {
log, _ := dao.MaterialVerifyLog.GetLastRejectedLogByMaterialID(ctx, img.ImageID, entity.VerifyStatusRejected)
var createdAtStr string
@@ -710,19 +760,13 @@ func (s *MaterialVerifyService) ExportRejectedData(ctx context.Context, material
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,
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
}
@@ -734,15 +778,13 @@ func (s *MaterialVerifyService) ExportRejectedData(ctx context.Context, material
condition := map[string]interface{}{
entity.TencentVideoCols.VerifyStatus: entity.VerifyStatusRejected,
}
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)
return nil, fmt.Errorf("查询不通过视频失败: %w", err)
}
for _, vid := range videos {
log, _ := dao.MaterialVerifyLog.GetLastRejectedLogByMaterialID(ctx, vid.VideoID, entity.VerifyStatusRejected)
var createdAtStr string
@@ -750,18 +792,12 @@ func (s *MaterialVerifyService) ExportRejectedData(ctx context.Context, material
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,
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
}
@@ -776,3 +812,36 @@ func (s *MaterialVerifyService) ExportRejectedData(ctx context.Context, material
func (s *MaterialVerifyService) GetPendingResultsCount(ctx context.Context) (int, error) {
return dao.MaterialVerifyLog.CountPendingResults(ctx)
}
// GetPendingResultsDetail 获取待查询结果的明细列表
type PendingResultItem struct {
LogID int64 `json:"logId"`
MaterialID string `json:"materialId"`
MaterialType string `json:"materialType"`
SourceTable string `json:"sourceTable"`
TaskID string `json:"taskId"`
CreatedAt string `json:"createdAt"`
}
func (s *MaterialVerifyService) GetPendingResultsDetail(ctx context.Context, limit int) ([]PendingResultItem, error) {
logs, err := dao.MaterialVerifyLog.GetPendingResults(ctx, limit)
if err != nil {
return nil, err
}
var items []PendingResultItem
for _, l := range logs {
createdAt := ""
if l.CreatedAt != nil {
createdAt = l.CreatedAt.Format("2006-01-02 15:04:05")
}
items = append(items, PendingResultItem{
LogID: l.Id,
MaterialID: l.MaterialID,
MaterialType: l.MaterialType,
SourceTable: l.SourceTable,
TaskID: l.TaskID,
CreatedAt: createdAt,
})
}
return items, nil
}
service/dataengine/tencent_content_check_service.go
+143 -46
View File
@@ -7,10 +7,12 @@ import (
yidunService "cid/service/yidun"
"context"
"encoding/json"
"sync"
"time"
"gitea.redpowerfuture.com/red-future/common/beans"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/os/grpool"
)
// ContentCheckConfig 送检配置
@@ -19,6 +21,7 @@ type ContentCheckConfig struct {
ImageEnabled bool `json:"image_enabled"`
VideoEnabled bool `json:"video_enabled"`
IntervalSeconds int `json:"interval_seconds"`
PollInterval int `json:"poll_interval"` // 自动轮询检测结果间隔（秒）
}
// DefaultConfig 默认配置
@@ -27,12 +30,16 @@ var DefaultConfig = ContentCheckConfig{
ImageEnabled: true,
VideoEnabled: true,
IntervalSeconds: 30,
PollInterval: 60,
}
// TencentContentCheckService 腾讯内容送检服务
type TencentContentCheckService struct {
mu sync.RWMutex
config ContentCheckConfig
isRunning bool
pool *grpool.Pool
cancel context.CancelFunc
}
// TencentContentCheck 送检服务单例
@@ -42,47 +49,92 @@ var TencentContentCheck = &TencentContentCheckService{
// SetConfig 设置配置
func (s *TencentContentCheckService) SetConfig(config ContentCheckConfig) {
s.mu.Lock()
s.config = config
s.mu.Unlock()
}
// Start 启动定时任务
func (s *TencentContentCheckService) Start(ctx context.Context) error {
s.mu.Lock()
if s.isRunning {
s.mu.Unlock()
g.Log().Info(ctx, "送检服务已在运行中，跳过启动")
return nil
}
s.isRunning = true
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)
config := s.config
s.pool = grpool.New(5)
s.mu.Unlock()
schedCtx := context.Background()
if user := ctx.Value("user"); user != nil {
schedCtx = context.WithValue(schedCtx, "user", user)
g.Log().Infof(ctx, "启动内容送检服务，配置: batch_size=%d, interval=%ds, poll=%ds, image=%v, video=%v",
config.BatchSize, config.IntervalSeconds, config.PollInterval, config.ImageEnabled, config.VideoEnabled)
schedCtx, cancel := context.WithCancel(context.Background())
s.cancel = cancel
// 定时送检协程
g.Go(schedCtx, func(ctx context.Context) {
ticker := time.NewTicker(time.Duration(config.IntervalSeconds) * time.Second)
defer ticker.Stop()
// 启动时先执行一次
_ = s.pool.Add(ctx, func(jobCtx context.Context) {
s.processAll(jobCtx)
})
for {
select {
case <-ticker.C:
_ = s.pool.Add(ctx, func(jobCtx context.Context) {
s.processAll(jobCtx)
})
case <-ctx.Done():
s.pool.Close()
return
}
}
}, nil)
// 自动轮询检测结果协程（无论回调/轮询模式，都定期查一次未处理的结果）
pollInterval := config.PollInterval
if pollInterval <= 0 {
pollInterval = 60
}
go s.runScheduler(schedCtx)
g.Go(schedCtx, func(ctx context.Context) {
pollTicker := time.NewTicker(time.Duration(pollInterval) * time.Second)
defer pollTicker.Stop()
g.Log().Infof(ctx, "启动自动轮询检测结果, 间隔=%ds", pollInterval)
for {
select {
case <-pollTicker.C:
_, _, _ = MaterialVerify.PollPendingResults(ctx)
case <-ctx.Done():
return
}
}
}, nil)
return nil
}
// Stop 停止定时任务
func (s *TencentContentCheckService) Stop(ctx context.Context) {
s.isRunning = false
g.Log().Info(ctx, "停止内容送检服务")
}
// runScheduler 定时调度器
func (s *TencentContentCheckService) runScheduler(ctx context.Context) {
ticker := time.NewTicker(time.Duration(s.config.IntervalSeconds) * time.Second)
defer ticker.Stop()
s.processAll(ctx)
for range ticker.C {
if !s.isRunning {
return
}
s.processAll(ctx)
s.mu.Lock()
if !s.isRunning {
s.mu.Unlock()
return
}
s.isRunning = false
s.mu.Unlock()
if s.cancel != nil {
s.cancel()
}
g.Log().Info(ctx, "停止内容送检服务")
}
// processAll 处理所有待送检数据
@@ -94,18 +146,33 @@ func (s *TencentContentCheckService) processAll(ctx context.Context) {
var totalProcessed int
if s.config.ImageEnabled {
imageCount, _ := dao.TencentImage.CountPending(ctx)
if imageCount > 0 {
count, _ := s.processImages(ctx)
s.mu.RLock()
imageEnabled := s.config.ImageEnabled
videoEnabled := s.config.VideoEnabled
s.mu.RUnlock()
if imageEnabled {
imageCount, err := dao.TencentImage.CountPending(ctx)
if err != nil {
g.Log().Errorf(ctx, "统计待送检图片数量失败: %v", err)
} else if imageCount > 0 {
count, procErr := s.processImages(ctx)
if procErr != nil {
g.Log().Errorf(ctx, "图片送检处理失败: %v", procErr)
}
totalProcessed += count
}
}
if s.config.VideoEnabled {
videoCount, _ := dao.TencentVideo.CountPending(ctx)
if videoCount > 0 {
count, _ := s.processVideos(ctx)
if videoEnabled {
videoCount, err := dao.TencentVideo.CountPending(ctx)
if err != nil {
g.Log().Errorf(ctx, "统计待送检视频数量失败: %v", err)
} else if videoCount > 0 {
count, procErr := s.processVideos(ctx)
if procErr != nil {
g.Log().Errorf(ctx, "视频送检处理失败: %v", procErr)
}
totalProcessed += count
}
}
@@ -116,7 +183,11 @@ func (s *TencentContentCheckService) processAll(ctx context.Context) {
// processImages 处理图片送检（统一走 MaterialVerify 系统）
func (s *TencentContentCheckService) processImages(ctx context.Context) (int, error) {
images, err := dao.TencentImage.GetPendingList(ctx, s.config.BatchSize)
s.mu.RLock()
batchSize := s.config.BatchSize
s.mu.RUnlock()
images, err := dao.TencentImage.GetPendingList(ctx, batchSize)
if err != nil {
g.Log().Errorf(ctx, "获取待送检图片失败: %v", err)
return 0, err
@@ -132,15 +203,12 @@ func (s *TencentContentCheckService) processImages(ctx context.Context) (int, er
failedCount := 0
for _, img := range images {
// 统一走 MaterialVerify 系统提交（处理完整校验流程：日志→提交→状态反写）
mLog, err := MaterialVerify.VerifyImageByID(ctx, img.ImageID)
_, err := MaterialVerify.VerifyImageByID(ctx, img.ImageID)
if err != nil {
failedCount++
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)
@@ -152,7 +220,11 @@ func (s *TencentContentCheckService) processImages(ctx context.Context) (int, er
// processVideos 处理视频送检（统一走 MaterialVerify 系统）
func (s *TencentContentCheckService) processVideos(ctx context.Context) (int, error) {
videos, err := dao.TencentVideo.GetPendingList(ctx, s.config.BatchSize)
s.mu.RLock()
batchSize := s.config.BatchSize
s.mu.RUnlock()
videos, err := dao.TencentVideo.GetPendingList(ctx, batchSize)
if err != nil {
g.Log().Errorf(ctx, "获取待送检视频失败: %v", err)
return 0, err
@@ -168,13 +240,12 @@ func (s *TencentContentCheckService) processVideos(ctx context.Context) (int, er
failedCount := 0
for _, video := range videos {
mLog, err := MaterialVerify.VerifyVideoByID(ctx, video.VideoID)
_, err := MaterialVerify.VerifyVideoByID(ctx, video.VideoID)
if err != nil {
failedCount++
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)
@@ -185,21 +256,32 @@ func (s *TencentContentCheckService) processVideos(ctx context.Context) (int, er
}
// writeAuditLog 写入审计日志（tencent_content_check_log）
func (s *TencentContentCheckService) writeAuditLog(ctx context.Context, sourceTable string, sourceID int64, mediaID string, mediaURL string, taskID string) {
// 当 suggestion<0 时记录为 SUBMITTING（已提交等待结果），否则记录为 COMPLETED（检测完成）
func (s *TencentContentCheckService) writeAuditLog(ctx context.Context, sourceTable string, sourceID int64, mediaID string, mediaURL string, taskID string, suggestion, label, resultType int, responseData string, duration int64) {
requestParam := map[string]interface{}{
"media_id": mediaID,
"url": mediaURL,
}
requestParamJSON, _ := json.Marshal(requestParam)
status := consts.CheckStatusSubmitting
if suggestion >= 0 {
status = consts.CheckStatusCompleted
}
log := &entity.TencentContentCheckLog{
SourceTable: sourceTable,
SourceID: sourceID,
RequestURL: "易盾内容安全检测接口",
RequestParam: string(requestParamJSON),
Status: consts.CheckStatusSuccess,
Status: status,
CheckTime: time.Now().UnixMilli(),
TaskID: taskID,
Suggestion: suggestion,
Label: label,
ResultType: resultType,
ResponseData: responseData,
Duration: duration,
}
id, err := dao.TencentContentCheckLog.Create(ctx, log)
@@ -220,7 +302,7 @@ func (s *TencentContentCheckService) SubmitImageByID(ctx context.Context, imageI
image, err := dao.TencentImage.GetByImageID(ctx, imageID)
if err == nil && image != nil {
s.writeAuditLog(ctx, consts.SourceTableTencentImage, image.Id, imageID, image.PreviewURL, mLog.TaskID)
s.writeAuditLog(ctx, consts.SourceTableTencentImage, image.Id, imageID, image.PreviewURL, mLog.TaskID, -1, 0, 0, "", 0)
}
return &yidunService.ImageSubmitResult{
@@ -237,7 +319,7 @@ func (s *TencentContentCheckService) SubmitVideoByID(ctx context.Context, videoI
video, err := dao.TencentVideo.GetByVideoID(ctx, videoID)
if err == nil && video != nil {
s.writeAuditLog(ctx, consts.SourceTableTencentVideo, video.Id, videoID, video.PreviewURL, mLog.TaskID)
s.writeAuditLog(ctx, consts.SourceTableTencentVideo, video.Id, videoID, video.PreviewURL, mLog.TaskID, -1, 0, 0, "", 0)
}
return &yidunService.VideoSubmitResult{
@@ -249,13 +331,24 @@ func (s *TencentContentCheckService) SubmitVideoByID(ctx context.Context, videoI
func (s *TencentContentCheckService) GetPendingStats(ctx context.Context) map[string]int {
stats := make(map[string]int)
if s.config.ImageEnabled {
count, _ := dao.TencentImage.CountPending(ctx)
s.mu.RLock()
imageEnabled := s.config.ImageEnabled
videoEnabled := s.config.VideoEnabled
s.mu.RUnlock()
if imageEnabled {
count, err := dao.TencentImage.CountPending(ctx)
if err != nil {
g.Log().Errorf(ctx, "统计待送检图片数量失败: %v", err)
}
stats["image_pending"] = count
}
if s.config.VideoEnabled {
count, _ := dao.TencentVideo.CountPending(ctx)
if videoEnabled {
count, err := dao.TencentVideo.CountPending(ctx)
if err != nil {
g.Log().Errorf(ctx, "统计待送检视频数量失败: %v", err)
}
stats["video_pending"] = count
}
@@ -264,10 +357,14 @@ func (s *TencentContentCheckService) GetPendingStats(ctx context.Context) map[st
// IsRunning 获取运行状态
func (s *TencentContentCheckService) IsRunning() bool {
s.mu.RLock()
defer s.mu.RUnlock()
return s.isRunning
}
// GetConfig 获取当前配置
func (s *TencentContentCheckService) GetConfig() ContentCheckConfig {
s.mu.RLock()
defer s.mu.RUnlock()
return s.config
}