李建琦
6d655c9903
Release / update-version (push) Has been cancelled
Release / build-frontend (push) Has been cancelled
Release / release (push) Has been cancelled
Release / sync-version-file (push) Has been cancelled
CI / shell (push) Canceled after 0s
CI / test (push) Canceled after 0s
CI / frontend (push) Canceled after 0s
CI / golangci-lint (push) Canceled after 0s
Security Scan / backend-security (push) Canceled after 0s
Security Scan / frontend-security (push) Canceled after 0s
3106 lines
106 KiB
Go
3106 lines
106 KiB
Go
package service
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/base64"
|
||
"encoding/binary"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"log/slog"
|
||
"net/http"
|
||
"net/url"
|
||
"sort"
|
||
"strings"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
|
||
"github.com/Wei-Shaw/sub2api/internal/pkg/httpclient"
|
||
"github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
|
||
"github.com/Wei-Shaw/sub2api/internal/pkg/servertiming"
|
||
)
|
||
|
||
const (
|
||
ContentModerationModeOff = "off"
|
||
ContentModerationModeObserve = "observe"
|
||
ContentModerationModePreBlock = "pre_block"
|
||
|
||
contentModerationAPIKeysModeAppend = "append"
|
||
contentModerationAPIKeysModeReplace = "replace"
|
||
|
||
ContentModerationActionAllow = "allow"
|
||
ContentModerationActionBlock = "block"
|
||
ContentModerationActionHashBlock = "hash_block"
|
||
ContentModerationActionKeywordBlock = "keyword_block"
|
||
ContentModerationActionError = "error"
|
||
ContentModerationActionCyberPolicy = "cyber_policy" // cyber_policy 硬阻断的风控日志 action（封号计数排除按此值过滤）
|
||
|
||
contentModerationKeywordCategory = "keyword"
|
||
|
||
ContentModerationKeywordModeKeywordOnly = "keyword_only"
|
||
ContentModerationKeywordModeKeywordAndAPI = "keyword_and_api"
|
||
ContentModerationKeywordModeAPIOnly = "api_only"
|
||
|
||
ContentModerationModelFilterAll = "all"
|
||
ContentModerationModelFilterInclude = "include"
|
||
ContentModerationModelFilterExclude = "exclude"
|
||
|
||
ContentModerationProtocolAnthropicMessages = "anthropic_messages"
|
||
ContentModerationProtocolOpenAIResponses = "openai_responses"
|
||
ContentModerationProtocolOpenAIChat = "openai_chat_completions"
|
||
ContentModerationProtocolGemini = "gemini"
|
||
ContentModerationProtocolOpenAIImages = "openai_images"
|
||
|
||
defaultContentModerationBaseURL = "https://api.openai.com"
|
||
defaultContentModerationModel = "omni-moderation-latest"
|
||
defaultContentModerationTimeoutMS = 3000
|
||
maxContentModerationTimeoutMS = 30000
|
||
maxModerationInputRunes = 12000
|
||
maxModerationExcerptRunes = 240
|
||
|
||
defaultContentModerationWorkerCount = 4
|
||
maxContentModerationWorkerCount = 32
|
||
defaultContentModerationQueueSize = 32768
|
||
maxContentModerationQueueSize = 100000
|
||
defaultContentModerationBanThreshold = 10
|
||
defaultContentModerationViolationWindowHours = 720
|
||
defaultContentModerationBlockHTTPStatus = http.StatusForbidden
|
||
defaultContentModerationBlockMessage = "内容审计命中风险规则，请调整输入后重试"
|
||
defaultContentModerationRetryCount = 2
|
||
maxContentModerationRetryCount = 5
|
||
defaultContentModerationHitRetentionDays = 180
|
||
defaultContentModerationNonHitRetentionDays = 3
|
||
maxContentModerationRetentionDays = 3650
|
||
maxContentModerationNonHitRetentionDays = 3
|
||
contentModerationKeyRateLimitFreezeDuration = time.Minute
|
||
contentModerationKeyAuthFreezeDuration = 10 * time.Minute
|
||
contentModerationKeyHTTPErrorFreezeDuration = 10 * time.Second
|
||
maxContentModerationInputImages = 1
|
||
maxContentModerationTestImages = maxContentModerationInputImages
|
||
maxContentModerationTestImageBytes = 8 * 1024 * 1024
|
||
maxContentModerationTestImageDataURLBytes = 12 * 1024 * 1024
|
||
maxContentModerationBlockedKeywords = 10000
|
||
maxContentModerationBlockedKeywordRunes = 200
|
||
maxContentModerationModelFilterModels = 1000
|
||
maxContentModerationModelFilterRunes = 200
|
||
|
||
contentModerationCleanupInterval = 24 * time.Hour
|
||
contentModerationCleanupTimeout = 30 * time.Minute
|
||
contentModerationCleanupDelay = 5 * time.Minute
|
||
|
||
contentModerationRuntimeCacheTTL = time.Second
|
||
contentModerationRuntimeRefreshTimeout = 5 * time.Second
|
||
)
|
||
|
||
var contentModerationCategoryOrder = []string{
|
||
"harassment",
|
||
"harassment/threatening",
|
||
"hate",
|
||
"hate/threatening",
|
||
"illicit",
|
||
"illicit/violent",
|
||
"self-harm",
|
||
"self-harm/intent",
|
||
"self-harm/instructions",
|
||
"sexual",
|
||
"sexual/minors",
|
||
"violence",
|
||
"violence/graphic",
|
||
}
|
||
|
||
func ContentModerationDefaultThresholds() map[string]float64 {
|
||
return map[string]float64{
|
||
"harassment": 0.98,
|
||
"harassment/threatening": 0.90,
|
||
"hate": 0.65,
|
||
"hate/threatening": 0.65,
|
||
"illicit": 0.95,
|
||
"illicit/violent": 0.95,
|
||
"self-harm": 0.65,
|
||
"self-harm/intent": 0.85,
|
||
"self-harm/instructions": 0.65,
|
||
"sexual": 0.65,
|
||
"sexual/minors": 0.65,
|
||
"violence": 0.95,
|
||
"violence/graphic": 0.95,
|
||
}
|
||
}
|
||
|
||
func ContentModerationCategories() []string {
|
||
out := make([]string, len(contentModerationCategoryOrder))
|
||
copy(out, contentModerationCategoryOrder)
|
||
return out
|
||
}
|
||
|
||
type ContentModerationConfig struct {
|
||
Enabled bool `json:"enabled"`
|
||
Mode string `json:"mode"`
|
||
BaseURL string `json:"base_url"`
|
||
Model string `json:"model"`
|
||
// ProxyID 指定审计请求使用的代理服务器（IP管理-代理服务器），nil 表示直连。
|
||
ProxyID *int64 `json:"proxy_id,omitempty"`
|
||
APIKey string `json:"api_key,omitempty"`
|
||
APIKeys []string `json:"api_keys,omitempty"`
|
||
TimeoutMS int `json:"timeout_ms"`
|
||
SampleRate int `json:"sample_rate"`
|
||
AllGroups bool `json:"all_groups"`
|
||
GroupIDs []int64 `json:"group_ids"`
|
||
RecordNonHits bool `json:"record_non_hits"`
|
||
Thresholds map[string]float64 `json:"thresholds"`
|
||
WorkerCount int `json:"worker_count"`
|
||
QueueSize int `json:"queue_size"`
|
||
BlockStatus int `json:"block_status"`
|
||
BlockMessage string `json:"block_message"`
|
||
EmailOnHit bool `json:"email_on_hit"`
|
||
AutoBanEnabled bool `json:"auto_ban_enabled"`
|
||
BanThreshold int `json:"ban_threshold"`
|
||
ViolationWindowHours int `json:"violation_window_hours"`
|
||
RetryCount int `json:"retry_count"`
|
||
HitRetentionDays int `json:"hit_retention_days"`
|
||
NonHitRetentionDays int `json:"non_hit_retention_days"`
|
||
PreHashCheckEnabled bool `json:"pre_hash_check_enabled"`
|
||
BlockedKeywords []string `json:"blocked_keywords"`
|
||
KeywordBlockingMode string `json:"keyword_blocking_mode"`
|
||
ModelFilter ContentModerationModelFilter `json:"model_filter"`
|
||
// CyberPolicyExcludeFromBanCount 为 true 时，cyber_policy 命中不参与自动封号计数：
|
||
// 当次不判定封号，且历史 cyber 行在 CountFlaggedByUserSince 中被排除。
|
||
// 默认 false（计入，与历史行为一致；旧配置 JSON 无此字段时反序列化为 false）。
|
||
CyberPolicyExcludeFromBanCount bool `json:"cyber_policy_exclude_from_ban_count"`
|
||
}
|
||
|
||
type ContentModerationConfigView struct {
|
||
Enabled bool `json:"enabled"`
|
||
Mode string `json:"mode"`
|
||
BaseURL string `json:"base_url"`
|
||
Model string `json:"model"`
|
||
ProxyID *int64 `json:"proxy_id"`
|
||
APIKeyConfigured bool `json:"api_key_configured"`
|
||
APIKeyMasked string `json:"api_key_masked"`
|
||
APIKeyCount int `json:"api_key_count"`
|
||
APIKeyMasks []string `json:"api_key_masks"`
|
||
APIKeyStatuses []ContentModerationAPIKeyStatus `json:"api_key_statuses"`
|
||
TimeoutMS int `json:"timeout_ms"`
|
||
SampleRate int `json:"sample_rate"`
|
||
AllGroups bool `json:"all_groups"`
|
||
GroupIDs []int64 `json:"group_ids"`
|
||
RecordNonHits bool `json:"record_non_hits"`
|
||
Thresholds map[string]float64 `json:"thresholds"`
|
||
WorkerCount int `json:"worker_count"`
|
||
QueueSize int `json:"queue_size"`
|
||
BlockStatus int `json:"block_status"`
|
||
BlockMessage string `json:"block_message"`
|
||
EmailOnHit bool `json:"email_on_hit"`
|
||
AutoBanEnabled bool `json:"auto_ban_enabled"`
|
||
BanThreshold int `json:"ban_threshold"`
|
||
ViolationWindowHours int `json:"violation_window_hours"`
|
||
RetryCount int `json:"retry_count"`
|
||
HitRetentionDays int `json:"hit_retention_days"`
|
||
NonHitRetentionDays int `json:"non_hit_retention_days"`
|
||
PreHashCheckEnabled bool `json:"pre_hash_check_enabled"`
|
||
BlockedKeywords []string `json:"blocked_keywords"`
|
||
KeywordBlockingMode string `json:"keyword_blocking_mode"`
|
||
ModelFilter ContentModerationModelFilter `json:"model_filter"`
|
||
CyberPolicyExcludeFromBanCount bool `json:"cyber_policy_exclude_from_ban_count"`
|
||
}
|
||
|
||
type ContentModerationAPIKeyStatus struct {
|
||
Index int `json:"index"`
|
||
KeyHash string `json:"key_hash"`
|
||
Masked string `json:"masked"`
|
||
Status string `json:"status"`
|
||
FailureCount int `json:"failure_count"`
|
||
SuccessCount int64 `json:"success_count"`
|
||
LastError string `json:"last_error"`
|
||
LastCheckedAt *time.Time `json:"last_checked_at,omitempty"`
|
||
FrozenUntil *time.Time `json:"frozen_until,omitempty"`
|
||
LastLatencyMS int `json:"last_latency_ms"`
|
||
LastHTTPStatus int `json:"last_http_status"`
|
||
LastTested bool `json:"last_tested"`
|
||
Configured bool `json:"configured"`
|
||
}
|
||
|
||
type ContentModerationAPIKeyLoad struct {
|
||
Index int `json:"index"`
|
||
KeyHash string `json:"key_hash"`
|
||
Masked string `json:"masked"`
|
||
Status string `json:"status"`
|
||
Active int64 `json:"active"`
|
||
Total int64 `json:"total"`
|
||
Success int64 `json:"success"`
|
||
Errors int64 `json:"errors"`
|
||
AvgLatencyMS int64 `json:"avg_latency_ms"`
|
||
LastLatencyMS int `json:"last_latency_ms"`
|
||
LastHTTPStatus int `json:"last_http_status"`
|
||
}
|
||
|
||
type TestContentModerationAPIKeysInput struct {
|
||
APIKeys []string `json:"api_keys"`
|
||
BaseURL string `json:"base_url"`
|
||
Model string `json:"model"`
|
||
TimeoutMS int `json:"timeout_ms"`
|
||
// ProxyID nil 表示沿用已保存配置的代理；<=0 表示强制直连测试；>0 表示指定代理测试。
|
||
ProxyID *int64 `json:"proxy_id"`
|
||
Prompt string `json:"prompt"`
|
||
Images []string `json:"images"`
|
||
}
|
||
|
||
type TestContentModerationAPIKeysResult struct {
|
||
Items []ContentModerationAPIKeyStatus `json:"items"`
|
||
AuditResult *ContentModerationTestAuditResult `json:"audit_result,omitempty"`
|
||
ImageCount int `json:"image_count"`
|
||
}
|
||
|
||
type ContentModerationTestAuditResult struct {
|
||
Flagged bool `json:"flagged"`
|
||
HighestCategory string `json:"highest_category"`
|
||
HighestScore float64 `json:"highest_score"`
|
||
CompositeScore float64 `json:"composite_score"`
|
||
CategoryScores map[string]float64 `json:"category_scores"`
|
||
Thresholds map[string]float64 `json:"thresholds"`
|
||
}
|
||
|
||
type UpdateContentModerationConfigInput struct {
|
||
Enabled *bool `json:"enabled"`
|
||
Mode *string `json:"mode"`
|
||
BaseURL *string `json:"base_url"`
|
||
Model *string `json:"model"`
|
||
// ProxyID nil 表示不修改；<=0 表示清除代理（恢复直连）；>0 表示指定代理。
|
||
ProxyID *int64 `json:"proxy_id"`
|
||
APIKey *string `json:"api_key"`
|
||
APIKeys *[]string `json:"api_keys"`
|
||
APIKeysMode string `json:"api_keys_mode"`
|
||
DeleteAPIKeyHashes *[]string `json:"delete_api_key_hashes"`
|
||
ClearAPIKey bool `json:"clear_api_key"`
|
||
TimeoutMS *int `json:"timeout_ms"`
|
||
SampleRate *int `json:"sample_rate"`
|
||
AllGroups *bool `json:"all_groups"`
|
||
GroupIDs *[]int64 `json:"group_ids"`
|
||
RecordNonHits *bool `json:"record_non_hits"`
|
||
Thresholds *map[string]float64 `json:"thresholds"`
|
||
WorkerCount *int `json:"worker_count"`
|
||
QueueSize *int `json:"queue_size"`
|
||
BlockStatus *int `json:"block_status"`
|
||
BlockMessage *string `json:"block_message"`
|
||
EmailOnHit *bool `json:"email_on_hit"`
|
||
AutoBanEnabled *bool `json:"auto_ban_enabled"`
|
||
BanThreshold *int `json:"ban_threshold"`
|
||
ViolationWindowHours *int `json:"violation_window_hours"`
|
||
RetryCount *int `json:"retry_count"`
|
||
HitRetentionDays *int `json:"hit_retention_days"`
|
||
NonHitRetentionDays *int `json:"non_hit_retention_days"`
|
||
PreHashCheckEnabled *bool `json:"pre_hash_check_enabled"`
|
||
BlockedKeywords *[]string `json:"blocked_keywords"`
|
||
KeywordBlockingMode *string `json:"keyword_blocking_mode"`
|
||
ModelFilter *ContentModerationModelFilter `json:"model_filter"`
|
||
CyberPolicyExcludeFromBanCount *bool `json:"cyber_policy_exclude_from_ban_count"`
|
||
}
|
||
|
||
type ContentModerationModelFilter struct {
|
||
Type string `json:"type"`
|
||
Models []string `json:"models"`
|
||
}
|
||
|
||
type ContentModerationCheckInput struct {
|
||
RequestID string
|
||
UserID int64
|
||
UserEmail string
|
||
APIKeyID int64
|
||
APIKeyName string
|
||
GroupID *int64
|
||
GroupName string
|
||
Endpoint string
|
||
Provider string
|
||
Model string
|
||
Protocol string
|
||
Body []byte
|
||
}
|
||
|
||
type ContentModerationInput struct {
|
||
Text string
|
||
Images []string
|
||
}
|
||
|
||
func (in *ContentModerationInput) Normalize() {
|
||
if in == nil {
|
||
return
|
||
}
|
||
in.Text = trimRunes(normalizeContentModerationText(in.Text), maxModerationInputRunes)
|
||
in.Images = normalizeModerationImages(in.Images)
|
||
}
|
||
|
||
func (in ContentModerationInput) IsEmpty() bool {
|
||
return strings.TrimSpace(in.Text) == "" && len(in.Images) == 0
|
||
}
|
||
|
||
func (in ContentModerationInput) ModerationInput() any {
|
||
images := limitContentModerationImages(in.Images)
|
||
if len(images) == 0 {
|
||
return in.Text
|
||
}
|
||
parts := make([]moderationAPIInputPart, 0, len(images)+1)
|
||
if strings.TrimSpace(in.Text) != "" {
|
||
parts = append(parts, moderationAPIInputPart{Type: "text", Text: in.Text})
|
||
}
|
||
for _, image := range images {
|
||
parts = append(parts, moderationAPIInputPart{
|
||
Type: "image_url",
|
||
ImageURL: &moderationAPIImageURLRef{URL: image},
|
||
})
|
||
}
|
||
return parts
|
||
}
|
||
|
||
func (in ContentModerationInput) ExcerptText() string {
|
||
return in.Text
|
||
}
|
||
|
||
func (in ContentModerationInput) Hash() string {
|
||
h := sha256.New()
|
||
_, _ = h.Write([]byte("text:"))
|
||
_, _ = h.Write([]byte(in.Text))
|
||
for _, image := range in.Images {
|
||
imageHash := sha256.Sum256([]byte(image))
|
||
_, _ = h.Write([]byte("\nimage:"))
|
||
_, _ = h.Write([]byte(hex.EncodeToString(imageHash[:])))
|
||
}
|
||
return hex.EncodeToString(h.Sum(nil))
|
||
}
|
||
|
||
type ContentModerationDecision struct {
|
||
Allowed bool `json:"allowed"`
|
||
Blocked bool `json:"blocked"`
|
||
Flagged bool `json:"flagged"`
|
||
Message string `json:"message"`
|
||
StatusCode int `json:"status_code"`
|
||
InputHash string `json:"input_hash,omitempty"`
|
||
HighestCategory string `json:"highest_category"`
|
||
HighestScore float64 `json:"highest_score"`
|
||
CategoryScores map[string]float64 `json:"category_scores"`
|
||
Action string `json:"action"`
|
||
}
|
||
|
||
type ContentModerationLog struct {
|
||
ID int64 `json:"id"`
|
||
RequestID string `json:"request_id"`
|
||
UserID *int64 `json:"user_id,omitempty"`
|
||
UserEmail string `json:"user_email"`
|
||
APIKeyID *int64 `json:"api_key_id,omitempty"`
|
||
APIKeyName string `json:"api_key_name"`
|
||
GroupID *int64 `json:"group_id,omitempty"`
|
||
GroupName string `json:"group_name"`
|
||
Endpoint string `json:"endpoint"`
|
||
Provider string `json:"provider"`
|
||
Model string `json:"model"`
|
||
Mode string `json:"mode"`
|
||
Action string `json:"action"`
|
||
Flagged bool `json:"flagged"`
|
||
HighestCategory string `json:"highest_category"`
|
||
HighestScore float64 `json:"highest_score"`
|
||
MatchedKeyword string `json:"matched_keyword"`
|
||
CategoryScores map[string]float64 `json:"category_scores"`
|
||
ThresholdSnapshot map[string]float64 `json:"threshold_snapshot"`
|
||
InputExcerpt string `json:"input_excerpt"`
|
||
UpstreamLatencyMS *int `json:"upstream_latency_ms,omitempty"`
|
||
Error string `json:"error"`
|
||
ViolationCount int `json:"violation_count"`
|
||
AutoBanned bool `json:"auto_banned"`
|
||
EmailSent bool `json:"email_sent"`
|
||
UserStatus string `json:"user_status"`
|
||
QueueDelayMS *int `json:"queue_delay_ms,omitempty"`
|
||
CreatedAt time.Time `json:"created_at"`
|
||
}
|
||
|
||
type ContentModerationLogFilter struct {
|
||
Pagination pagination.PaginationParams
|
||
Result string
|
||
GroupID *int64
|
||
Endpoint string
|
||
Search string
|
||
From *time.Time
|
||
To *time.Time
|
||
}
|
||
|
||
type ContentModerationCleanupResult struct {
|
||
DeletedHit int64 `json:"deleted_hit"`
|
||
DeletedNonHit int64 `json:"deleted_non_hit"`
|
||
FinishedAt time.Time `json:"finished_at"`
|
||
}
|
||
|
||
type ContentModerationRuntimeStatus struct {
|
||
Enabled bool `json:"enabled"`
|
||
RiskControlEnabled bool `json:"risk_control_enabled"`
|
||
Mode string `json:"mode"`
|
||
WorkerCount int `json:"worker_count"`
|
||
MaxWorkers int `json:"max_workers"`
|
||
ActiveWorkers int `json:"active_workers"`
|
||
IdleWorkers int `json:"idle_workers"`
|
||
QueueSize int `json:"queue_size"`
|
||
QueueLength int `json:"queue_length"`
|
||
QueueUsagePercent float64 `json:"queue_usage_percent"`
|
||
Enqueued int64 `json:"enqueued"`
|
||
Dropped int64 `json:"dropped"`
|
||
Processed int64 `json:"processed"`
|
||
Errors int64 `json:"errors"`
|
||
PreBlockActive int `json:"pre_block_active"`
|
||
PreBlockChecked int64 `json:"pre_block_checked"`
|
||
PreBlockAllowed int64 `json:"pre_block_allowed"`
|
||
PreBlockBlocked int64 `json:"pre_block_blocked"`
|
||
PreBlockErrors int64 `json:"pre_block_errors"`
|
||
PreBlockAvgLatencyMS int64 `json:"pre_block_avg_latency_ms"`
|
||
PreBlockAPIKeyActive int64 `json:"pre_block_api_key_active"`
|
||
PreBlockAPIKeyAvailableCount int64 `json:"pre_block_api_key_available_count"`
|
||
PreBlockAPIKeyTotalCalls int64 `json:"pre_block_api_key_total_calls"`
|
||
PreBlockAPIKeyLoads []ContentModerationAPIKeyLoad `json:"pre_block_api_key_loads"`
|
||
APIKeyStatuses []ContentModerationAPIKeyStatus `json:"api_key_statuses"`
|
||
FlaggedHashCount int64 `json:"flagged_hash_count"`
|
||
LastCleanupAt *time.Time `json:"last_cleanup_at,omitempty"`
|
||
LastCleanupDeletedHit int64 `json:"last_cleanup_deleted_hit"`
|
||
LastCleanupDeletedNonHit int64 `json:"last_cleanup_deleted_non_hit"`
|
||
}
|
||
|
||
type ContentModerationUnbanUserResult struct {
|
||
UserID int64 `json:"user_id"`
|
||
Status string `json:"status"`
|
||
}
|
||
|
||
type ContentModerationDeleteHashResult struct {
|
||
InputHash string `json:"input_hash"`
|
||
Deleted bool `json:"deleted"`
|
||
}
|
||
|
||
type ContentModerationClearHashesResult struct {
|
||
Deleted int64 `json:"deleted"`
|
||
}
|
||
|
||
type ContentModerationRepository interface {
|
||
CreateLog(ctx context.Context, log *ContentModerationLog) error
|
||
ListLogs(ctx context.Context, filter ContentModerationLogFilter) ([]ContentModerationLog, *pagination.PaginationResult, error)
|
||
// CountFlaggedByUserSince 统计窗口内计入封号的违规次数（排除 hash_block；
|
||
// excludeCyberPolicy 为 true 时额外排除 cyber_policy 行）。
|
||
CountFlaggedByUserSince(ctx context.Context, userID int64, since time.Time, excludeCyberPolicy bool) (int, error)
|
||
CleanupExpiredLogs(ctx context.Context, hitBefore time.Time, nonHitBefore time.Time) (*ContentModerationCleanupResult, error)
|
||
// UpdateLogEmailSent 回写邮件发送结果（F7：CreateLog 先行后补 EmailSent）。
|
||
UpdateLogEmailSent(ctx context.Context, id int64, sent bool) error
|
||
}
|
||
|
||
type ContentModerationHashCache interface {
|
||
RecordFlaggedInputHash(ctx context.Context, inputHash string) error
|
||
HasFlaggedInputHash(ctx context.Context, inputHash string) (bool, error)
|
||
DeleteFlaggedInputHash(ctx context.Context, inputHash string) (bool, error)
|
||
ClearFlaggedInputHashes(ctx context.Context) (int64, error)
|
||
CountFlaggedInputHashes(ctx context.Context) (int64, error)
|
||
}
|
||
|
||
type ContentModerationService struct {
|
||
settingRepo SettingRepository
|
||
repo ContentModerationRepository
|
||
hashCache ContentModerationHashCache
|
||
groupRepo GroupRepository
|
||
userRepo UserRepository
|
||
proxyRepo ProxyRepository
|
||
authCacheInvalidator APIKeyAuthCacheInvalidator
|
||
emailService *EmailService
|
||
httpClient *http.Client
|
||
moderationProxyCache atomic.Pointer[moderationProxyURLCacheEntry]
|
||
asyncQueue chan contentModerationTask
|
||
workerCount int
|
||
apiKeyCursor atomic.Uint64
|
||
asyncActive atomic.Int64
|
||
asyncEnqueued atomic.Int64
|
||
asyncDropped atomic.Int64
|
||
asyncProcessed atomic.Int64
|
||
asyncErrors atomic.Int64
|
||
preBlockActive atomic.Int64
|
||
preBlockChecked atomic.Int64
|
||
preBlockAllowed atomic.Int64
|
||
preBlockBlocked atomic.Int64
|
||
preBlockErrors atomic.Int64
|
||
preBlockLatencyTotalMS atomic.Int64
|
||
lastCleanupUnix atomic.Int64
|
||
lastCleanupDeletedHit atomic.Int64
|
||
lastCleanupDeletedNonHit atomic.Int64
|
||
runtimeSnapshot atomic.Pointer[contentModerationRuntimeSnapshot]
|
||
runtimeRefreshMu sync.Mutex
|
||
runtimeCacheTTL time.Duration
|
||
runtimeRefreshRetryAt atomic.Int64
|
||
keyHealthMu sync.Mutex
|
||
keyHealth map[string]*contentModerationKeyHealth
|
||
}
|
||
|
||
type contentModerationRuntimeSnapshot struct {
|
||
riskControlEnabled bool
|
||
config *ContentModerationConfig
|
||
keywordMatcher *contentModerationKeywordMatcher
|
||
configDigest [sha256.Size]byte
|
||
loadedAt time.Time
|
||
}
|
||
|
||
type contentModerationTask struct {
|
||
input ContentModerationCheckInput
|
||
content ContentModerationInput
|
||
inputHash string
|
||
log *ContentModerationLog
|
||
config *ContentModerationConfig
|
||
recordHash bool
|
||
applySideEffects bool
|
||
enqueuedAt time.Time
|
||
}
|
||
|
||
type contentModerationKeyHealth struct {
|
||
Hash string
|
||
Masked string
|
||
FailureCount int
|
||
SuccessCount int64
|
||
LastError string
|
||
LastCheckedAt time.Time
|
||
FrozenUntil time.Time
|
||
LastLatencyMS int
|
||
LastHTTPStatus int
|
||
LastTested bool
|
||
SyncActive int64
|
||
SyncTotal int64
|
||
SyncSuccess int64
|
||
SyncErrors int64
|
||
SyncLatencyMS int64
|
||
}
|
||
|
||
func NewContentModerationService(
|
||
settingRepo SettingRepository,
|
||
repo ContentModerationRepository,
|
||
hashCache ContentModerationHashCache,
|
||
groupRepo GroupRepository,
|
||
userRepo UserRepository,
|
||
proxyRepo ProxyRepository,
|
||
authCacheInvalidator APIKeyAuthCacheInvalidator,
|
||
emailService *EmailService,
|
||
) *ContentModerationService {
|
||
svc := &ContentModerationService{
|
||
settingRepo: settingRepo,
|
||
repo: repo,
|
||
hashCache: hashCache,
|
||
groupRepo: groupRepo,
|
||
userRepo: userRepo,
|
||
proxyRepo: proxyRepo,
|
||
authCacheInvalidator: authCacheInvalidator,
|
||
emailService: emailService,
|
||
httpClient: servertiming.InstrumentClient(nil),
|
||
workerCount: maxContentModerationWorkerCount,
|
||
asyncQueue: make(chan contentModerationTask, maxContentModerationQueueSize),
|
||
keyHealth: make(map[string]*contentModerationKeyHealth),
|
||
}
|
||
if settingRepo != nil && repo != nil {
|
||
for i := 0; i < svc.workerCount; i++ {
|
||
go svc.worker(i)
|
||
}
|
||
go svc.cleanupWorker()
|
||
}
|
||
return svc
|
||
}
|
||
|
||
func (s *ContentModerationService) GetConfig(ctx context.Context) (*ContentModerationConfigView, error) {
|
||
cfg, err := s.loadConfig(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return s.configView(cfg), nil
|
||
}
|
||
|
||
func (s *ContentModerationService) UpdateConfig(ctx context.Context, input UpdateContentModerationConfigInput) (*ContentModerationConfigView, error) {
|
||
cfg, err := s.loadConfig(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if input.Enabled != nil {
|
||
cfg.Enabled = *input.Enabled
|
||
}
|
||
if input.Mode != nil {
|
||
cfg.Mode = strings.TrimSpace(*input.Mode)
|
||
}
|
||
if input.BaseURL != nil {
|
||
cfg.BaseURL = strings.TrimSpace(*input.BaseURL)
|
||
}
|
||
if input.Model != nil {
|
||
cfg.Model = strings.TrimSpace(*input.Model)
|
||
}
|
||
if input.ProxyID != nil {
|
||
if *input.ProxyID > 0 {
|
||
id := *input.ProxyID
|
||
cfg.ProxyID = &id
|
||
} else {
|
||
cfg.ProxyID = nil
|
||
}
|
||
}
|
||
if input.TimeoutMS != nil {
|
||
cfg.TimeoutMS = *input.TimeoutMS
|
||
}
|
||
if input.SampleRate != nil {
|
||
cfg.SampleRate = *input.SampleRate
|
||
}
|
||
if input.WorkerCount != nil {
|
||
cfg.WorkerCount = *input.WorkerCount
|
||
}
|
||
if input.QueueSize != nil {
|
||
cfg.QueueSize = *input.QueueSize
|
||
}
|
||
if input.BlockStatus != nil {
|
||
cfg.BlockStatus = *input.BlockStatus
|
||
}
|
||
if input.BlockMessage != nil {
|
||
cfg.BlockMessage = strings.TrimSpace(*input.BlockMessage)
|
||
}
|
||
if input.EmailOnHit != nil {
|
||
cfg.EmailOnHit = *input.EmailOnHit
|
||
}
|
||
if input.AutoBanEnabled != nil {
|
||
cfg.AutoBanEnabled = *input.AutoBanEnabled
|
||
}
|
||
if input.BanThreshold != nil {
|
||
cfg.BanThreshold = *input.BanThreshold
|
||
}
|
||
if input.ViolationWindowHours != nil {
|
||
cfg.ViolationWindowHours = *input.ViolationWindowHours
|
||
}
|
||
if input.RetryCount != nil {
|
||
cfg.RetryCount = *input.RetryCount
|
||
}
|
||
if input.HitRetentionDays != nil {
|
||
cfg.HitRetentionDays = *input.HitRetentionDays
|
||
}
|
||
if input.NonHitRetentionDays != nil {
|
||
cfg.NonHitRetentionDays = *input.NonHitRetentionDays
|
||
}
|
||
if input.PreHashCheckEnabled != nil {
|
||
cfg.PreHashCheckEnabled = *input.PreHashCheckEnabled
|
||
}
|
||
if input.BlockedKeywords != nil {
|
||
cfg.BlockedKeywords = normalizeBlockedKeywords(*input.BlockedKeywords)
|
||
}
|
||
if input.KeywordBlockingMode != nil {
|
||
cfg.KeywordBlockingMode = strings.TrimSpace(*input.KeywordBlockingMode)
|
||
}
|
||
if input.ModelFilter != nil {
|
||
cfg.ModelFilter = *input.ModelFilter
|
||
}
|
||
if input.AllGroups != nil {
|
||
cfg.AllGroups = *input.AllGroups
|
||
}
|
||
if input.GroupIDs != nil {
|
||
cfg.GroupIDs = normalizeInt64IDs(*input.GroupIDs)
|
||
}
|
||
if input.RecordNonHits != nil {
|
||
cfg.RecordNonHits = *input.RecordNonHits
|
||
}
|
||
if input.CyberPolicyExcludeFromBanCount != nil {
|
||
cfg.CyberPolicyExcludeFromBanCount = *input.CyberPolicyExcludeFromBanCount
|
||
}
|
||
if input.Thresholds != nil {
|
||
cfg.Thresholds = mergeContentModerationThresholds(ContentModerationDefaultThresholds(), *input.Thresholds)
|
||
}
|
||
if input.ClearAPIKey {
|
||
cfg.APIKey = ""
|
||
cfg.APIKeys = []string{}
|
||
} else {
|
||
apiKeysMode := normalizeContentModerationAPIKeysMode(input.APIKeysMode)
|
||
if input.DeleteAPIKeyHashes != nil && apiKeysMode != contentModerationAPIKeysModeReplace {
|
||
cfg.APIKeys = deleteModerationAPIKeysByHash(cfg.apiKeys(), *input.DeleteAPIKeyHashes)
|
||
cfg.APIKey = ""
|
||
}
|
||
if input.APIKeys != nil {
|
||
if apiKeysMode == contentModerationAPIKeysModeReplace {
|
||
cfg.APIKeys = normalizeModerationAPIKeys(*input.APIKeys)
|
||
} else {
|
||
cfg.APIKeys = normalizeModerationAPIKeys(append(cfg.apiKeys(), *input.APIKeys...))
|
||
}
|
||
cfg.APIKey = ""
|
||
}
|
||
if input.APIKey != nil && strings.TrimSpace(*input.APIKey) != "" {
|
||
cfg.APIKeys = normalizeModerationAPIKeys(append(cfg.APIKeys, *input.APIKey))
|
||
cfg.APIKey = ""
|
||
}
|
||
}
|
||
if err := s.validateConfig(ctx, cfg); err != nil {
|
||
return nil, err
|
||
}
|
||
cfg.normalize()
|
||
raw, err := json.Marshal(cfg)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("marshal content moderation config: %w", err)
|
||
}
|
||
if err := s.settingRepo.Set(ctx, SettingKeyContentModerationConfig, string(raw)); err != nil {
|
||
return nil, fmt.Errorf("save content moderation config: %w", err)
|
||
}
|
||
s.replaceRuntimeConfig(cfg, raw)
|
||
// 代理选择可能已变化，丢弃已解析的代理 URL 缓存，下次调用即时生效。
|
||
s.moderationProxyCache.Store(nil)
|
||
return s.configView(cfg), nil
|
||
}
|
||
|
||
func (s *ContentModerationService) TestAPIKeys(ctx context.Context, input TestContentModerationAPIKeysInput) (*TestContentModerationAPIKeysResult, error) {
|
||
cfg, err := s.loadConfig(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
keys := normalizeModerationAPIKeys(input.APIKeys)
|
||
configured := false
|
||
if len(keys) == 0 {
|
||
keys = cfg.apiKeys()
|
||
configured = true
|
||
}
|
||
if strings.TrimSpace(input.BaseURL) != "" {
|
||
cfg.BaseURL = input.BaseURL
|
||
}
|
||
if strings.TrimSpace(input.Model) != "" {
|
||
cfg.Model = input.Model
|
||
}
|
||
if input.TimeoutMS > 0 {
|
||
cfg.TimeoutMS = input.TimeoutMS
|
||
}
|
||
if input.ProxyID != nil {
|
||
if *input.ProxyID > 0 {
|
||
id := *input.ProxyID
|
||
cfg.ProxyID = &id
|
||
} else {
|
||
cfg.ProxyID = nil
|
||
}
|
||
}
|
||
cfg.normalize()
|
||
testInput, imageCount, err := buildModerationTestInput(input.Prompt, input.Images)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
auditOnly := contentModerationTestHasAuditInput(input.Prompt, input.Images)
|
||
if configured && auditOnly {
|
||
key, ok := s.nextUsableAPIKey(cfg)
|
||
if !ok {
|
||
return &TestContentModerationAPIKeysResult{
|
||
Items: s.apiKeyStatuses(keys),
|
||
ImageCount: imageCount,
|
||
}, nil
|
||
}
|
||
keys = []string{key}
|
||
}
|
||
if len(keys) == 0 {
|
||
return &TestContentModerationAPIKeysResult{Items: []ContentModerationAPIKeyStatus{}, ImageCount: imageCount}, nil
|
||
}
|
||
items := make([]ContentModerationAPIKeyStatus, 0, len(keys))
|
||
var auditResult *ContentModerationTestAuditResult
|
||
for idx, key := range keys {
|
||
start := time.Now()
|
||
httpStatus := 0
|
||
result, err := s.callModerationOnceWithInput(ctx, cfg, key, testInput, &httpStatus)
|
||
latency := int(time.Since(start).Milliseconds())
|
||
keyHash := moderationAPIKeyHash(key)
|
||
if err != nil {
|
||
s.markAPIKeyError(key, err.Error(), latency, httpStatus)
|
||
} else {
|
||
s.markAPIKeySuccess(key, latency, httpStatus)
|
||
if auditResult == nil {
|
||
auditResult = buildContentModerationTestAuditResult(result, cfg.Thresholds)
|
||
}
|
||
}
|
||
status := s.apiKeyStatusForHash(idx, keyHash, maskSecretTail(key), configured)
|
||
status.LastTested = true
|
||
items = append(items, status)
|
||
}
|
||
return &TestContentModerationAPIKeysResult{Items: items, AuditResult: auditResult, ImageCount: imageCount}, nil
|
||
}
|
||
|
||
func (s *ContentModerationService) Check(ctx context.Context, input ContentModerationCheckInput) (*ContentModerationDecision, error) {
|
||
allow := &ContentModerationDecision{Allowed: true, Action: ContentModerationActionAllow}
|
||
if s == nil || s.settingRepo == nil || s.repo == nil {
|
||
slog.Info("content_moderation.skip_unavailable",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol)
|
||
return allow, nil
|
||
}
|
||
runtimeSnapshot, err := s.loadRuntimeSnapshot(ctx)
|
||
if err != nil {
|
||
slog.Warn("content_moderation.skip_config_load_failed",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol,
|
||
"error", err)
|
||
return allow, nil
|
||
}
|
||
if !runtimeSnapshot.riskControlEnabled {
|
||
slog.Info("content_moderation.skip_feature_disabled",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol)
|
||
return allow, nil
|
||
}
|
||
cfg := runtimeSnapshot.config
|
||
inGroupScope := cfg.includesGroup(input.GroupID)
|
||
inModelScope := cfg.includesModel(input.Model)
|
||
slog.Info("content_moderation.config_loaded",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"group_name", input.GroupName,
|
||
"endpoint", input.Endpoint,
|
||
"provider", input.Provider,
|
||
"protocol", input.Protocol,
|
||
"model", input.Model,
|
||
"enabled", cfg.Enabled,
|
||
"mode", cfg.Mode,
|
||
"all_groups", cfg.AllGroups,
|
||
"configured_group_ids", cfg.GroupIDs,
|
||
"in_group_scope", inGroupScope,
|
||
"model_filter_type", cfg.ModelFilter.Type,
|
||
"configured_models", cfg.ModelFilter.Models,
|
||
"in_model_scope", inModelScope,
|
||
"sample_rate", cfg.SampleRate,
|
||
"api_key_count", len(cfg.apiKeys()),
|
||
"pre_hash_check_enabled", cfg.PreHashCheckEnabled,
|
||
"record_non_hits", cfg.RecordNonHits)
|
||
if !cfg.Enabled {
|
||
slog.Info("content_moderation.skip_config_disabled",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol)
|
||
return allow, nil
|
||
}
|
||
if cfg.Mode == ContentModerationModeOff {
|
||
slog.Info("content_moderation.skip_mode_off",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol)
|
||
return allow, nil
|
||
}
|
||
if !inGroupScope {
|
||
slog.Info("content_moderation.skip_group_out_of_scope",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"group_name", input.GroupName,
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol,
|
||
"all_groups", cfg.AllGroups,
|
||
"configured_group_ids", cfg.GroupIDs)
|
||
return allow, nil
|
||
}
|
||
if !inModelScope {
|
||
slog.Info("content_moderation.skip_model_out_of_scope",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"group_name", input.GroupName,
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol,
|
||
"model", input.Model,
|
||
"model_filter_type", cfg.ModelFilter.Type,
|
||
"configured_models", cfg.ModelFilter.Models)
|
||
return allow, nil
|
||
}
|
||
content := ExtractContentModerationInput(input.Protocol, input.Body)
|
||
if content.IsEmpty() {
|
||
slog.Info("content_moderation.skip_empty_input",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol,
|
||
"body_bytes", len(input.Body))
|
||
return allow, nil
|
||
}
|
||
content.Normalize()
|
||
slog.Info("content_moderation.input_extracted",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol,
|
||
"text_runes", len([]rune(content.Text)),
|
||
"image_count", len(content.Images))
|
||
hashText := content.Hash()
|
||
if cfg.Mode == ContentModerationModePreBlock {
|
||
if cfg.KeywordBlockingMode != ContentModerationKeywordModeAPIOnly && len(cfg.BlockedKeywords) > 0 {
|
||
if keyword, hit := runtimeSnapshot.matchBlockedKeyword(content.Text); hit {
|
||
s.recordPreBlockSyncMetric(0, ContentModerationActionKeywordBlock)
|
||
slog.Info("content_moderation.keyword_block",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol,
|
||
"keyword_blocking_mode", cfg.KeywordBlockingMode,
|
||
"keyword", keyword)
|
||
scores := map[string]float64{contentModerationKeywordCategory: 1.0}
|
||
log := s.buildLog(input, cfg, ContentModerationActionKeywordBlock, true, contentModerationKeywordCategory, 1.0, scores, content.ExcerptText(), nil, nil, "")
|
||
log.MatchedKeyword = keyword
|
||
s.enqueueRecord(input, cfg, log, hashText, false, true)
|
||
return &ContentModerationDecision{
|
||
Allowed: false,
|
||
Blocked: true,
|
||
Flagged: true,
|
||
Message: cfg.BlockMessage,
|
||
StatusCode: cfg.BlockStatus,
|
||
HighestCategory: contentModerationKeywordCategory,
|
||
HighestScore: 1.0,
|
||
CategoryScores: scores,
|
||
Action: ContentModerationActionKeywordBlock,
|
||
}, nil
|
||
}
|
||
}
|
||
if cfg.KeywordBlockingMode == ContentModerationKeywordModeKeywordOnly {
|
||
s.recordPreBlockSyncMetric(0, ContentModerationActionAllow)
|
||
slog.Info("content_moderation.skip_api_keyword_only",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol)
|
||
return allow, nil
|
||
}
|
||
}
|
||
if cfg.PreHashCheckEnabled && s.hashCache != nil {
|
||
matched, err := s.hashCache.HasFlaggedInputHash(ctx, hashText)
|
||
if err != nil {
|
||
slog.Warn("content_moderation.hash_check_failed", "user_id", input.UserID, "endpoint", input.Endpoint, "error", err)
|
||
}
|
||
if matched {
|
||
if cfg.Mode == ContentModerationModePreBlock {
|
||
s.recordPreBlockSyncMetric(0, ContentModerationActionHashBlock)
|
||
}
|
||
slog.Info("content_moderation.hash_block",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol,
|
||
"input_hash", hashText)
|
||
message := cfg.BlockMessage
|
||
if message != "" {
|
||
message = fmt.Sprintf("%s（hash: %s）", message, hashText)
|
||
}
|
||
scores := map[string]float64{"hash": 1.0}
|
||
log := s.buildLog(input, cfg, ContentModerationActionHashBlock, true, "hash", 1.0, scores, content.ExcerptText(), nil, nil, "")
|
||
s.enqueueRecord(input, cfg, log, hashText, false, false)
|
||
return &ContentModerationDecision{
|
||
Allowed: false,
|
||
Blocked: true,
|
||
Flagged: true,
|
||
Message: message,
|
||
StatusCode: cfg.BlockStatus,
|
||
InputHash: hashText,
|
||
Action: ContentModerationActionHashBlock,
|
||
}, nil
|
||
}
|
||
}
|
||
if !cfg.shouldSample(hashText) {
|
||
if cfg.Mode == ContentModerationModePreBlock {
|
||
s.recordPreBlockSyncMetric(0, ContentModerationActionAllow)
|
||
}
|
||
slog.Info("content_moderation.skip_sample_rate",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol,
|
||
"sample_rate", cfg.SampleRate)
|
||
return allow, nil
|
||
}
|
||
if len(cfg.apiKeys()) == 0 {
|
||
if cfg.Mode == ContentModerationModePreBlock {
|
||
s.recordPreBlockSyncMetric(0, ContentModerationActionError)
|
||
}
|
||
slog.Warn("content_moderation.skip_no_audit_api_keys",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol)
|
||
return allow, nil
|
||
}
|
||
if cfg.Mode == ContentModerationModeObserve {
|
||
slog.Info("content_moderation.enqueue_observe",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol,
|
||
"queue_len", len(s.asyncQueue))
|
||
s.enqueueAsync(input, cfg, content, hashText)
|
||
return allow, nil
|
||
}
|
||
|
||
return s.checkSync(ctx, input, cfg, content, hashText, nil, true), nil
|
||
}
|
||
|
||
func (s *ContentModerationService) checkSync(ctx context.Context, input ContentModerationCheckInput, cfg *ContentModerationConfig, content ContentModerationInput, hashText string, queueDelay *int, allowBlock bool) *ContentModerationDecision {
|
||
allow := &ContentModerationDecision{Allowed: true, Action: ContentModerationActionAllow}
|
||
trackPreBlock := queueDelay == nil && allowBlock && cfg != nil && cfg.Mode == ContentModerationModePreBlock
|
||
if trackPreBlock {
|
||
s.preBlockActive.Add(1)
|
||
defer s.preBlockActive.Add(-1)
|
||
}
|
||
start := time.Now()
|
||
result, err := s.callModeration(ctx, cfg, content.ModerationInput(), trackPreBlock)
|
||
latency := int(time.Since(start).Milliseconds())
|
||
if err != nil {
|
||
if trackPreBlock {
|
||
s.recordPreBlockSyncMetric(latency, ContentModerationActionError)
|
||
}
|
||
slog.Warn("content_moderation.audit_api_failed",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol,
|
||
"mode", cfg.Mode,
|
||
"allow_block", allowBlock,
|
||
"queue_delay_ms", queueDelay,
|
||
"latency_ms", latency,
|
||
"error", err)
|
||
if queueDelay != nil {
|
||
s.asyncErrors.Add(1)
|
||
}
|
||
if cfg.RecordNonHits {
|
||
log := s.buildLog(input, cfg, ContentModerationActionError, false, "", 0, nil, content.ExcerptText(), &latency, queueDelay, err.Error())
|
||
_ = s.repo.CreateLog(ctx, log)
|
||
}
|
||
return allow
|
||
}
|
||
|
||
flagged, highestCategory, highestScore := evaluateModerationScores(result.CategoryScores, cfg.Thresholds)
|
||
action := ContentModerationActionAllow
|
||
blocked := false
|
||
if allowBlock && flagged && cfg.Mode == ContentModerationModePreBlock {
|
||
action = ContentModerationActionBlock
|
||
blocked = true
|
||
}
|
||
if trackPreBlock {
|
||
s.recordPreBlockSyncMetric(latency, action)
|
||
}
|
||
slog.Info("content_moderation.audit_result",
|
||
"user_id", input.UserID,
|
||
"api_key_id", input.APIKeyID,
|
||
"group_id", contentModerationLogGroupID(input.GroupID),
|
||
"group_name", input.GroupName,
|
||
"endpoint", input.Endpoint,
|
||
"protocol", input.Protocol,
|
||
"mode", cfg.Mode,
|
||
"allow_block", allowBlock,
|
||
"flagged", flagged,
|
||
"blocked", blocked,
|
||
"action", action,
|
||
"highest_category", highestCategory,
|
||
"highest_score", highestScore,
|
||
"latency_ms", latency,
|
||
"queue_delay_ms", queueDelay)
|
||
if flagged || cfg.RecordNonHits {
|
||
log := s.buildLog(input, cfg, action, flagged, highestCategory, highestScore, result.CategoryScores, content.ExcerptText(), &latency, queueDelay, "")
|
||
if queueDelay == nil && cfg.Mode == ContentModerationModePreBlock {
|
||
s.enqueueRecord(input, cfg, log, hashText, flagged, flagged)
|
||
} else {
|
||
s.persistContentModerationLog(ctx, cfg, log, hashText, flagged, flagged)
|
||
}
|
||
}
|
||
if blocked {
|
||
return &ContentModerationDecision{
|
||
Allowed: false,
|
||
Blocked: true,
|
||
Flagged: true,
|
||
Message: cfg.BlockMessage,
|
||
StatusCode: cfg.BlockStatus,
|
||
HighestCategory: highestCategory,
|
||
HighestScore: highestScore,
|
||
CategoryScores: result.CategoryScores,
|
||
Action: action,
|
||
}
|
||
}
|
||
return &ContentModerationDecision{
|
||
Allowed: true,
|
||
Flagged: flagged,
|
||
Message: "",
|
||
HighestCategory: highestCategory,
|
||
HighestScore: highestScore,
|
||
CategoryScores: result.CategoryScores,
|
||
Action: action,
|
||
}
|
||
}
|
||
|
||
func (s *ContentModerationService) recordPreBlockSyncMetric(latencyMS int, action string) {
|
||
if s == nil {
|
||
return
|
||
}
|
||
s.preBlockChecked.Add(1)
|
||
if latencyMS < 0 {
|
||
latencyMS = 0
|
||
}
|
||
s.preBlockLatencyTotalMS.Add(int64(latencyMS))
|
||
switch action {
|
||
case ContentModerationActionBlock, ContentModerationActionHashBlock, ContentModerationActionKeywordBlock:
|
||
s.preBlockBlocked.Add(1)
|
||
case ContentModerationActionError:
|
||
s.preBlockErrors.Add(1)
|
||
default:
|
||
s.preBlockAllowed.Add(1)
|
||
}
|
||
}
|
||
|
||
func (s *ContentModerationService) enqueueAsync(input ContentModerationCheckInput, cfg *ContentModerationConfig, content ContentModerationInput, hashText string) {
|
||
if s == nil || s.asyncQueue == nil {
|
||
return
|
||
}
|
||
queueSize := defaultContentModerationQueueSize
|
||
if cfg != nil && cfg.QueueSize > 0 {
|
||
queueSize = cfg.QueueSize
|
||
}
|
||
if len(s.asyncQueue) >= queueSize {
|
||
slog.Warn("content_moderation.async_queue_full", "user_id", input.UserID, "endpoint", input.Endpoint, "queue_size", queueSize)
|
||
s.asyncDropped.Add(1)
|
||
return
|
||
}
|
||
task := contentModerationTask{
|
||
input: input,
|
||
content: content,
|
||
inputHash: hashText,
|
||
enqueuedAt: time.Now(),
|
||
}
|
||
select {
|
||
case s.asyncQueue <- task:
|
||
s.asyncEnqueued.Add(1)
|
||
default:
|
||
slog.Warn("content_moderation.async_queue_full", "user_id", input.UserID, "endpoint", input.Endpoint)
|
||
s.asyncDropped.Add(1)
|
||
}
|
||
}
|
||
|
||
func (s *ContentModerationService) enqueueRecord(input ContentModerationCheckInput, cfg *ContentModerationConfig, log *ContentModerationLog, inputHash string, recordHash bool, applySideEffects bool) {
|
||
if s == nil || s.asyncQueue == nil || log == nil {
|
||
return
|
||
}
|
||
queueSize := defaultContentModerationQueueSize
|
||
if cfg != nil && cfg.QueueSize > 0 {
|
||
queueSize = cfg.QueueSize
|
||
}
|
||
if len(s.asyncQueue) >= queueSize {
|
||
slog.Warn("content_moderation.record_queue_full",
|
||
"user_id", input.UserID,
|
||
"endpoint", input.Endpoint,
|
||
"action", log.Action,
|
||
"queue_size", queueSize)
|
||
s.asyncDropped.Add(1)
|
||
return
|
||
}
|
||
task := contentModerationTask{
|
||
input: input,
|
||
inputHash: inputHash,
|
||
log: log,
|
||
config: cloneContentModerationConfig(cfg),
|
||
recordHash: recordHash,
|
||
applySideEffects: applySideEffects,
|
||
enqueuedAt: time.Now(),
|
||
}
|
||
select {
|
||
case s.asyncQueue <- task:
|
||
s.asyncEnqueued.Add(1)
|
||
default:
|
||
slog.Warn("content_moderation.record_queue_full",
|
||
"user_id", input.UserID,
|
||
"endpoint", input.Endpoint,
|
||
"action", log.Action)
|
||
s.asyncDropped.Add(1)
|
||
}
|
||
}
|
||
|
||
func (s *ContentModerationService) worker(id int) {
|
||
for {
|
||
ctx, cancel := context.WithTimeout(context.Background(), maxContentModerationTimeoutMS*time.Millisecond+10*time.Second)
|
||
runtimeSnapshot, err := s.loadRuntimeSnapshot(ctx)
|
||
if err != nil || runtimeSnapshot == nil || runtimeSnapshot.config == nil || id >= runtimeSnapshot.config.WorkerCount {
|
||
cancel()
|
||
time.Sleep(time.Second)
|
||
continue
|
||
}
|
||
cfg := runtimeSnapshot.config
|
||
task, ok := s.dequeueAsyncTask(ctx, time.Second)
|
||
if !ok {
|
||
cancel()
|
||
continue
|
||
}
|
||
func() {
|
||
defer cancel()
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
slog.Error("content_moderation.worker_panic", "worker_id", id, "recover", r)
|
||
}
|
||
}()
|
||
if task.log != nil {
|
||
s.asyncActive.Add(1)
|
||
defer s.asyncActive.Add(-1)
|
||
queueDelay := int(time.Since(task.enqueuedAt).Milliseconds())
|
||
task.log.QueueDelayMS = &queueDelay
|
||
taskCfg := task.config
|
||
if taskCfg == nil {
|
||
taskCfg = cfg
|
||
}
|
||
s.persistContentModerationLog(ctx, taskCfg, task.log, task.inputHash, task.recordHash, task.applySideEffects)
|
||
s.asyncProcessed.Add(1)
|
||
return
|
||
}
|
||
if !cfg.Enabled || cfg.Mode == ContentModerationModeOff || len(cfg.apiKeys()) == 0 {
|
||
return
|
||
}
|
||
if !cfg.includesGroup(task.input.GroupID) {
|
||
return
|
||
}
|
||
if !cfg.includesModel(task.input.Model) {
|
||
return
|
||
}
|
||
s.asyncActive.Add(1)
|
||
defer s.asyncActive.Add(-1)
|
||
queueDelay := int(time.Since(task.enqueuedAt).Milliseconds())
|
||
_ = s.checkSync(ctx, task.input, cfg, task.content, task.inputHash, &queueDelay, false)
|
||
s.asyncProcessed.Add(1)
|
||
}()
|
||
}
|
||
}
|
||
|
||
func (s *ContentModerationService) dequeueAsyncTask(ctx context.Context, idleWait time.Duration) (contentModerationTask, bool) {
|
||
var zero contentModerationTask
|
||
if s == nil || s.asyncQueue == nil {
|
||
return zero, false
|
||
}
|
||
if idleWait <= 0 {
|
||
idleWait = time.Second
|
||
}
|
||
timer := time.NewTimer(idleWait)
|
||
defer timer.Stop()
|
||
select {
|
||
case task, ok := <-s.asyncQueue:
|
||
return task, ok
|
||
case <-ctx.Done():
|
||
return zero, false
|
||
case <-timer.C:
|
||
return zero, false
|
||
}
|
||
}
|
||
|
||
func (s *ContentModerationService) ListLogs(ctx context.Context, filter ContentModerationLogFilter) ([]ContentModerationLog, *pagination.PaginationResult, error) {
|
||
if filter.Pagination.Page <= 0 {
|
||
filter.Pagination.Page = 1
|
||
}
|
||
if filter.Pagination.PageSize <= 0 {
|
||
filter.Pagination.PageSize = 20
|
||
}
|
||
if filter.Pagination.PageSize > 100 {
|
||
filter.Pagination.PageSize = 100
|
||
}
|
||
if filter.Pagination.SortOrder == "" {
|
||
filter.Pagination.SortOrder = pagination.SortOrderDesc
|
||
}
|
||
return s.repo.ListLogs(ctx, filter)
|
||
}
|
||
|
||
func (s *ContentModerationService) UnbanUser(ctx context.Context, userID int64) (*ContentModerationUnbanUserResult, error) {
|
||
if s == nil || s.userRepo == nil {
|
||
return nil, infraerrors.InternalServer("CONTENT_MODERATION_USER_REPOSITORY_UNAVAILABLE", "用户仓储不可用")
|
||
}
|
||
if userID <= 0 {
|
||
return nil, infraerrors.BadRequest("INVALID_USER_ID", "用户 ID 无效")
|
||
}
|
||
user, err := s.userRepo.GetByID(ctx, userID)
|
||
if err != nil {
|
||
if errors.Is(err, ErrUserNotFound) {
|
||
return nil, infraerrors.NotFound("USER_NOT_FOUND", "用户不存在")
|
||
}
|
||
return nil, fmt.Errorf("get content moderation unban user: %w", err)
|
||
}
|
||
if user.Status != StatusActive {
|
||
user.Status = StatusActive
|
||
if err := s.userRepo.Update(ctx, user, UserUpdateFields{Status: true}); err != nil {
|
||
return nil, fmt.Errorf("update content moderation unban user: %w", err)
|
||
}
|
||
}
|
||
if s.authCacheInvalidator != nil {
|
||
s.authCacheInvalidator.InvalidateAuthCacheByUserID(ctx, userID)
|
||
}
|
||
return &ContentModerationUnbanUserResult{
|
||
UserID: userID,
|
||
Status: StatusActive,
|
||
}, nil
|
||
}
|
||
|
||
func (s *ContentModerationService) DeleteFlaggedInputHash(ctx context.Context, inputHash string) (*ContentModerationDeleteHashResult, error) {
|
||
inputHash = normalizeContentModerationHash(inputHash)
|
||
if inputHash == "" {
|
||
return nil, infraerrors.BadRequest("INVALID_CONTENT_MODERATION_HASH", "风险输入哈希无效")
|
||
}
|
||
if s == nil || s.hashCache == nil {
|
||
return nil, infraerrors.InternalServer("CONTENT_MODERATION_HASH_CACHE_UNAVAILABLE", "内容审计哈希缓存不可用")
|
||
}
|
||
deleted, err := s.hashCache.DeleteFlaggedInputHash(ctx, inputHash)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("delete content moderation flagged hash: %w", err)
|
||
}
|
||
return &ContentModerationDeleteHashResult{
|
||
InputHash: inputHash,
|
||
Deleted: deleted,
|
||
}, nil
|
||
}
|
||
|
||
func (s *ContentModerationService) ClearFlaggedInputHashes(ctx context.Context) (*ContentModerationClearHashesResult, error) {
|
||
if s == nil || s.hashCache == nil {
|
||
return nil, infraerrors.InternalServer("CONTENT_MODERATION_HASH_CACHE_UNAVAILABLE", "内容审计哈希缓存不可用")
|
||
}
|
||
deleted, err := s.hashCache.ClearFlaggedInputHashes(ctx)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("clear content moderation flagged hashes: %w", err)
|
||
}
|
||
return &ContentModerationClearHashesResult{Deleted: deleted}, nil
|
||
}
|
||
|
||
func (s *ContentModerationService) GetStatus(ctx context.Context) (*ContentModerationRuntimeStatus, error) {
|
||
if s == nil {
|
||
return &ContentModerationRuntimeStatus{}, nil
|
||
}
|
||
cfg, err := s.loadConfig(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
riskEnabled := s.isRiskControlEnabled(ctx)
|
||
active := int(s.asyncActive.Load())
|
||
if active < 0 {
|
||
active = 0
|
||
}
|
||
if active > cfg.WorkerCount {
|
||
active = cfg.WorkerCount
|
||
}
|
||
preBlockActive := int(s.preBlockActive.Load())
|
||
if preBlockActive < 0 {
|
||
preBlockActive = 0
|
||
}
|
||
preBlockChecked := s.preBlockChecked.Load()
|
||
preBlockAvgLatency := int64(0)
|
||
if preBlockChecked > 0 {
|
||
preBlockAvgLatency = s.preBlockLatencyTotalMS.Load() / preBlockChecked
|
||
}
|
||
queueLength := 0
|
||
if s.asyncQueue != nil {
|
||
queueLength = len(s.asyncQueue)
|
||
}
|
||
queueUsage := 0.0
|
||
if cfg.QueueSize > 0 {
|
||
queueUsage = float64(queueLength) * 100 / float64(cfg.QueueSize)
|
||
}
|
||
var flaggedHashCount int64
|
||
if s.hashCache != nil {
|
||
if n, err := s.hashCache.CountFlaggedInputHashes(ctx); err == nil {
|
||
flaggedHashCount = n
|
||
} else {
|
||
slog.Warn("content_moderation.hash_count_failed", "error", err)
|
||
}
|
||
}
|
||
var lastCleanupAt *time.Time
|
||
if unix := s.lastCleanupUnix.Load(); unix > 0 {
|
||
t := time.Unix(unix, 0)
|
||
lastCleanupAt = &t
|
||
}
|
||
return &ContentModerationRuntimeStatus{
|
||
Enabled: cfg.Enabled,
|
||
RiskControlEnabled: riskEnabled,
|
||
Mode: cfg.Mode,
|
||
WorkerCount: cfg.WorkerCount,
|
||
MaxWorkers: maxContentModerationWorkerCount,
|
||
ActiveWorkers: active,
|
||
IdleWorkers: cfg.WorkerCount - active,
|
||
QueueSize: cfg.QueueSize,
|
||
QueueLength: queueLength,
|
||
QueueUsagePercent: queueUsage,
|
||
Enqueued: s.asyncEnqueued.Load(),
|
||
Dropped: s.asyncDropped.Load(),
|
||
Processed: s.asyncProcessed.Load(),
|
||
Errors: s.asyncErrors.Load(),
|
||
PreBlockActive: preBlockActive,
|
||
PreBlockChecked: preBlockChecked,
|
||
PreBlockAllowed: s.preBlockAllowed.Load(),
|
||
PreBlockBlocked: s.preBlockBlocked.Load(),
|
||
PreBlockErrors: s.preBlockErrors.Load(),
|
||
PreBlockAvgLatencyMS: preBlockAvgLatency,
|
||
PreBlockAPIKeyActive: s.preBlockAPIKeyActive(cfg.apiKeys()),
|
||
PreBlockAPIKeyAvailableCount: s.preBlockAPIKeyAvailableCount(cfg.apiKeys()),
|
||
PreBlockAPIKeyTotalCalls: s.preBlockAPIKeyTotalCalls(cfg.apiKeys()),
|
||
PreBlockAPIKeyLoads: s.preBlockAPIKeyLoads(cfg.apiKeys()),
|
||
APIKeyStatuses: s.apiKeyStatuses(cfg.apiKeys()),
|
||
FlaggedHashCount: flaggedHashCount,
|
||
LastCleanupAt: lastCleanupAt,
|
||
LastCleanupDeletedHit: s.lastCleanupDeletedHit.Load(),
|
||
LastCleanupDeletedNonHit: s.lastCleanupDeletedNonHit.Load(),
|
||
}, nil
|
||
}
|
||
|
||
func (s *ContentModerationService) cleanupWorker() {
|
||
timer := time.NewTimer(contentModerationCleanupDelay)
|
||
defer timer.Stop()
|
||
for {
|
||
<-timer.C
|
||
s.runCleanupOnce()
|
||
timer.Reset(contentModerationCleanupInterval)
|
||
}
|
||
}
|
||
|
||
func (s *ContentModerationService) runCleanupOnce() {
|
||
if s == nil || s.repo == nil || s.settingRepo == nil {
|
||
return
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), contentModerationCleanupTimeout)
|
||
defer cancel()
|
||
cfg, err := s.loadConfig(ctx)
|
||
if err != nil {
|
||
slog.Warn("content_moderation.cleanup_load_config_failed", "error", err)
|
||
return
|
||
}
|
||
now := time.Now()
|
||
hitBefore := now.AddDate(0, 0, -cfg.HitRetentionDays)
|
||
nonHitBefore := now.AddDate(0, 0, -cfg.NonHitRetentionDays)
|
||
result, err := s.repo.CleanupExpiredLogs(ctx, hitBefore, nonHitBefore)
|
||
if err != nil {
|
||
slog.Warn("content_moderation.cleanup_failed", "error", err)
|
||
return
|
||
}
|
||
if result == nil {
|
||
return
|
||
}
|
||
s.lastCleanupUnix.Store(result.FinishedAt.Unix())
|
||
s.lastCleanupDeletedHit.Store(result.DeletedHit)
|
||
s.lastCleanupDeletedNonHit.Store(result.DeletedNonHit)
|
||
}
|
||
|
||
func (s *ContentModerationService) loadConfig(ctx context.Context) (*ContentModerationConfig, error) {
|
||
raw, err := s.settingRepo.GetValue(ctx, SettingKeyContentModerationConfig)
|
||
if err != nil {
|
||
if errors.Is(err, ErrSettingNotFound) {
|
||
return parseContentModerationConfig("")
|
||
}
|
||
return nil, fmt.Errorf("get content moderation config: %w", err)
|
||
}
|
||
return parseContentModerationConfig(raw)
|
||
}
|
||
|
||
func parseContentModerationConfig(raw string) (*ContentModerationConfig, error) {
|
||
cfg := defaultContentModerationConfig()
|
||
if strings.TrimSpace(raw) == "" {
|
||
cfg.normalize()
|
||
return cfg, nil
|
||
}
|
||
if err := json.Unmarshal([]byte(raw), cfg); err != nil {
|
||
return nil, infraerrors.BadRequest("INVALID_CONTENT_MODERATION_CONFIG", "内容审计配置不是有效 JSON")
|
||
}
|
||
cfg.normalize()
|
||
return cfg, nil
|
||
}
|
||
|
||
func (s *ContentModerationService) loadRuntimeSnapshot(ctx context.Context) (*contentModerationRuntimeSnapshot, error) {
|
||
if s == nil || s.settingRepo == nil {
|
||
return nil, errors.New("content moderation setting repository unavailable")
|
||
}
|
||
now := time.Now()
|
||
if snapshot := s.runtimeSnapshot.Load(); snapshot != nil {
|
||
if now.Sub(snapshot.loadedAt) < s.runtimeSnapshotTTL() {
|
||
return snapshot, nil
|
||
}
|
||
s.triggerRuntimeSnapshotRefresh()
|
||
return snapshot, nil
|
||
}
|
||
|
||
s.runtimeRefreshMu.Lock()
|
||
defer s.runtimeRefreshMu.Unlock()
|
||
if snapshot := s.runtimeSnapshot.Load(); snapshot != nil {
|
||
return snapshot, nil
|
||
}
|
||
return s.refreshRuntimeSnapshot(ctx)
|
||
}
|
||
|
||
func (s *ContentModerationService) runtimeSnapshotTTL() time.Duration {
|
||
if s != nil && s.runtimeCacheTTL > 0 {
|
||
return s.runtimeCacheTTL
|
||
}
|
||
return contentModerationRuntimeCacheTTL
|
||
}
|
||
|
||
func (s *ContentModerationService) triggerRuntimeSnapshotRefresh() {
|
||
if s == nil || s.runtimeRefreshDeferred() || !s.runtimeRefreshMu.TryLock() {
|
||
return
|
||
}
|
||
if s.runtimeRefreshDeferred() {
|
||
s.runtimeRefreshMu.Unlock()
|
||
return
|
||
}
|
||
go func() {
|
||
defer s.runtimeRefreshMu.Unlock()
|
||
ctx, cancel := context.WithTimeout(context.Background(), contentModerationRuntimeRefreshTimeout)
|
||
defer cancel()
|
||
if _, err := s.refreshRuntimeSnapshot(ctx); err != nil {
|
||
s.runtimeRefreshRetryAt.Store(time.Now().Add(s.runtimeSnapshotTTL()).UnixNano())
|
||
slog.Warn("content_moderation.runtime_snapshot_refresh_failed", "error", err)
|
||
}
|
||
}()
|
||
}
|
||
|
||
func (s *ContentModerationService) runtimeRefreshDeferred() bool {
|
||
if s == nil {
|
||
return false
|
||
}
|
||
return time.Now().UnixNano() < s.runtimeRefreshRetryAt.Load()
|
||
}
|
||
|
||
func (s *ContentModerationService) refreshRuntimeSnapshot(ctx context.Context) (*contentModerationRuntimeSnapshot, error) {
|
||
values, err := s.settingRepo.GetMultiple(ctx, []string{
|
||
SettingKeyRiskControlEnabled,
|
||
SettingKeyContentModerationConfig,
|
||
})
|
||
if err != nil {
|
||
return nil, fmt.Errorf("get content moderation runtime settings: %w", err)
|
||
}
|
||
rawConfig := values[SettingKeyContentModerationConfig]
|
||
configDigest := sha256.Sum256([]byte(rawConfig))
|
||
if current := s.runtimeSnapshot.Load(); current != nil && current.configDigest == configDigest {
|
||
snapshot := &contentModerationRuntimeSnapshot{
|
||
riskControlEnabled: values[SettingKeyRiskControlEnabled] == "true",
|
||
config: current.config,
|
||
keywordMatcher: current.keywordMatcher,
|
||
configDigest: configDigest,
|
||
loadedAt: time.Now(),
|
||
}
|
||
s.runtimeSnapshot.Store(snapshot)
|
||
s.runtimeRefreshRetryAt.Store(0)
|
||
return snapshot, nil
|
||
}
|
||
cfg, err := parseContentModerationConfig(rawConfig)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
snapshot := &contentModerationRuntimeSnapshot{
|
||
riskControlEnabled: values[SettingKeyRiskControlEnabled] == "true",
|
||
config: cfg,
|
||
keywordMatcher: newContentModerationKeywordMatcher(cfg.BlockedKeywords),
|
||
configDigest: configDigest,
|
||
loadedAt: time.Now(),
|
||
}
|
||
s.runtimeSnapshot.Store(snapshot)
|
||
s.runtimeRefreshRetryAt.Store(0)
|
||
return snapshot, nil
|
||
}
|
||
|
||
func (s *ContentModerationService) replaceRuntimeConfig(cfg *ContentModerationConfig, raw []byte) {
|
||
if s == nil || cfg == nil {
|
||
return
|
||
}
|
||
s.runtimeRefreshMu.Lock()
|
||
hasSnapshot := s.runtimeSnapshot.Load() != nil
|
||
s.runtimeRefreshMu.Unlock()
|
||
if !hasSnapshot {
|
||
return
|
||
}
|
||
config := cloneContentModerationConfig(cfg)
|
||
keywordMatcher := newContentModerationKeywordMatcher(cfg.BlockedKeywords)
|
||
configDigest := sha256.Sum256(raw)
|
||
|
||
s.runtimeRefreshMu.Lock()
|
||
defer s.runtimeRefreshMu.Unlock()
|
||
current := s.runtimeSnapshot.Load()
|
||
if current == nil {
|
||
return
|
||
}
|
||
s.runtimeSnapshot.Store(&contentModerationRuntimeSnapshot{
|
||
riskControlEnabled: current.riskControlEnabled,
|
||
config: config,
|
||
keywordMatcher: keywordMatcher,
|
||
configDigest: configDigest,
|
||
loadedAt: time.Now(),
|
||
})
|
||
}
|
||
|
||
func (s *contentModerationRuntimeSnapshot) matchBlockedKeyword(text string) (string, bool) {
|
||
if s == nil || s.config == nil {
|
||
return "", false
|
||
}
|
||
if s.keywordMatcher != nil {
|
||
return s.keywordMatcher.Match(text)
|
||
}
|
||
return matchBlockedKeyword(text, s.config.BlockedKeywords)
|
||
}
|
||
|
||
func (s *ContentModerationService) isRiskControlEnabled(ctx context.Context) bool {
|
||
raw, err := s.settingRepo.GetValue(ctx, SettingKeyRiskControlEnabled)
|
||
if err != nil {
|
||
return false
|
||
}
|
||
return raw == "true"
|
||
}
|
||
|
||
func (s *ContentModerationService) validateConfig(ctx context.Context, cfg *ContentModerationConfig) error {
|
||
if cfg == nil {
|
||
return infraerrors.BadRequest("INVALID_CONTENT_MODERATION_CONFIG", "内容审计配置不能为空")
|
||
}
|
||
cfg.normalize()
|
||
switch cfg.Mode {
|
||
case ContentModerationModeOff, ContentModerationModeObserve, ContentModerationModePreBlock:
|
||
default:
|
||
return infraerrors.BadRequest("INVALID_CONTENT_MODERATION_MODE", "内容审计模式无效")
|
||
}
|
||
if _, err := url.ParseRequestURI(cfg.BaseURL); err != nil {
|
||
return infraerrors.BadRequest("INVALID_CONTENT_MODERATION_BASE_URL", "OpenAI Base URL 无效")
|
||
}
|
||
if cfg.ProxyID != nil && s.proxyRepo != nil {
|
||
if _, err := s.proxyRepo.GetByID(ctx, *cfg.ProxyID); err != nil {
|
||
return infraerrors.BadRequest("INVALID_CONTENT_MODERATION_PROXY", fmt.Sprintf("代理服务器不存在: %d", *cfg.ProxyID))
|
||
}
|
||
}
|
||
if cfg.BlockStatus < 400 || cfg.BlockStatus > 599 {
|
||
return infraerrors.BadRequest("INVALID_CONTENT_MODERATION_BLOCK_STATUS", "拦截 HTTP 状态码必须在 400-599 之间")
|
||
}
|
||
if cfg.ModelFilter.Type != ContentModerationModelFilterAll && len(cfg.ModelFilter.Models) == 0 {
|
||
return infraerrors.BadRequest("INVALID_CONTENT_MODERATION_MODEL_FILTER", "指定或排除模型时至少需要配置 1 个模型")
|
||
}
|
||
if !cfg.AllGroups && len(cfg.GroupIDs) > 0 && s.groupRepo != nil {
|
||
for _, groupID := range cfg.GroupIDs {
|
||
if _, err := s.groupRepo.GetByIDLite(ctx, groupID); err != nil {
|
||
return infraerrors.BadRequest("INVALID_CONTENT_MODERATION_GROUP", fmt.Sprintf("审计分组不存在: %d", groupID))
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *ContentModerationService) callModeration(ctx context.Context, cfg *ContentModerationConfig, input any, trackKeyLoad ...bool) (*moderationAPIResult, error) {
|
||
attempts := cfg.RetryCount + 1
|
||
if attempts <= 0 {
|
||
attempts = 1
|
||
}
|
||
if attempts > maxContentModerationRetryCount+1 {
|
||
attempts = maxContentModerationRetryCount + 1
|
||
}
|
||
trackLoad := len(trackKeyLoad) > 0 && trackKeyLoad[0]
|
||
var lastErr error
|
||
for attempt := 0; attempt < attempts; attempt++ {
|
||
key, ok := s.nextUsableAPIKey(cfg)
|
||
if !ok {
|
||
lastErr = errors.New("no moderation api key available")
|
||
break
|
||
}
|
||
if trackLoad {
|
||
s.beginModerationAPIKeyCall(key)
|
||
}
|
||
start := time.Now()
|
||
httpStatus := 0
|
||
result, err := s.callModerationOnceWithInput(ctx, cfg, key, input, &httpStatus)
|
||
latency := int(time.Since(start).Milliseconds())
|
||
if err == nil {
|
||
if trackLoad {
|
||
s.finishModerationAPIKeyCall(key, latency, true)
|
||
}
|
||
s.markAPIKeySuccess(key, latency, httpStatus)
|
||
return result, nil
|
||
}
|
||
if trackLoad {
|
||
s.finishModerationAPIKeyCall(key, latency, false)
|
||
}
|
||
s.markAPIKeyError(key, err.Error(), latency, httpStatus)
|
||
lastErr = err
|
||
if httpStatus == http.StatusBadRequest {
|
||
break
|
||
}
|
||
if attempt == attempts-1 {
|
||
break
|
||
}
|
||
wait := time.Duration(100*(attempt+1)) * time.Millisecond
|
||
select {
|
||
case <-ctx.Done():
|
||
return nil, ctx.Err()
|
||
case <-time.After(wait):
|
||
}
|
||
}
|
||
return nil, lastErr
|
||
}
|
||
|
||
func (s *ContentModerationService) callModerationOnceWithInput(ctx context.Context, cfg *ContentModerationConfig, apiKey string, input any, httpStatus *int) (*moderationAPIResult, error) {
|
||
base := strings.TrimRight(cfg.BaseURL, "/")
|
||
endpoint, err := url.JoinPath(base, "/v1/moderations")
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
payload := moderationAPIRequest{
|
||
Model: cfg.Model,
|
||
Input: input,
|
||
}
|
||
raw, err := json.Marshal(payload)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
timeout := time.Duration(cfg.TimeoutMS) * time.Millisecond
|
||
reqCtx, cancel := context.WithTimeout(ctx, timeout)
|
||
defer cancel()
|
||
|
||
req, err := http.NewRequestWithContext(reqCtx, http.MethodPost, endpoint, bytes.NewReader(raw))
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
req.Header.Set("Authorization", "Bearer "+apiKey)
|
||
req.Header.Set("Content-Type", "application/json")
|
||
|
||
client, err := s.moderationHTTPClient(ctx, cfg)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
resp, err := client.Do(req)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer func() { _ = resp.Body.Close() }()
|
||
if httpStatus != nil {
|
||
*httpStatus = resp.StatusCode
|
||
}
|
||
|
||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||
body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
|
||
return nil, fmt.Errorf("moderation api status %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
|
||
}
|
||
var out moderationAPIResponse
|
||
if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
|
||
return nil, err
|
||
}
|
||
if len(out.Results) == 0 {
|
||
return nil, errors.New("moderation api returned empty results")
|
||
}
|
||
return &out.Results[0], nil
|
||
}
|
||
|
||
// moderationProxyURLCacheEntry 缓存 proxy_id 到代理 URL 的解析结果，
|
||
// 避免审计热路径上每次调用都查询数据库。
|
||
type moderationProxyURLCacheEntry struct {
|
||
proxyID int64
|
||
url string
|
||
expiresAt time.Time
|
||
}
|
||
|
||
const contentModerationProxyURLCacheTTL = time.Minute
|
||
|
||
// moderationHTTPClient 返回本次审计调用应使用的 HTTP 客户端。
|
||
// 未配置代理时沿用默认客户端；配置了代理时通过共享客户端池构建，
|
||
// 代理解析/构建失败直接返回错误，绝不回退直连（避免 IP 关联风险）。
|
||
func (s *ContentModerationService) moderationHTTPClient(ctx context.Context, cfg *ContentModerationConfig) (*http.Client, error) {
|
||
if cfg == nil || cfg.ProxyID == nil {
|
||
if s.httpClient == nil {
|
||
return http.DefaultClient, nil
|
||
}
|
||
return s.httpClient, nil
|
||
}
|
||
proxyURL, err := s.resolveModerationProxyURL(ctx, *cfg.ProxyID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
client, err := httpclient.GetClient(httpclient.Options{ProxyURL: proxyURL})
|
||
if err != nil {
|
||
return nil, fmt.Errorf("build moderation proxy client: %w", err)
|
||
}
|
||
return client, nil
|
||
}
|
||
|
||
func (s *ContentModerationService) resolveModerationProxyURL(ctx context.Context, proxyID int64) (string, error) {
|
||
now := time.Now()
|
||
prev := s.moderationProxyCache.Load()
|
||
if prev != nil && prev.proxyID == proxyID && now.Before(prev.expiresAt) {
|
||
return prev.url, nil
|
||
}
|
||
if s.proxyRepo == nil {
|
||
return "", errors.New("moderation proxy repository unavailable")
|
||
}
|
||
px, err := s.proxyRepo.GetByID(ctx, proxyID)
|
||
if err != nil {
|
||
return "", fmt.Errorf("resolve moderation proxy %d: %w", proxyID, err)
|
||
}
|
||
if !px.IsActive() || px.IsExpired(now) {
|
||
slog.Warn("content_moderation.proxy_not_active",
|
||
"proxy_id", proxyID,
|
||
"proxy_name", px.Name,
|
||
"status", px.Status,
|
||
"expired", px.IsExpired(now))
|
||
}
|
||
proxyURL := px.URL()
|
||
if prev == nil || prev.proxyID != proxyID || prev.url != proxyURL {
|
||
// 不打印完整 URL（可能含认证信息），仅记录可定位的地址。
|
||
slog.Info("content_moderation.proxy_enabled",
|
||
"proxy_id", proxyID,
|
||
"proxy_name", px.Name,
|
||
"proxy_addr", fmt.Sprintf("%s://%s:%d", px.Protocol, px.Host, px.Port))
|
||
}
|
||
s.moderationProxyCache.Store(&moderationProxyURLCacheEntry{
|
||
proxyID: proxyID,
|
||
url: proxyURL,
|
||
expiresAt: now.Add(contentModerationProxyURLCacheTTL),
|
||
})
|
||
return proxyURL, nil
|
||
}
|
||
|
||
func (s *ContentModerationService) buildLog(input ContentModerationCheckInput, cfg *ContentModerationConfig, action string, flagged bool, highestCategory string, highestScore float64, scores map[string]float64, text string, latency *int, queueDelay *int, errText string) *ContentModerationLog {
|
||
var userID *int64
|
||
if input.UserID > 0 {
|
||
userID = &input.UserID
|
||
}
|
||
var apiKeyID *int64
|
||
if input.APIKeyID > 0 {
|
||
apiKeyID = &input.APIKeyID
|
||
}
|
||
return &ContentModerationLog{
|
||
RequestID: input.RequestID,
|
||
UserID: userID,
|
||
UserEmail: input.UserEmail,
|
||
APIKeyID: apiKeyID,
|
||
APIKeyName: input.APIKeyName,
|
||
GroupID: cloneInt64Ptr(input.GroupID),
|
||
GroupName: input.GroupName,
|
||
Endpoint: input.Endpoint,
|
||
Provider: input.Provider,
|
||
Model: input.Model,
|
||
Mode: cfg.Mode,
|
||
Action: action,
|
||
Flagged: flagged,
|
||
HighestCategory: highestCategory,
|
||
HighestScore: highestScore,
|
||
CategoryScores: cloneFloatMap(scores),
|
||
ThresholdSnapshot: cloneFloatMap(cfg.Thresholds),
|
||
InputExcerpt: trimRunes(redactContentModerationSecrets(text), maxModerationExcerptRunes),
|
||
UpstreamLatencyMS: latency,
|
||
QueueDelayMS: queueDelay,
|
||
Error: errText,
|
||
}
|
||
}
|
||
|
||
func (s *ContentModerationService) persistContentModerationLog(ctx context.Context, cfg *ContentModerationConfig, log *ContentModerationLog, hashText string, recordHash bool, applySideEffects bool) {
|
||
if s == nil || log == nil {
|
||
return
|
||
}
|
||
if recordHash && s.hashCache != nil {
|
||
if err := s.hashCache.RecordFlaggedInputHash(ctx, hashText); err != nil {
|
||
slog.Warn("content_moderation.record_hash_failed", "user_id", contentModerationEmailUserID(log), "endpoint", log.Endpoint, "error", err)
|
||
}
|
||
}
|
||
autoBanJustApplied := false
|
||
if applySideEffects {
|
||
autoBanJustApplied = s.applyFlaggedAccountSideEffects(ctx, cfg, log)
|
||
s.sendFlaggedNotificationSideEffects(ctx, cfg, log, autoBanJustApplied)
|
||
}
|
||
if s.repo != nil {
|
||
if err := s.repo.CreateLog(ctx, log); err != nil {
|
||
slog.Warn("content_moderation.create_log_failed", "user_id", contentModerationEmailUserID(log), "endpoint", log.Endpoint, "action", log.Action, "error", err)
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *ContentModerationService) applyFlaggedAccountSideEffects(ctx context.Context, cfg *ContentModerationConfig, log *ContentModerationLog) bool {
|
||
if s == nil || cfg == nil || log == nil || !log.Flagged || log.UserID == nil || *log.UserID <= 0 {
|
||
return false
|
||
}
|
||
count := 1
|
||
if s.repo != nil && cfg.ViolationWindowHours > 0 {
|
||
since := time.Now().Add(-time.Duration(cfg.ViolationWindowHours) * time.Hour)
|
||
if n, err := s.repo.CountFlaggedByUserSince(ctx, *log.UserID, since, cfg.CyberPolicyExcludeFromBanCount); err == nil {
|
||
count = n + 1
|
||
}
|
||
}
|
||
log.ViolationCount = count
|
||
autoBanJustApplied := false
|
||
if cfg.AutoBanEnabled && cfg.BanThreshold > 0 && count >= cfg.BanThreshold && s.userRepo != nil {
|
||
user, err := s.userRepo.GetByID(ctx, *log.UserID)
|
||
if err != nil {
|
||
slog.Warn("content_moderation.ban_get_user_failed", "user_id", *log.UserID, "error", err)
|
||
return false
|
||
}
|
||
if user.IsAdmin() {
|
||
slog.Warn("content_moderation.autoban_skipped_admin", "user_id", *log.UserID, "role", user.Role, "count", count, "threshold", cfg.BanThreshold)
|
||
// TODO: Disable the triggering API key instead when API key mutation is available here.
|
||
return false
|
||
}
|
||
if user.Status != StatusDisabled {
|
||
user.Status = StatusDisabled
|
||
if err := s.userRepo.Update(ctx, user, UserUpdateFields{Status: true}); err != nil {
|
||
slog.Warn("content_moderation.ban_update_user_failed", "user_id", *log.UserID, "error", err)
|
||
return false
|
||
}
|
||
if s.authCacheInvalidator != nil {
|
||
s.authCacheInvalidator.InvalidateAuthCacheByUserID(ctx, *log.UserID)
|
||
}
|
||
autoBanJustApplied = true
|
||
}
|
||
log.AutoBanned = true
|
||
}
|
||
return autoBanJustApplied
|
||
}
|
||
|
||
func (s *ContentModerationService) sendFlaggedNotificationSideEffects(ctx context.Context, cfg *ContentModerationConfig, log *ContentModerationLog, autoBanJustApplied bool) {
|
||
if s == nil || cfg == nil || log == nil || !log.Flagged {
|
||
return
|
||
}
|
||
if s.emailService == nil || strings.TrimSpace(log.UserEmail) == "" {
|
||
return
|
||
}
|
||
emailSent := false
|
||
if cfg.EmailOnHit {
|
||
if err := s.sendViolationEmail(ctx, cfg, log); err != nil {
|
||
slog.Warn("content_moderation.email_failed", "user_id", *log.UserID, "email", log.UserEmail, "error", err)
|
||
} else {
|
||
emailSent = true
|
||
}
|
||
}
|
||
if autoBanJustApplied {
|
||
if err := s.sendAccountDisabledEmail(ctx, cfg, log); err != nil {
|
||
slog.Warn("content_moderation.ban_email_failed", "user_id", *log.UserID, "email", log.UserEmail, "error", err)
|
||
} else {
|
||
emailSent = true
|
||
}
|
||
}
|
||
log.EmailSent = emailSent
|
||
}
|
||
|
||
func (s *ContentModerationService) sendViolationEmail(ctx context.Context, cfg *ContentModerationConfig, log *ContentModerationLog) error {
|
||
siteName := s.siteName(ctx)
|
||
if s.emailService.notificationEmailService != nil {
|
||
if err := s.emailService.notificationEmailService.Send(ctx, NotificationEmailSendInput{
|
||
Event: NotificationEmailEventContentModerationViolation,
|
||
RecipientEmail: log.UserEmail,
|
||
RecipientName: emailRecipientName(log.UserEmail),
|
||
UserID: contentModerationEmailUserID(log),
|
||
SourceType: "content_moderation",
|
||
SourceID: contentModerationEmailSourceID(log),
|
||
Variables: contentModerationEmailVariables(log, cfg),
|
||
}); err == nil {
|
||
return nil
|
||
} else {
|
||
if !shouldFallbackNotificationEmail(err) {
|
||
return err
|
||
}
|
||
slog.Warn("template content moderation violation email failed; falling back to built-in body", "log_id", log.ID, "recipient_hash", notificationEmailHash(log.UserEmail), "err", err.Error())
|
||
}
|
||
}
|
||
subject := fmt.Sprintf("[%s] 账户风控提醒 / Risk Control Notice", sanitizeEmailHeader(siteName))
|
||
body := buildContentModerationViolationEmailBody(siteName, log, cfg)
|
||
return s.emailService.SendEmail(ctx, log.UserEmail, subject, body)
|
||
}
|
||
|
||
func (s *ContentModerationService) sendAccountDisabledEmail(ctx context.Context, cfg *ContentModerationConfig, log *ContentModerationLog) error {
|
||
siteName := s.siteName(ctx)
|
||
if s.emailService.notificationEmailService != nil {
|
||
if err := s.emailService.notificationEmailService.Send(ctx, NotificationEmailSendInput{
|
||
Event: NotificationEmailEventContentModerationDisabled,
|
||
RecipientEmail: log.UserEmail,
|
||
RecipientName: emailRecipientName(log.UserEmail),
|
||
UserID: contentModerationEmailUserID(log),
|
||
SourceType: "content_moderation",
|
||
SourceID: contentModerationEmailSourceID(log),
|
||
Variables: contentModerationEmailVariables(log, cfg),
|
||
}); err == nil {
|
||
return nil
|
||
} else {
|
||
if !shouldFallbackNotificationEmail(err) {
|
||
return err
|
||
}
|
||
slog.Warn("template content moderation disabled email failed; falling back to built-in body", "log_id", log.ID, "recipient_hash", notificationEmailHash(log.UserEmail), "err", err.Error())
|
||
}
|
||
}
|
||
subject := fmt.Sprintf("[%s] 账户已被禁用 / Account Disabled", sanitizeEmailHeader(siteName))
|
||
body := buildContentModerationAccountDisabledEmailBody(siteName, log, cfg)
|
||
return s.emailService.SendEmail(ctx, log.UserEmail, subject, body)
|
||
}
|
||
|
||
func contentModerationEmailUserID(log *ContentModerationLog) int64 {
|
||
if log == nil || log.UserID == nil {
|
||
return 0
|
||
}
|
||
return *log.UserID
|
||
}
|
||
|
||
func contentModerationEmailSourceID(log *ContentModerationLog) string {
|
||
if log == nil || log.ID <= 0 {
|
||
return ""
|
||
}
|
||
return fmt.Sprintf("%d", log.ID)
|
||
}
|
||
|
||
func contentModerationEmailVariables(log *ContentModerationLog, cfg *ContentModerationConfig) map[string]string {
|
||
variables := map[string]string{
|
||
"triggered_at": time.Now().UTC().Format(time.RFC3339),
|
||
"group_name": "-",
|
||
"moderation_category": "-",
|
||
"moderation_score": "0.000",
|
||
"violation_count": "0",
|
||
"ban_threshold": "0",
|
||
}
|
||
if log != nil {
|
||
if !log.CreatedAt.IsZero() {
|
||
variables["triggered_at"] = log.CreatedAt.UTC().Format(time.RFC3339)
|
||
}
|
||
if strings.TrimSpace(log.GroupName) != "" {
|
||
variables["group_name"] = strings.TrimSpace(log.GroupName)
|
||
}
|
||
if strings.TrimSpace(log.HighestCategory) != "" {
|
||
variables["moderation_category"] = strings.TrimSpace(log.HighestCategory)
|
||
}
|
||
variables["moderation_score"] = fmt.Sprintf("%.3f", log.HighestScore)
|
||
variables["violation_count"] = fmt.Sprintf("%d", log.ViolationCount)
|
||
}
|
||
if cfg != nil {
|
||
variables["ban_threshold"] = fmt.Sprintf("%d", cfg.BanThreshold)
|
||
}
|
||
return variables
|
||
}
|
||
|
||
func (s *ContentModerationService) siteName(ctx context.Context) string {
|
||
if s == nil || s.settingRepo == nil {
|
||
return "Sub2API"
|
||
}
|
||
name, err := s.settingRepo.GetValue(ctx, SettingKeySiteName)
|
||
if err != nil || strings.TrimSpace(name) == "" {
|
||
return "Sub2API"
|
||
}
|
||
return strings.TrimSpace(name)
|
||
}
|
||
|
||
func defaultContentModerationConfig() *ContentModerationConfig {
|
||
return &ContentModerationConfig{
|
||
Enabled: false,
|
||
Mode: ContentModerationModePreBlock,
|
||
BaseURL: defaultContentModerationBaseURL,
|
||
Model: defaultContentModerationModel,
|
||
TimeoutMS: defaultContentModerationTimeoutMS,
|
||
SampleRate: 100,
|
||
AllGroups: true,
|
||
GroupIDs: []int64{},
|
||
RecordNonHits: false,
|
||
Thresholds: ContentModerationDefaultThresholds(),
|
||
WorkerCount: defaultContentModerationWorkerCount,
|
||
QueueSize: defaultContentModerationQueueSize,
|
||
BlockStatus: defaultContentModerationBlockHTTPStatus,
|
||
BlockMessage: defaultContentModerationBlockMessage,
|
||
EmailOnHit: true,
|
||
AutoBanEnabled: true,
|
||
BanThreshold: defaultContentModerationBanThreshold,
|
||
ViolationWindowHours: defaultContentModerationViolationWindowHours,
|
||
RetryCount: defaultContentModerationRetryCount,
|
||
HitRetentionDays: defaultContentModerationHitRetentionDays,
|
||
NonHitRetentionDays: defaultContentModerationNonHitRetentionDays,
|
||
PreHashCheckEnabled: false,
|
||
BlockedKeywords: []string{},
|
||
KeywordBlockingMode: ContentModerationKeywordModeKeywordAndAPI,
|
||
ModelFilter: ContentModerationModelFilter{
|
||
Type: ContentModerationModelFilterAll,
|
||
Models: []string{},
|
||
},
|
||
CyberPolicyExcludeFromBanCount: false,
|
||
}
|
||
}
|
||
|
||
func cloneContentModerationConfig(cfg *ContentModerationConfig) *ContentModerationConfig {
|
||
if cfg == nil {
|
||
return nil
|
||
}
|
||
clone := *cfg
|
||
clone.ProxyID = cloneInt64Ptr(cfg.ProxyID)
|
||
clone.APIKeys = append([]string(nil), cfg.APIKeys...)
|
||
clone.GroupIDs = append([]int64(nil), cfg.GroupIDs...)
|
||
clone.BlockedKeywords = append([]string(nil), cfg.BlockedKeywords...)
|
||
clone.Thresholds = cloneFloatMap(cfg.Thresholds)
|
||
clone.ModelFilter = ContentModerationModelFilter{
|
||
Type: cfg.ModelFilter.Type,
|
||
Models: append([]string(nil), cfg.ModelFilter.Models...),
|
||
}
|
||
return &clone
|
||
}
|
||
|
||
func (cfg *ContentModerationConfig) normalize() {
|
||
if cfg.APIKey != "" {
|
||
cfg.APIKeys = normalizeModerationAPIKeys(append(cfg.APIKeys, cfg.APIKey))
|
||
cfg.APIKey = ""
|
||
} else {
|
||
cfg.APIKeys = normalizeModerationAPIKeys(cfg.APIKeys)
|
||
}
|
||
if cfg.Mode == "" {
|
||
cfg.Mode = ContentModerationModePreBlock
|
||
}
|
||
if cfg.BaseURL == "" {
|
||
cfg.BaseURL = defaultContentModerationBaseURL
|
||
}
|
||
cfg.BaseURL = strings.TrimRight(strings.TrimSpace(cfg.BaseURL), "/")
|
||
if cfg.Model == "" {
|
||
cfg.Model = defaultContentModerationModel
|
||
}
|
||
cfg.Model = strings.TrimSpace(cfg.Model)
|
||
if cfg.ProxyID != nil && *cfg.ProxyID <= 0 {
|
||
cfg.ProxyID = nil
|
||
}
|
||
if cfg.TimeoutMS <= 0 {
|
||
cfg.TimeoutMS = defaultContentModerationTimeoutMS
|
||
}
|
||
if cfg.TimeoutMS > maxContentModerationTimeoutMS {
|
||
cfg.TimeoutMS = maxContentModerationTimeoutMS
|
||
}
|
||
if cfg.SampleRate < 0 {
|
||
cfg.SampleRate = 0
|
||
}
|
||
if cfg.SampleRate > 100 {
|
||
cfg.SampleRate = 100
|
||
}
|
||
if cfg.WorkerCount <= 0 {
|
||
cfg.WorkerCount = defaultContentModerationWorkerCount
|
||
}
|
||
if cfg.WorkerCount > maxContentModerationWorkerCount {
|
||
cfg.WorkerCount = maxContentModerationWorkerCount
|
||
}
|
||
if cfg.QueueSize <= 0 {
|
||
cfg.QueueSize = defaultContentModerationQueueSize
|
||
}
|
||
if cfg.QueueSize > maxContentModerationQueueSize {
|
||
cfg.QueueSize = maxContentModerationQueueSize
|
||
}
|
||
if strings.TrimSpace(cfg.BlockMessage) == "" {
|
||
cfg.BlockMessage = defaultContentModerationBlockMessage
|
||
}
|
||
cfg.BlockMessage = strings.TrimSpace(cfg.BlockMessage)
|
||
if cfg.BlockStatus <= 0 {
|
||
cfg.BlockStatus = defaultContentModerationBlockHTTPStatus
|
||
}
|
||
if cfg.BanThreshold <= 0 {
|
||
cfg.BanThreshold = defaultContentModerationBanThreshold
|
||
}
|
||
if cfg.ViolationWindowHours <= 0 {
|
||
cfg.ViolationWindowHours = defaultContentModerationViolationWindowHours
|
||
}
|
||
if cfg.RetryCount < 0 {
|
||
cfg.RetryCount = 0
|
||
}
|
||
if cfg.RetryCount > maxContentModerationRetryCount {
|
||
cfg.RetryCount = maxContentModerationRetryCount
|
||
}
|
||
if cfg.HitRetentionDays <= 0 {
|
||
cfg.HitRetentionDays = defaultContentModerationHitRetentionDays
|
||
}
|
||
if cfg.HitRetentionDays > maxContentModerationRetentionDays {
|
||
cfg.HitRetentionDays = maxContentModerationRetentionDays
|
||
}
|
||
if cfg.NonHitRetentionDays <= 0 {
|
||
cfg.NonHitRetentionDays = defaultContentModerationNonHitRetentionDays
|
||
}
|
||
if cfg.NonHitRetentionDays > maxContentModerationNonHitRetentionDays {
|
||
cfg.NonHitRetentionDays = maxContentModerationNonHitRetentionDays
|
||
}
|
||
cfg.GroupIDs = normalizeInt64IDs(cfg.GroupIDs)
|
||
cfg.Thresholds = mergeContentModerationThresholds(ContentModerationDefaultThresholds(), cfg.Thresholds)
|
||
cfg.BlockedKeywords = normalizeBlockedKeywords(cfg.BlockedKeywords)
|
||
cfg.KeywordBlockingMode = normalizeKeywordBlockingMode(cfg.KeywordBlockingMode)
|
||
cfg.ModelFilter = normalizeContentModerationModelFilter(cfg.ModelFilter)
|
||
}
|
||
|
||
func (cfg *ContentModerationConfig) includesGroup(groupID *int64) bool {
|
||
if cfg.AllGroups {
|
||
return true
|
||
}
|
||
if groupID == nil {
|
||
return false
|
||
}
|
||
for _, id := range cfg.GroupIDs {
|
||
if id == *groupID {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func (cfg *ContentModerationConfig) includesModel(model string) bool {
|
||
if cfg == nil {
|
||
return true
|
||
}
|
||
filter := normalizeContentModerationModelFilter(cfg.ModelFilter)
|
||
switch filter.Type {
|
||
case ContentModerationModelFilterInclude:
|
||
return contentModerationModelListContains(filter.Models, model)
|
||
case ContentModerationModelFilterExclude:
|
||
return !contentModerationModelListContains(filter.Models, model)
|
||
default:
|
||
return true
|
||
}
|
||
}
|
||
|
||
func contentModerationLogGroupID(groupID *int64) int64 {
|
||
if groupID == nil {
|
||
return 0
|
||
}
|
||
return *groupID
|
||
}
|
||
|
||
func (cfg *ContentModerationConfig) shouldSample(hashText string) bool {
|
||
if cfg.SampleRate >= 100 {
|
||
return true
|
||
}
|
||
if cfg.SampleRate <= 0 {
|
||
return false
|
||
}
|
||
raw, err := hex.DecodeString(hashText)
|
||
if err != nil || len(raw) < 2 {
|
||
return true
|
||
}
|
||
return int(binary.BigEndian.Uint16(raw[:2])%100) < cfg.SampleRate
|
||
}
|
||
|
||
func (cfg *ContentModerationConfig) apiKeys() []string {
|
||
if cfg == nil {
|
||
return nil
|
||
}
|
||
return normalizeModerationAPIKeys(cfg.APIKeys)
|
||
}
|
||
|
||
func (s *ContentModerationService) nextUsableAPIKey(cfg *ContentModerationConfig) (string, bool) {
|
||
keys := cfg.apiKeys()
|
||
if len(keys) == 0 {
|
||
return "", false
|
||
}
|
||
now := time.Now()
|
||
for i := 0; i < len(keys); i++ {
|
||
idx := int(s.apiKeyCursor.Add(1)-1) % len(keys)
|
||
key := keys[idx]
|
||
if !s.isAPIKeyFrozen(key, now) {
|
||
return key, true
|
||
}
|
||
}
|
||
return "", false
|
||
}
|
||
|
||
func (s *ContentModerationService) isAPIKeyFrozen(key string, now time.Time) bool {
|
||
hash := moderationAPIKeyHash(key)
|
||
if hash == "" || s == nil {
|
||
return false
|
||
}
|
||
s.keyHealthMu.Lock()
|
||
defer s.keyHealthMu.Unlock()
|
||
state := s.keyHealth[hash]
|
||
return state != nil && state.FrozenUntil.After(now)
|
||
}
|
||
|
||
func (s *ContentModerationService) beginModerationAPIKeyCall(key string) {
|
||
hash := moderationAPIKeyHash(key)
|
||
if hash == "" || s == nil {
|
||
return
|
||
}
|
||
s.keyHealthMu.Lock()
|
||
defer s.keyHealthMu.Unlock()
|
||
state := s.ensureAPIKeyHealthLocked(hash, maskSecretTail(key))
|
||
state.SyncActive++
|
||
}
|
||
|
||
func (s *ContentModerationService) finishModerationAPIKeyCall(key string, latencyMS int, success bool) {
|
||
hash := moderationAPIKeyHash(key)
|
||
if hash == "" || s == nil {
|
||
return
|
||
}
|
||
if latencyMS < 0 {
|
||
latencyMS = 0
|
||
}
|
||
s.keyHealthMu.Lock()
|
||
defer s.keyHealthMu.Unlock()
|
||
state := s.ensureAPIKeyHealthLocked(hash, maskSecretTail(key))
|
||
if state.SyncActive > 0 {
|
||
state.SyncActive--
|
||
}
|
||
state.SyncTotal++
|
||
state.SyncLatencyMS += int64(latencyMS)
|
||
if success {
|
||
state.SyncSuccess++
|
||
return
|
||
}
|
||
state.SyncErrors++
|
||
}
|
||
|
||
func (s *ContentModerationService) markAPIKeySuccess(key string, latencyMS int, httpStatus int) {
|
||
hash := moderationAPIKeyHash(key)
|
||
if hash == "" || s == nil {
|
||
return
|
||
}
|
||
s.keyHealthMu.Lock()
|
||
defer s.keyHealthMu.Unlock()
|
||
state := s.ensureAPIKeyHealthLocked(hash, maskSecretTail(key))
|
||
state.FailureCount = 0
|
||
state.SuccessCount++
|
||
state.LastError = ""
|
||
state.LastCheckedAt = time.Now()
|
||
state.FrozenUntil = time.Time{}
|
||
state.LastLatencyMS = latencyMS
|
||
state.LastHTTPStatus = httpStatus
|
||
state.LastTested = true
|
||
}
|
||
|
||
func (s *ContentModerationService) markAPIKeyError(key string, errText string, latencyMS int, httpStatus int) {
|
||
hash := moderationAPIKeyHash(key)
|
||
if hash == "" || s == nil {
|
||
return
|
||
}
|
||
s.keyHealthMu.Lock()
|
||
defer s.keyHealthMu.Unlock()
|
||
state := s.ensureAPIKeyHealthLocked(hash, maskSecretTail(key))
|
||
if contentModerationFreezeDurationForHTTPStatus(httpStatus) > 0 {
|
||
state.FailureCount++
|
||
}
|
||
state.LastError = trimRunes(errText, 180)
|
||
state.LastCheckedAt = time.Now()
|
||
state.LastLatencyMS = latencyMS
|
||
state.LastHTTPStatus = httpStatus
|
||
state.LastTested = true
|
||
if freezeDuration := contentModerationFreezeDurationForHTTPStatus(httpStatus); freezeDuration > 0 {
|
||
state.FrozenUntil = time.Now().Add(freezeDuration)
|
||
}
|
||
}
|
||
|
||
func contentModerationFreezeDurationForHTTPStatus(httpStatus int) time.Duration {
|
||
switch httpStatus {
|
||
case 0, http.StatusBadRequest:
|
||
return 0
|
||
case http.StatusUnauthorized, http.StatusForbidden:
|
||
return contentModerationKeyAuthFreezeDuration
|
||
case http.StatusTooManyRequests, 529:
|
||
return contentModerationKeyRateLimitFreezeDuration
|
||
default:
|
||
return contentModerationKeyHTTPErrorFreezeDuration
|
||
}
|
||
}
|
||
|
||
func (s *ContentModerationService) ensureAPIKeyHealthLocked(hash string, masked string) *contentModerationKeyHealth {
|
||
if s.keyHealth == nil {
|
||
s.keyHealth = make(map[string]*contentModerationKeyHealth)
|
||
}
|
||
state := s.keyHealth[hash]
|
||
if state == nil {
|
||
state = &contentModerationKeyHealth{Hash: hash}
|
||
s.keyHealth[hash] = state
|
||
}
|
||
if strings.TrimSpace(masked) != "" {
|
||
state.Masked = masked
|
||
}
|
||
return state
|
||
}
|
||
|
||
func (s *ContentModerationService) configView(cfg *ContentModerationConfig) *ContentModerationConfigView {
|
||
keys := cfg.apiKeys()
|
||
masks := make([]string, 0, len(keys))
|
||
for _, key := range keys {
|
||
masks = append(masks, maskSecretTail(key))
|
||
}
|
||
apiKeyMasked := ""
|
||
if len(masks) > 0 {
|
||
apiKeyMasked = masks[0]
|
||
}
|
||
return &ContentModerationConfigView{
|
||
Enabled: cfg.Enabled,
|
||
Mode: cfg.Mode,
|
||
BaseURL: cfg.BaseURL,
|
||
Model: cfg.Model,
|
||
ProxyID: cloneInt64Ptr(cfg.ProxyID),
|
||
APIKeyConfigured: len(keys) > 0,
|
||
APIKeyMasked: apiKeyMasked,
|
||
APIKeyCount: len(keys),
|
||
APIKeyMasks: masks,
|
||
APIKeyStatuses: s.apiKeyStatuses(keys),
|
||
TimeoutMS: cfg.TimeoutMS,
|
||
SampleRate: cfg.SampleRate,
|
||
AllGroups: cfg.AllGroups,
|
||
GroupIDs: append([]int64(nil), cfg.GroupIDs...),
|
||
RecordNonHits: cfg.RecordNonHits,
|
||
Thresholds: cloneFloatMap(cfg.Thresholds),
|
||
WorkerCount: cfg.WorkerCount,
|
||
QueueSize: cfg.QueueSize,
|
||
BlockStatus: cfg.BlockStatus,
|
||
BlockMessage: cfg.BlockMessage,
|
||
EmailOnHit: cfg.EmailOnHit,
|
||
AutoBanEnabled: cfg.AutoBanEnabled,
|
||
BanThreshold: cfg.BanThreshold,
|
||
ViolationWindowHours: cfg.ViolationWindowHours,
|
||
RetryCount: cfg.RetryCount,
|
||
HitRetentionDays: cfg.HitRetentionDays,
|
||
NonHitRetentionDays: cfg.NonHitRetentionDays,
|
||
PreHashCheckEnabled: cfg.PreHashCheckEnabled,
|
||
BlockedKeywords: append([]string(nil), cfg.BlockedKeywords...),
|
||
KeywordBlockingMode: cfg.KeywordBlockingMode,
|
||
ModelFilter: cloneContentModerationModelFilter(cfg.ModelFilter),
|
||
CyberPolicyExcludeFromBanCount: cfg.CyberPolicyExcludeFromBanCount,
|
||
}
|
||
}
|
||
|
||
func (s *ContentModerationService) apiKeyStatuses(keys []string) []ContentModerationAPIKeyStatus {
|
||
out := make([]ContentModerationAPIKeyStatus, 0, len(keys))
|
||
for idx, key := range keys {
|
||
out = append(out, s.apiKeyStatusForHash(idx, moderationAPIKeyHash(key), maskSecretTail(key), true))
|
||
}
|
||
return out
|
||
}
|
||
|
||
func (s *ContentModerationService) preBlockAPIKeyLoads(keys []string) []ContentModerationAPIKeyLoad {
|
||
out := make([]ContentModerationAPIKeyLoad, 0, len(keys))
|
||
for idx, key := range keys {
|
||
out = append(out, s.preBlockAPIKeyLoadForHash(idx, moderationAPIKeyHash(key), maskSecretTail(key)))
|
||
}
|
||
return out
|
||
}
|
||
|
||
func (s *ContentModerationService) preBlockAPIKeyActive(keys []string) int64 {
|
||
var total int64
|
||
for _, item := range s.preBlockAPIKeyLoads(keys) {
|
||
total += item.Active
|
||
}
|
||
return total
|
||
}
|
||
|
||
func (s *ContentModerationService) preBlockAPIKeyAvailableCount(keys []string) int64 {
|
||
now := time.Now()
|
||
var count int64
|
||
for _, key := range keys {
|
||
if !s.isAPIKeyFrozen(key, now) {
|
||
count++
|
||
}
|
||
}
|
||
return count
|
||
}
|
||
|
||
func (s *ContentModerationService) preBlockAPIKeyTotalCalls(keys []string) int64 {
|
||
var total int64
|
||
for _, item := range s.preBlockAPIKeyLoads(keys) {
|
||
total += item.Total
|
||
}
|
||
return total
|
||
}
|
||
|
||
func (s *ContentModerationService) preBlockAPIKeyLoadForHash(index int, hash string, masked string) ContentModerationAPIKeyLoad {
|
||
load := ContentModerationAPIKeyLoad{
|
||
Index: index,
|
||
KeyHash: hash,
|
||
Masked: masked,
|
||
Status: "unknown",
|
||
}
|
||
status := s.apiKeyStatusForHash(index, hash, masked, true)
|
||
load.Status = status.Status
|
||
load.LastLatencyMS = status.LastLatencyMS
|
||
load.LastHTTPStatus = status.LastHTTPStatus
|
||
if hash == "" || s == nil {
|
||
return load
|
||
}
|
||
s.keyHealthMu.Lock()
|
||
defer s.keyHealthMu.Unlock()
|
||
state := s.keyHealth[hash]
|
||
if state == nil {
|
||
return load
|
||
}
|
||
load.Active = state.SyncActive
|
||
load.Total = state.SyncTotal
|
||
load.Success = state.SyncSuccess
|
||
load.Errors = state.SyncErrors
|
||
if state.SyncTotal > 0 {
|
||
load.AvgLatencyMS = state.SyncLatencyMS / state.SyncTotal
|
||
}
|
||
return load
|
||
}
|
||
|
||
func (s *ContentModerationService) apiKeyStatusForHash(index int, hash string, masked string, configured bool) ContentModerationAPIKeyStatus {
|
||
status := ContentModerationAPIKeyStatus{
|
||
Index: index,
|
||
KeyHash: hash,
|
||
Masked: masked,
|
||
Status: "unknown",
|
||
Configured: configured,
|
||
}
|
||
if hash == "" || s == nil {
|
||
return status
|
||
}
|
||
now := time.Now()
|
||
s.keyHealthMu.Lock()
|
||
defer s.keyHealthMu.Unlock()
|
||
state := s.keyHealth[hash]
|
||
if state == nil {
|
||
return status
|
||
}
|
||
status.FailureCount = state.FailureCount
|
||
status.SuccessCount = state.SuccessCount
|
||
status.LastError = state.LastError
|
||
status.LastLatencyMS = state.LastLatencyMS
|
||
status.LastHTTPStatus = state.LastHTTPStatus
|
||
status.LastTested = state.LastTested
|
||
if !state.LastCheckedAt.IsZero() {
|
||
t := state.LastCheckedAt
|
||
status.LastCheckedAt = &t
|
||
}
|
||
if state.FrozenUntil.After(now) {
|
||
t := state.FrozenUntil
|
||
status.FrozenUntil = &t
|
||
status.Status = "frozen"
|
||
return status
|
||
}
|
||
if state.LastError != "" {
|
||
status.Status = "error"
|
||
return status
|
||
}
|
||
if state.SuccessCount > 0 || state.LastTested {
|
||
status.Status = "ok"
|
||
}
|
||
return status
|
||
}
|
||
|
||
func moderationAPIKeyHash(key string) string {
|
||
key = strings.TrimSpace(key)
|
||
if key == "" {
|
||
return ""
|
||
}
|
||
sum := sha256.Sum256([]byte(key))
|
||
return hex.EncodeToString(sum[:])
|
||
}
|
||
|
||
func buildModerationTestInput(prompt string, images []string) (any, int, error) {
|
||
prompt = trimRunes(normalizeContentModerationText(prompt), maxModerationInputRunes)
|
||
normalizedImages := make([]string, 0, len(images))
|
||
for _, image := range images {
|
||
image = strings.TrimSpace(image)
|
||
if image == "" {
|
||
continue
|
||
}
|
||
if len(normalizedImages) >= maxContentModerationTestImages {
|
||
return nil, 0, infraerrors.BadRequest("TOO_MANY_MODERATION_TEST_IMAGES", fmt.Sprintf("最多上传 %d 张测试图片", maxContentModerationTestImages))
|
||
}
|
||
if err := validateModerationTestImageDataURL(image); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
normalizedImages = append(normalizedImages, image)
|
||
}
|
||
if prompt == "" && len(normalizedImages) == 0 {
|
||
return "hello", 0, nil
|
||
}
|
||
if len(normalizedImages) == 0 {
|
||
return prompt, 0, nil
|
||
}
|
||
parts := make([]moderationAPIInputPart, 0, len(normalizedImages)+1)
|
||
if prompt != "" {
|
||
parts = append(parts, moderationAPIInputPart{Type: "text", Text: prompt})
|
||
}
|
||
for _, image := range normalizedImages {
|
||
parts = append(parts, moderationAPIInputPart{
|
||
Type: "image_url",
|
||
ImageURL: &moderationAPIImageURLRef{URL: image},
|
||
})
|
||
}
|
||
return parts, len(normalizedImages), nil
|
||
}
|
||
|
||
func contentModerationTestHasAuditInput(prompt string, images []string) bool {
|
||
if normalizeContentModerationText(prompt) != "" {
|
||
return true
|
||
}
|
||
for _, image := range images {
|
||
if strings.TrimSpace(image) != "" {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func validateModerationTestImageDataURL(value string) error {
|
||
if len(value) > maxContentModerationTestImageDataURLBytes {
|
||
return infraerrors.BadRequest("MODERATION_TEST_IMAGE_TOO_LARGE", "测试图片不能超过 8MB")
|
||
}
|
||
if !strings.HasPrefix(value, "data:image/") {
|
||
return infraerrors.BadRequest("INVALID_MODERATION_TEST_IMAGE", "测试图片必须是 data:image/* base64")
|
||
}
|
||
parts := strings.SplitN(value, ",", 2)
|
||
if len(parts) != 2 || !strings.Contains(parts[0], ";base64") {
|
||
return infraerrors.BadRequest("INVALID_MODERATION_TEST_IMAGE", "测试图片必须是 base64 data URL")
|
||
}
|
||
raw, err := base64.StdEncoding.DecodeString(parts[1])
|
||
if err != nil {
|
||
return infraerrors.BadRequest("INVALID_MODERATION_TEST_IMAGE", "测试图片 base64 无效")
|
||
}
|
||
if len(raw) > maxContentModerationTestImageBytes {
|
||
return infraerrors.BadRequest("MODERATION_TEST_IMAGE_TOO_LARGE", "测试图片不能超过 8MB")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func buildContentModerationTestAuditResult(result *moderationAPIResult, thresholds map[string]float64) *ContentModerationTestAuditResult {
|
||
if result == nil {
|
||
return nil
|
||
}
|
||
scores := make(map[string]float64, len(result.CategoryScores))
|
||
for category, score := range result.CategoryScores {
|
||
scores[category] = score
|
||
}
|
||
thresholdSnapshot := mergeContentModerationThresholds(ContentModerationDefaultThresholds(), thresholds)
|
||
flagged, highestCategory, highestScore := evaluateModerationScores(scores, thresholdSnapshot)
|
||
compositeScore := highestScore
|
||
return &ContentModerationTestAuditResult{
|
||
Flagged: flagged,
|
||
HighestCategory: highestCategory,
|
||
HighestScore: highestScore,
|
||
CompositeScore: compositeScore,
|
||
CategoryScores: scores,
|
||
Thresholds: thresholdSnapshot,
|
||
}
|
||
}
|
||
|
||
type moderationAPIRequest struct {
|
||
Model string `json:"model"`
|
||
Input any `json:"input"`
|
||
}
|
||
|
||
type moderationAPIInputPart struct {
|
||
Type string `json:"type"`
|
||
Text string `json:"text,omitempty"`
|
||
ImageURL *moderationAPIImageURLRef `json:"image_url,omitempty"`
|
||
}
|
||
|
||
type moderationAPIImageURLRef struct {
|
||
URL string `json:"url"`
|
||
}
|
||
|
||
type moderationAPIResponse struct {
|
||
Results []moderationAPIResult `json:"results"`
|
||
}
|
||
|
||
type moderationAPIResult struct {
|
||
Flagged bool `json:"flagged"`
|
||
CategoryScores map[string]float64 `json:"category_scores"`
|
||
}
|
||
|
||
func evaluateModerationScores(scores map[string]float64, thresholds map[string]float64) (bool, string, float64) {
|
||
flagged := false
|
||
highestCategory := ""
|
||
highestScore := 0.0
|
||
for _, category := range contentModerationCategoryOrder {
|
||
score := scores[category]
|
||
if score > highestScore || highestCategory == "" {
|
||
highestScore = score
|
||
highestCategory = category
|
||
}
|
||
if score >= thresholds[category] {
|
||
flagged = true
|
||
}
|
||
}
|
||
for category, score := range scores {
|
||
if score > highestScore || highestCategory == "" {
|
||
highestScore = score
|
||
highestCategory = category
|
||
}
|
||
}
|
||
return flagged, highestCategory, highestScore
|
||
}
|
||
|
||
func mergeContentModerationThresholds(base map[string]float64, override map[string]float64) map[string]float64 {
|
||
out := cloneFloatMap(base)
|
||
if out == nil {
|
||
out = map[string]float64{}
|
||
}
|
||
for _, category := range contentModerationCategoryOrder {
|
||
if v, ok := override[category]; ok {
|
||
if v < 0 {
|
||
v = 0
|
||
}
|
||
if v > 1 {
|
||
v = 1
|
||
}
|
||
out[category] = v
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func normalizeInt64IDs(ids []int64) []int64 {
|
||
if len(ids) == 0 {
|
||
return []int64{}
|
||
}
|
||
seen := make(map[int64]struct{}, len(ids))
|
||
out := make([]int64, 0, len(ids))
|
||
for _, id := range ids {
|
||
if id <= 0 {
|
||
continue
|
||
}
|
||
if _, ok := seen[id]; ok {
|
||
continue
|
||
}
|
||
seen[id] = struct{}{}
|
||
out = append(out, id)
|
||
}
|
||
sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
|
||
return out
|
||
}
|
||
|
||
func normalizeBlockedKeywords(in []string) []string {
|
||
if len(in) == 0 {
|
||
return []string{}
|
||
}
|
||
out := make([]string, 0, len(in))
|
||
seen := make(map[string]struct{}, len(in))
|
||
for _, raw := range in {
|
||
kw := strings.TrimSpace(raw)
|
||
if kw == "" {
|
||
continue
|
||
}
|
||
kw = trimRunes(kw, maxContentModerationBlockedKeywordRunes)
|
||
key := strings.ToLower(kw)
|
||
if _, ok := seen[key]; ok {
|
||
continue
|
||
}
|
||
seen[key] = struct{}{}
|
||
out = append(out, kw)
|
||
if len(out) >= maxContentModerationBlockedKeywords {
|
||
break
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func normalizeKeywordBlockingMode(mode string) string {
|
||
switch strings.TrimSpace(mode) {
|
||
case ContentModerationKeywordModeKeywordOnly:
|
||
return ContentModerationKeywordModeKeywordOnly
|
||
case ContentModerationKeywordModeAPIOnly:
|
||
return ContentModerationKeywordModeAPIOnly
|
||
case ContentModerationKeywordModeKeywordAndAPI:
|
||
return ContentModerationKeywordModeKeywordAndAPI
|
||
default:
|
||
return ContentModerationKeywordModeKeywordAndAPI
|
||
}
|
||
}
|
||
|
||
func normalizeContentModerationModelFilter(filter ContentModerationModelFilter) ContentModerationModelFilter {
|
||
out := ContentModerationModelFilter{
|
||
Type: normalizeContentModerationModelFilterType(filter.Type),
|
||
Models: normalizeContentModerationModelNames(filter.Models),
|
||
}
|
||
if out.Type == ContentModerationModelFilterAll {
|
||
out.Models = []string{}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func cloneContentModerationModelFilter(filter ContentModerationModelFilter) ContentModerationModelFilter {
|
||
normalized := normalizeContentModerationModelFilter(filter)
|
||
normalized.Models = append([]string(nil), normalized.Models...)
|
||
return normalized
|
||
}
|
||
|
||
func normalizeContentModerationModelFilterType(filterType string) string {
|
||
switch strings.ToLower(strings.TrimSpace(filterType)) {
|
||
case ContentModerationModelFilterInclude:
|
||
return ContentModerationModelFilterInclude
|
||
case ContentModerationModelFilterExclude:
|
||
return ContentModerationModelFilterExclude
|
||
case ContentModerationModelFilterAll:
|
||
return ContentModerationModelFilterAll
|
||
default:
|
||
return ContentModerationModelFilterAll
|
||
}
|
||
}
|
||
|
||
func normalizeContentModerationModelNames(models []string) []string {
|
||
if len(models) == 0 {
|
||
return []string{}
|
||
}
|
||
out := make([]string, 0, len(models))
|
||
seen := make(map[string]struct{}, len(models))
|
||
for _, raw := range models {
|
||
model := trimRunes(strings.TrimSpace(raw), maxContentModerationModelFilterRunes)
|
||
if model == "" {
|
||
continue
|
||
}
|
||
key := strings.ToLower(model)
|
||
if _, ok := seen[key]; ok {
|
||
continue
|
||
}
|
||
seen[key] = struct{}{}
|
||
out = append(out, model)
|
||
if len(out) >= maxContentModerationModelFilterModels {
|
||
break
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func contentModerationModelListContains(models []string, model string) bool {
|
||
model = strings.ToLower(strings.TrimSpace(model))
|
||
if model == "" {
|
||
return false
|
||
}
|
||
for _, candidate := range models {
|
||
if strings.ToLower(strings.TrimSpace(candidate)) == model {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func matchBlockedKeyword(text string, keywords []string) (string, bool) {
|
||
if text == "" || len(keywords) == 0 {
|
||
return "", false
|
||
}
|
||
lower := strings.ToLower(text)
|
||
for _, kw := range keywords {
|
||
if kw == "" {
|
||
continue
|
||
}
|
||
if strings.Contains(lower, strings.ToLower(kw)) {
|
||
return kw, true
|
||
}
|
||
}
|
||
return "", false
|
||
}
|
||
|
||
func normalizeModerationAPIKeys(keys []string) []string {
|
||
if len(keys) == 0 {
|
||
return []string{}
|
||
}
|
||
seen := make(map[string]struct{}, len(keys))
|
||
out := make([]string, 0, len(keys))
|
||
for _, key := range keys {
|
||
key = strings.TrimSpace(key)
|
||
if key == "" {
|
||
continue
|
||
}
|
||
if _, ok := seen[key]; ok {
|
||
continue
|
||
}
|
||
seen[key] = struct{}{}
|
||
out = append(out, key)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func deleteModerationAPIKeysByHash(keys []string, hashes []string) []string {
|
||
keys = normalizeModerationAPIKeys(keys)
|
||
deleteHashes := make(map[string]struct{}, len(hashes))
|
||
for _, hash := range hashes {
|
||
hash = normalizeContentModerationHash(hash)
|
||
if hash != "" {
|
||
deleteHashes[hash] = struct{}{}
|
||
}
|
||
}
|
||
if len(deleteHashes) == 0 {
|
||
return keys
|
||
}
|
||
out := make([]string, 0, len(keys))
|
||
for _, key := range keys {
|
||
if _, ok := deleteHashes[moderationAPIKeyHash(key)]; ok {
|
||
continue
|
||
}
|
||
out = append(out, key)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func normalizeContentModerationAPIKeysMode(mode string) string {
|
||
switch strings.ToLower(strings.TrimSpace(mode)) {
|
||
case contentModerationAPIKeysModeReplace:
|
||
return contentModerationAPIKeysModeReplace
|
||
default:
|
||
return contentModerationAPIKeysModeAppend
|
||
}
|
||
}
|
||
|
||
func normalizeContentModerationHash(inputHash string) string {
|
||
inputHash = strings.ToLower(strings.TrimSpace(inputHash))
|
||
if len(inputHash) != sha256.Size*2 {
|
||
return ""
|
||
}
|
||
if _, err := hex.DecodeString(inputHash); err != nil {
|
||
return ""
|
||
}
|
||
return inputHash
|
||
}
|
||
|
||
func cloneFloatMap(in map[string]float64) map[string]float64 {
|
||
if in == nil {
|
||
return map[string]float64{}
|
||
}
|
||
out := make(map[string]float64, len(in))
|
||
for k, v := range in {
|
||
out[k] = v
|
||
}
|
||
return out
|
||
}
|
||
|
||
func cloneInt64Ptr(in *int64) *int64 {
|
||
if in == nil {
|
||
return nil
|
||
}
|
||
v := *in
|
||
return &v
|
||
}
|
||
|
||
func trimRunes(text string, max int) string {
|
||
if max <= 0 {
|
||
return ""
|
||
}
|
||
runes := []rune(text)
|
||
if len(runes) <= max {
|
||
return text
|
||
}
|
||
return string(runes[:max])
|
||
}
|
||
|
||
func maskSecretTail(secret string) string {
|
||
secret = strings.TrimSpace(secret)
|
||
if secret == "" {
|
||
return ""
|
||
}
|
||
if len(secret) <= 4 {
|
||
return "****"
|
||
}
|
||
return strings.Repeat("*", 8) + secret[len(secret)-4:]
|
||
}
|
||
|
||
// CyberPolicyRecordInput 是一次 cyber_policy 硬阻断的风控记录入参。
|
||
type CyberPolicyRecordInput struct {
|
||
RequestID string
|
||
UserID int64
|
||
UserEmail string
|
||
APIKeyID int64
|
||
APIKeyName string
|
||
GroupID *int64
|
||
GroupName string
|
||
Endpoint string
|
||
Model string
|
||
UpstreamMessage string
|
||
UpstreamBody string
|
||
UpstreamStatus int
|
||
UpstreamInTok int
|
||
UpstreamOutTok int
|
||
}
|
||
|
||
// RecordCyberPolicyEvent 把一次 cyber_policy 硬阻断写入风控中心日志、计入违规计数、
|
||
// 并给用户发邮件。当前请求已由 gateway 透传给用户；本方法仅做事后记录/通知/计数。
|
||
// 受 risk_control_enabled 总开关和内容审核 group/model scope 约束，
|
||
// 不受内容审核 Enabled/Mode/sample 约束。
|
||
func (s *ContentModerationService) RecordCyberPolicyEvent(ctx context.Context, in CyberPolicyRecordInput) {
|
||
if s == nil || s.repo == nil {
|
||
return
|
||
}
|
||
runtimeSnapshot, err := s.loadRuntimeSnapshot(ctx)
|
||
if err != nil {
|
||
slog.Warn("content_moderation.cyber_runtime_snapshot_load_failed", "error", err)
|
||
return
|
||
}
|
||
if !runtimeSnapshot.riskControlEnabled {
|
||
return
|
||
}
|
||
cfg := runtimeSnapshot.config
|
||
if !cfg.includesGroup(in.GroupID) || !cfg.includesModel(in.Model) {
|
||
return
|
||
}
|
||
var userID *int64
|
||
if in.UserID > 0 {
|
||
userID = &in.UserID
|
||
}
|
||
var apiKeyID *int64
|
||
if in.APIKeyID > 0 {
|
||
apiKeyID = &in.APIKeyID
|
||
}
|
||
errBody := strings.TrimSpace(in.UpstreamMessage)
|
||
if b := strings.TrimSpace(in.UpstreamBody); b != "" {
|
||
// 原始 body 不在此预脱敏；写入 log.Error 前由 redactContentModerationSecrets 统一脱敏。
|
||
errBody = strings.TrimSpace(errBody + "\n" + b)
|
||
}
|
||
if in.UpstreamInTok > 0 || in.UpstreamOutTok > 0 {
|
||
errBody = fmt.Sprintf("%s\nupstream_usage=in:%d,out:%d", errBody, in.UpstreamInTok, in.UpstreamOutTok)
|
||
}
|
||
log := &ContentModerationLog{
|
||
RequestID: in.RequestID,
|
||
UserID: userID,
|
||
UserEmail: in.UserEmail,
|
||
APIKeyID: apiKeyID,
|
||
APIKeyName: in.APIKeyName,
|
||
GroupID: cloneInt64Ptr(in.GroupID),
|
||
GroupName: in.GroupName,
|
||
Endpoint: in.Endpoint,
|
||
Provider: "openai",
|
||
Model: in.Model,
|
||
Mode: "post_upstream",
|
||
Action: ContentModerationActionCyberPolicy,
|
||
Flagged: true,
|
||
HighestCategory: "cyber_policy",
|
||
HighestScore: 1.0,
|
||
Error: trimRunes(redactContentModerationSecrets(errBody), maxModerationExcerptRunes*4),
|
||
CreatedAt: time.Now(),
|
||
}
|
||
// 开关开时 cyber_policy 不参与封号计数：当次不判定（此处跳过），
|
||
// 历史行由 CountFlaggedByUserSince 的 excludeCyberPolicy 排除。
|
||
autoBanned := false
|
||
if !cfg.CyberPolicyExcludeFromBanCount {
|
||
autoBanned = s.applyFlaggedAccountSideEffects(ctx, cfg, log)
|
||
}
|
||
log.EmailSent = false
|
||
logPersisted := true
|
||
if err := s.repo.CreateLog(ctx, log); err != nil {
|
||
logPersisted = false
|
||
slog.Warn("content_moderation.cyber_create_log_failed", "user_id", in.UserID, "error", err)
|
||
}
|
||
emailSent := false
|
||
if s.emailService != nil && strings.TrimSpace(log.UserEmail) != "" {
|
||
if err := s.sendCyberPolicyEmail(ctx, log); err != nil {
|
||
slog.Warn("content_moderation.cyber_email_failed", "user_id", in.UserID, "error", err)
|
||
} else {
|
||
emailSent = true
|
||
}
|
||
if autoBanned {
|
||
if err := s.sendAccountDisabledEmail(ctx, cfg, log); err != nil {
|
||
slog.Warn("content_moderation.cyber_ban_email_failed", "user_id", in.UserID, "error", err)
|
||
} else {
|
||
emailSent = true
|
||
}
|
||
}
|
||
}
|
||
if logPersisted && emailSent {
|
||
if err := s.repo.UpdateLogEmailSent(ctx, log.ID, true); err != nil {
|
||
slog.Warn("content_moderation.cyber_update_email_sent_failed", "log_id", log.ID, "error", err)
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *ContentModerationService) sendCyberPolicyEmail(ctx context.Context, log *ContentModerationLog) error {
|
||
siteName := s.siteName(ctx)
|
||
if s.emailService.notificationEmailService != nil {
|
||
variables := map[string]string{
|
||
"triggered_at": log.CreatedAt.UTC().Format(time.RFC3339),
|
||
"model": defaultContentModerationString(log.Model, "-"),
|
||
"group_name": defaultContentModerationString(log.GroupName, "-"),
|
||
"upstream_message": defaultContentModerationString(log.Error, "-"),
|
||
}
|
||
err := s.emailService.notificationEmailService.Send(ctx, NotificationEmailSendInput{
|
||
Event: NotificationEmailEventCyberPolicyNotice,
|
||
RecipientEmail: log.UserEmail,
|
||
RecipientName: emailRecipientName(log.UserEmail),
|
||
UserID: contentModerationEmailUserID(log),
|
||
SourceType: "content_moderation",
|
||
SourceID: contentModerationEmailSourceID(log),
|
||
Variables: variables,
|
||
})
|
||
if err == nil {
|
||
return nil
|
||
}
|
||
if !shouldFallbackNotificationEmail(err) {
|
||
return err
|
||
}
|
||
slog.Warn("template cyber policy email failed; falling back", "err", err.Error())
|
||
}
|
||
subject := fmt.Sprintf("[%s] 网络安全策略拦截 / Cyber Policy Notice", sanitizeEmailHeader(siteName))
|
||
return s.emailService.SendEmail(ctx, log.UserEmail, subject, buildCyberPolicyNoticeEmailBody(siteName, log))
|
||
}
|