96 lines
2.7 KiB
Go
96 lines
2.7 KiB
Go
package contentreviewser
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
"91porn-server/common/log"
|
|
"91porn-server/models/v/contentreviewmod"
|
|
|
|
"go.mongodb.org/mongo-driver/bson/primitive"
|
|
)
|
|
|
|
const scanBatchSize = 100
|
|
|
|
// ScanResult 一批扫描的产出
|
|
type ScanResult struct {
|
|
Issues []*contentreviewmod.ReviewIssue // 本批生成的命中记录
|
|
Checked int64 // 本批实际扫描的目标数量
|
|
NextID primitive.ObjectID // 下一批的游标
|
|
Done bool // true 表示已扫完，runScan 退出循环
|
|
}
|
|
|
|
// Scanner 内容审查扫描策略；具体任务类型实现各自的 ScanBatch
|
|
type Scanner interface {
|
|
CountTotal() (int64, error)
|
|
ScanBatch(taskID, lastID primitive.ObjectID, terms []string) (*ScanResult, error)
|
|
}
|
|
|
|
// newScanner Strategy 工厂：根据任务类型返回具体扫描器
|
|
func newScanner(taskType string) (Scanner, error) {
|
|
switch taskType {
|
|
case contentreviewmod.TaskTypeVideo:
|
|
return &videoScanner{}, nil
|
|
case contentreviewmod.TaskTypeACG:
|
|
return &acgScanner{}, nil
|
|
default:
|
|
return nil, fmt.Errorf("unsupported task type: %s", taskType)
|
|
}
|
|
}
|
|
|
|
// runScan Template Method：把"分页游标 + 进度回写 + 批量插入"骨架抽出，差异由 Scanner 提供
|
|
// 支持从 task.LastTargetID 断点续跑
|
|
func runScan(task *contentreviewmod.ReviewTask) (int64, int64, error) {
|
|
scanner, err := newScanner(task.TaskType)
|
|
if err != nil {
|
|
return task.CheckedCount, task.IssueCount, err
|
|
}
|
|
|
|
terms := allTerms(task.SnapshotWords)
|
|
|
|
total := task.TotalCount
|
|
if total <= 0 {
|
|
t, err := scanner.CountTotal()
|
|
if err != nil {
|
|
return task.CheckedCount, task.IssueCount, fmt.Errorf("count total: %w", err)
|
|
}
|
|
total = t
|
|
}
|
|
// 同步到内存对象，便于 executeTask 拼 TG 消息时正确取值
|
|
task.TotalCount = total
|
|
|
|
lastID := task.LastTargetID
|
|
checked := task.CheckedCount
|
|
issues := task.IssueCount
|
|
|
|
flushProgress(task.ID, lastID, checked, issues, total)
|
|
|
|
for {
|
|
result, err := scanner.ScanBatch(task.ID, lastID, terms)
|
|
if err != nil {
|
|
return checked, issues, fmt.Errorf("scan batch: %w", err)
|
|
}
|
|
if result.Done {
|
|
break
|
|
}
|
|
if len(result.Issues) > 0 {
|
|
if err := contentreviewmod.InsertIssues(result.Issues); err != nil {
|
|
log.Warn("insert issues fail", log.Any("count", len(result.Issues)), log.E(err))
|
|
} else {
|
|
issues += int64(len(result.Issues))
|
|
}
|
|
}
|
|
checked += result.Checked
|
|
lastID = result.NextID
|
|
|
|
flushProgress(task.ID, lastID, checked, issues, total)
|
|
}
|
|
return checked, issues, nil
|
|
}
|
|
|
|
func flushProgress(id, lastID primitive.ObjectID, checked, issues, total int64) {
|
|
if err := contentreviewmod.UpdateProgress(id, lastID, checked, issues, total); err != nil {
|
|
log.Warn("update progress fail", log.E(err))
|
|
}
|
|
}
|
|
|