274 lines
7.7 KiB
Go
274 lines
7.7 KiB
Go
package dataengine
|
||
|
||
import (
|
||
consts "cid/consts/dataengine"
|
||
dao "cid/dao/dataengine"
|
||
entity "cid/model/entity/dataengine"
|
||
yidunService "cid/service/yidun"
|
||
"context"
|
||
"encoding/json"
|
||
"time"
|
||
|
||
"gitea.redpowerfuture.com/red-future/common/beans"
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
)
|
||
|
||
// ContentCheckConfig 送检配置
|
||
type ContentCheckConfig struct {
|
||
BatchSize int `json:"batch_size"`
|
||
ImageEnabled bool `json:"image_enabled"`
|
||
VideoEnabled bool `json:"video_enabled"`
|
||
IntervalSeconds int `json:"interval_seconds"`
|
||
}
|
||
|
||
// DefaultConfig 默认配置
|
||
var DefaultConfig = ContentCheckConfig{
|
||
BatchSize: 10,
|
||
ImageEnabled: true,
|
||
VideoEnabled: true,
|
||
IntervalSeconds: 30,
|
||
}
|
||
|
||
// TencentContentCheckService 腾讯内容送检服务
|
||
type TencentContentCheckService struct {
|
||
config ContentCheckConfig
|
||
isRunning bool
|
||
}
|
||
|
||
// TencentContentCheck 送检服务单例
|
||
var TencentContentCheck = &TencentContentCheckService{
|
||
config: DefaultConfig,
|
||
}
|
||
|
||
// SetConfig 设置配置
|
||
func (s *TencentContentCheckService) SetConfig(config ContentCheckConfig) {
|
||
s.config = config
|
||
}
|
||
|
||
// Start 启动定时任务
|
||
func (s *TencentContentCheckService) Start(ctx context.Context) error {
|
||
if s.isRunning {
|
||
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)
|
||
|
||
schedCtx := context.Background()
|
||
if user := ctx.Value("user"); user != nil {
|
||
schedCtx = context.WithValue(schedCtx, "user", user)
|
||
}
|
||
go s.runScheduler(schedCtx)
|
||
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)
|
||
}
|
||
}
|
||
|
||
// processAll 处理所有待送检数据
|
||
func (s *TencentContentCheckService) processAll(ctx context.Context) {
|
||
ctx = context.WithValue(ctx, "user", &beans.User{UserName: "system", TenantId: 1})
|
||
|
||
startTime := time.Now()
|
||
g.Log().Info(ctx, "开始处理待送检数据...")
|
||
|
||
var totalProcessed int
|
||
|
||
if s.config.ImageEnabled {
|
||
imageCount, _ := dao.TencentImage.CountPending(ctx)
|
||
if imageCount > 0 {
|
||
count, _ := s.processImages(ctx)
|
||
totalProcessed += count
|
||
}
|
||
}
|
||
|
||
if s.config.VideoEnabled {
|
||
videoCount, _ := dao.TencentVideo.CountPending(ctx)
|
||
if videoCount > 0 {
|
||
count, _ := s.processVideos(ctx)
|
||
totalProcessed += count
|
||
}
|
||
}
|
||
|
||
duration := time.Since(startTime).Milliseconds()
|
||
g.Log().Infof(ctx, "处理完成，共处理 %d 条数据，耗时 %dms", totalProcessed, duration)
|
||
}
|
||
|
||
// 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)
|
||
return 0, err
|
||
}
|
||
|
||
if len(images) == 0 {
|
||
return 0, nil
|
||
}
|
||
|
||
g.Log().Infof(ctx, "开始送检 %d 张图片", len(images))
|
||
|
||
successCount := 0
|
||
failedCount := 0
|
||
|
||
for _, img := range images {
|
||
// 统一走 MaterialVerify 系统提交（处理完整校验流程：日志→提交→状态反写）
|
||
mLog, 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)
|
||
}
|
||
|
||
g.Log().Infof(ctx, "图片送检完成，成功: %d，失败: %d", successCount, failedCount)
|
||
return len(images), nil
|
||
}
|
||
|
||
// 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)
|
||
return 0, err
|
||
}
|
||
|
||
if len(videos) == 0 {
|
||
return 0, nil
|
||
}
|
||
|
||
g.Log().Infof(ctx, "开始送检 %d 个视频", len(videos))
|
||
|
||
successCount := 0
|
||
failedCount := 0
|
||
|
||
for _, video := range videos {
|
||
mLog, 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)
|
||
}
|
||
|
||
g.Log().Infof(ctx, "视频送检完成，成功: %d，失败: %d", successCount, failedCount)
|
||
return len(videos), nil
|
||
}
|
||
|
||
// 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,
|
||
}
|
||
requestParamJSON, _ := json.Marshal(requestParam)
|
||
|
||
log := &entity.TencentContentCheckLog{
|
||
SourceTable: sourceTable,
|
||
SourceID: sourceID,
|
||
RequestURL: "易盾内容安全检测接口",
|
||
RequestParam: string(requestParamJSON),
|
||
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
|
||
}
|
||
|
||
g.Log().Debugf(ctx, "创建送检审计日志成功, id=%d, sourceTable=%s, sourceID=%d, taskId=%s", id, sourceTable, sourceID, taskID)
|
||
}
|
||
|
||
// SubmitImageByID 根据图片ID手动提交送检（统一走 MaterialVerify 系统）
|
||
func (s *TencentContentCheckService) SubmitImageByID(ctx context.Context, imageID string) (*yidunService.ImageSubmitResult, error) {
|
||
mLog, err := MaterialVerify.VerifyImageByID(ctx, imageID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
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手动提交送检（统一走 MaterialVerify 系统）
|
||
func (s *TencentContentCheckService) SubmitVideoByID(ctx context.Context, videoID string) (*yidunService.VideoSubmitResult, error) {
|
||
mLog, err := MaterialVerify.VerifyVideoByID(ctx, videoID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
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 获取待送检统计
|
||
func (s *TencentContentCheckService) GetPendingStats(ctx context.Context) map[string]int {
|
||
stats := make(map[string]int)
|
||
|
||
if s.config.ImageEnabled {
|
||
count, _ := dao.TencentImage.CountPending(ctx)
|
||
stats["image_pending"] = count
|
||
}
|
||
|
||
if s.config.VideoEnabled {
|
||
count, _ := dao.TencentVideo.CountPending(ctx)
|
||
stats["video_pending"] = count
|
||
}
|
||
|
||
return stats
|
||
}
|
||
|
||
// IsRunning 获取运行状态
|
||
func (s *TencentContentCheckService) IsRunning() bool {
|
||
return s.isRunning
|
||
}
|
||
|
||
// GetConfig 获取当前配置
|
||
func (s *TencentContentCheckService) GetConfig() ContentCheckConfig {
|
||
return s.config
|
||
}
|