130 lines
3.8 KiB
Go
130 lines
3.8 KiB
Go
package contentreviewser
|
|
|
|
import (
|
|
"fmt"
|
|
"runtime/debug"
|
|
"sync/atomic"
|
|
|
|
"91porn-server/common/log"
|
|
"91porn-server/models/v/contentreviewmod"
|
|
"91porn-server/models/v/sensitivewordmod"
|
|
"91porn-server/skd/service/export_task"
|
|
|
|
tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api"
|
|
)
|
|
|
|
// tgChatID 内容审查任务完成通知群（与导出任务复用同一群）
|
|
const tgChatID int64 = -1003399433452
|
|
|
|
var runningFlag int32 // 进程内互斥：0=空闲 1=运行中
|
|
|
|
// TryRunPendingTask 调度入口，由 skd 的 cron 定时调用
|
|
// skd 单进程，仅需 atomic 互斥防止 cron tick 重入；每次只挑一个最早的任务处理
|
|
// 取到 pending 视为首跑；取到心跳超时的 running 视为续跑，从 lastTargetId 继续
|
|
func TryRunPendingTask() {
|
|
if !atomic.CompareAndSwapInt32(&runningFlag, 0, 1) {
|
|
return
|
|
}
|
|
defer atomic.StoreInt32(&runningFlag, 0)
|
|
|
|
task, err := contentreviewmod.PickNextTask()
|
|
if err != nil {
|
|
log.Warn("content review: pick next fail", log.E(err))
|
|
return
|
|
}
|
|
if task == nil {
|
|
return
|
|
}
|
|
if task.ID.IsZero() {
|
|
return
|
|
}
|
|
isResume := task.Status == contentreviewmod.TaskStatusRunning
|
|
if err := contentreviewmod.AcquireTask(task.ID, isResume); err != nil {
|
|
log.Warn("content review: acquire task fail", log.Any("taskId", task.ID), log.E(err))
|
|
return
|
|
}
|
|
if isResume {
|
|
log.Info("content review: resume stale task",
|
|
log.Any("taskId", task.ID), log.Any("lastTargetId", task.LastTargetID))
|
|
}
|
|
|
|
executeTask(task)
|
|
}
|
|
|
|
// executeTask 真正执行扫描
|
|
func executeTask(task *contentreviewmod.ReviewTask) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
stack := string(debug.Stack())
|
|
log.Error("content review task panic",
|
|
log.Any("taskId", task.ID), log.Any("panic", r), log.Any("stack", stack))
|
|
_ = contentreviewmod.MarkFailed(task.ID, fmt.Sprintf("panic: %v", r))
|
|
}
|
|
}()
|
|
|
|
log.Info("content review task start",
|
|
log.Any("taskId", task.ID), log.Any("type", task.TaskType))
|
|
|
|
checked, issues, err := runScan(task)
|
|
if err != nil {
|
|
log.Error("content review task fail",
|
|
log.Any("taskId", task.ID), log.E(err))
|
|
_ = contentreviewmod.MarkFailed(task.ID, err.Error())
|
|
return
|
|
}
|
|
|
|
if err := contentreviewmod.MarkFinished(task.ID, checked, issues); err != nil {
|
|
log.Error("content review mark finished fail", log.E(err))
|
|
return
|
|
}
|
|
|
|
// 发送 TG
|
|
if sendTGNotify(task, checked, issues) {
|
|
_ = contentreviewmod.MarkTgSent(task.ID)
|
|
}
|
|
|
|
log.Info("content review task done",
|
|
log.Any("taskId", task.ID), log.Any("checked", checked), log.Any("issues", issues))
|
|
}
|
|
|
|
func sendTGNotify(task *contentreviewmod.ReviewTask, checked, issues int64) bool {
|
|
if export_task.Bot == nil {
|
|
log.Warn("content review: tg bot not initialized, skip notify")
|
|
return false
|
|
}
|
|
typeName := "视频帖子"
|
|
if task.TaskType == contentreviewmod.TaskTypeACG {
|
|
typeName = "ACG"
|
|
}
|
|
operator := task.Operator
|
|
if operator == "" {
|
|
operator = "-"
|
|
}
|
|
text := fmt.Sprintf(
|
|
"【%s 内容检测任务完成】\n操作人: %s\n类型: %s\n任务ID: %s\n总记录: %d\n已检测: %d\n命中: %d",
|
|
export_task.GetProName(), operator, typeName, task.ID.Hex(), task.TotalCount, checked, issues,
|
|
)
|
|
if _, err := export_task.Bot.Send(tgbotapi.NewMessage(tgChatID, text)); err != nil {
|
|
log.Warn("content review: send tg fail", log.Any("taskId", task.ID), log.E(err))
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
// BuildSnapshot 拉取当前启用的敏感词作为快照（web 创建任务时复用）
|
|
func BuildSnapshot() ([]contentreviewmod.SensitiveWordSnap, error) {
|
|
enabled := sensitivewordmod.StatusEnabled
|
|
list, err := sensitivewordmod.FindAll(&sensitivewordmod.ListReq{Status: &enabled})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := make([]contentreviewmod.SensitiveWordSnap, 0, len(list))
|
|
for _, w := range list {
|
|
out = append(out, contentreviewmod.SensitiveWordSnap{
|
|
Word: w.Word,
|
|
Category: w.Category,
|
|
})
|
|
}
|
|
return out, nil
|
|
}
|