package app import ( "context" "encoding/json" "fmt" "runtime/debug" "strconv" "strings" "sync" "time" "go.uber.org/zap" "github.com/geo-platform/tenant-api/internal/shared/auth" "github.com/geo-platform/tenant-api/internal/shared/response" ) const ( scheduleEnterpriseSiteTargetLimit = 64 scheduleAutoPublishTimeout = 15 * time.Minute scheduleEnterpriseSiteTargetTimeout = 45 * time.Second scheduleEnterpriseSitePublishParallel = 4 ) var runScheduleAutoPublishBackground = func(fn func()) { go fn() } type ScheduleEnterpriseSiteTarget struct { SiteID int64 `json:"site_id"` CategoryID string `json:"category_id"` PublishType string `json:"publish_type,omitempty"` } type ScheduleAutoPublishInput struct { TenantID int64 WorkspaceID int64 OperatorID int64 BrandID int64 ArticleID int64 GenerationTaskID int64 Title string InputParams map[string]interface{} PublishJobs *PublishJobService EnterpriseSites *EnterpriseSiteService Logger *zap.Logger } func StartScheduleAutoPublish(input ScheduleAutoPublishInput) { if !extractBool(input.InputParams, "schedule_auto_publish") { return } runScheduleAutoPublishBackground(func() { defer func() { if recovered := recover(); recovered != nil { logScheduleAutoPublishWarn(input.Logger, "schedule auto publish panic recovered", nil, zap.Any("panic", recovered), zap.ByteString("stack", debug.Stack()), zap.Int64("generation_task_id", input.GenerationTaskID), zap.Int64("article_id", input.ArticleID), ) } }() ctx, cancel := context.WithTimeout(context.Background(), scheduleAutoPublishTimeout) defer cancel() RunScheduleAutoPublish(ctx, input) }) } func RunScheduleAutoPublish(ctx context.Context, input ScheduleAutoPublishInput) { if !extractBool(input.InputParams, "schedule_auto_publish") { return } accountIDs := normalizeSchedulePublishAccountIDs(extractStringList(input.InputParams["schedule_publish_account_ids"], 64)) enterpriseTargets := extractScheduleEnterpriseSiteTargets(input.InputParams["schedule_publish_enterprise_site_targets"], scheduleEnterpriseSiteTargetLimit) if len(accountIDs) == 0 && len(enterpriseTargets) == 0 { return } if input.WorkspaceID <= 0 || input.OperatorID <= 0 || input.ArticleID <= 0 { logScheduleAutoPublishWarn(input.Logger, "schedule auto publish skipped because configuration is incomplete", nil, zap.Int64("generation_task_id", input.GenerationTaskID), zap.Int64("article_id", input.ArticleID), zap.Int64("workspace_id", input.WorkspaceID), zap.Int64("operator_id", input.OperatorID), zap.Int("account_count", len(accountIDs)), zap.Int("enterprise_site_count", len(enterpriseTargets)), ) return } brandID := input.BrandID if brandID <= 0 { brandID = int64(extractInt(input.InputParams, "brand_id")) } if brandID <= 0 { logScheduleAutoPublishWarn(input.Logger, "schedule auto publish skipped because brand context is missing", nil, zap.Int64("generation_task_id", input.GenerationTaskID), zap.Int64("article_id", input.ArticleID), ) return } publishTitle := strings.TrimSpace(input.Title) if publishTitle == "" { publishTitle = strings.TrimSpace(extractString(input.InputParams, "task_name")) } if publishTitle == "" { publishTitle = "定时生成文章" } publishCtx := auth.WithCurrentBrandID(ctx, brandID) publishCtx = auth.WithCurrentWorkspaceID(publishCtx, input.WorkspaceID) actor := auth.Actor{ TenantID: input.TenantID, UserID: input.OperatorID, PrimaryWorkspaceID: input.WorkspaceID, } publishCtx = auth.WithActor(publishCtx, actor) if len(accountIDs) > 0 { runScheduleMediaAccountAutoPublish(publishCtx, input, actor, accountIDs, publishTitle) } if len(enterpriseTargets) > 0 { runScheduleEnterpriseSiteAutoPublish(publishCtx, input, enterpriseTargets) } } func runScheduleMediaAccountAutoPublish(ctx context.Context, input ScheduleAutoPublishInput, actor auth.Actor, accountIDs []string, title string) { if input.PublishJobs == nil { logScheduleAutoPublishWarn(input.Logger, "schedule media auto publish skipped because publish job service is unavailable", nil, zap.Int64("generation_task_id", input.GenerationTaskID), zap.Int64("article_id", input.ArticleID), zap.Int("account_count", len(accountIDs)), ) return } accounts := make([]CreatePublishJobAccountRequest, 0, len(accountIDs)) for _, accountID := range accountIDs { accountID = strings.TrimSpace(accountID) if accountID == "" { continue } accounts = append(accounts, CreatePublishJobAccountRequest{AccountID: accountID}) } if len(accounts) == 0 { return } if _, err := input.PublishJobs.Create(ctx, actor, CreatePublishJobRequest{ Title: title, ContentRef: map[string]any{ "article_id": input.ArticleID, }, Accounts: accounts, }); err != nil { logScheduleAutoPublishWarn(input.Logger, "schedule media auto publish enqueue failed", err, zap.Int64("generation_task_id", input.GenerationTaskID), zap.Int64("article_id", input.ArticleID), zap.Int64("workspace_id", input.WorkspaceID), zap.Int("account_count", len(accounts)), ) } } func runScheduleEnterpriseSiteAutoPublish(ctx context.Context, input ScheduleAutoPublishInput, targets []ScheduleEnterpriseSiteTarget) { if input.EnterpriseSites == nil { logScheduleAutoPublishWarn(input.Logger, "schedule enterprise site auto publish skipped because enterprise site service is unavailable", nil, zap.Int64("generation_task_id", input.GenerationTaskID), zap.Int64("article_id", input.ArticleID), zap.Int("enterprise_site_count", len(targets)), ) return } parallelism := scheduleEnterpriseSitePublishParallel if parallelism <= 0 { parallelism = 1 } if parallelism > len(targets) { parallelism = len(targets) } sem := make(chan struct{}, parallelism) var wg sync.WaitGroup for _, target := range targets { if target.SiteID <= 0 || strings.TrimSpace(target.CategoryID) == "" { continue } select { case <-ctx.Done(): logScheduleAutoPublishWarn(input.Logger, "schedule enterprise site auto publish stopped because context is done", ctx.Err(), zap.Int64("generation_task_id", input.GenerationTaskID), zap.Int64("article_id", input.ArticleID), zap.Int64("workspace_id", input.WorkspaceID), ) wg.Wait() return case sem <- struct{}{}: } wg.Add(1) go func(target ScheduleEnterpriseSiteTarget) { defer wg.Done() defer func() { <-sem }() defer func() { if recovered := recover(); recovered != nil { logScheduleAutoPublishWarn(input.Logger, "schedule enterprise site auto publish panic recovered", nil, zap.Any("panic", recovered), zap.ByteString("stack", debug.Stack()), zap.Int64("generation_task_id", input.GenerationTaskID), zap.Int64("article_id", input.ArticleID), zap.Int64("workspace_id", input.WorkspaceID), zap.Int64("enterprise_site_id", target.SiteID), ) } }() runScheduleEnterpriseSiteTargetAutoPublish(ctx, input, target) }(target) } wg.Wait() } func runScheduleEnterpriseSiteTargetAutoPublish(ctx context.Context, input ScheduleAutoPublishInput, target ScheduleEnterpriseSiteTarget) { targetCtx, cancel := context.WithTimeout(ctx, scheduleEnterpriseSiteTargetTimeout) defer cancel() if _, err := input.EnterpriseSites.PublishArticle(targetCtx, target.SiteID, PublishEnterpriseSiteArticleRequest{ ArticleID: input.ArticleID, CategoryID: target.CategoryID, PublishType: target.PublishType, }); err != nil { logScheduleAutoPublishWarn(input.Logger, "schedule enterprise site auto publish failed", err, zap.Int64("generation_task_id", input.GenerationTaskID), zap.Int64("article_id", input.ArticleID), zap.Int64("workspace_id", input.WorkspaceID), zap.Int64("enterprise_site_id", target.SiteID), zap.String("category_id", target.CategoryID), ) } } func logScheduleAutoPublishWarn(logger *zap.Logger, message string, err error, fields ...zap.Field) { if logger == nil { return } if err != nil { fields = append(fields, zap.Error(err)) } logger.Warn(message, fields...) } func NormalizeScheduleEnterpriseSiteTargets(values []ScheduleEnterpriseSiteTarget) []ScheduleEnterpriseSiteTarget { if len(values) == 0 { return []ScheduleEnterpriseSiteTarget{} } seen := make(map[int64]struct{}, len(values)) normalized := make([]ScheduleEnterpriseSiteTarget, 0, len(values)) for _, value := range values { siteID := value.SiteID categoryID := strings.TrimSpace(value.CategoryID) if siteID <= 0 || categoryID == "" { continue } if _, exists := seen[siteID]; exists { continue } seen[siteID] = struct{}{} normalized = append(normalized, ScheduleEnterpriseSiteTarget{ SiteID: siteID, CategoryID: categoryID, PublishType: normalizeEnterprisePublishType(value.PublishType), }) } return normalized } func DecodeScheduleEnterpriseSiteTargets(raw []byte) []ScheduleEnterpriseSiteTarget { if len(raw) == 0 { return []ScheduleEnterpriseSiteTarget{} } var values []ScheduleEnterpriseSiteTarget if err := json.Unmarshal(raw, &values); err == nil { return NormalizeScheduleEnterpriseSiteTargets(values) } var loose []map[string]interface{} if err := json.Unmarshal(raw, &loose); err != nil { return []ScheduleEnterpriseSiteTarget{} } return normalizeScheduleEnterpriseSiteTargetMaps(loose, scheduleEnterpriseSiteTargetLimit) } func extractScheduleEnterpriseSiteTargets(value interface{}, limit int) []ScheduleEnterpriseSiteTarget { switch typed := value.(type) { case []ScheduleEnterpriseSiteTarget: return limitScheduleEnterpriseSiteTargets(NormalizeScheduleEnterpriseSiteTargets(typed), limit) case []map[string]interface{}: return normalizeScheduleEnterpriseSiteTargetMaps(typed, limit) case []interface{}: items := make([]map[string]interface{}, 0, len(typed)) for _, item := range typed { switch target := item.(type) { case map[string]interface{}: items = append(items, target) case ScheduleEnterpriseSiteTarget: items = append(items, map[string]interface{}{ "site_id": target.SiteID, "category_id": target.CategoryID, "publish_type": target.PublishType, }) } if limit > 0 && len(items) >= limit { break } } return normalizeScheduleEnterpriseSiteTargetMaps(items, limit) default: return []ScheduleEnterpriseSiteTarget{} } } func normalizeScheduleEnterpriseSiteTargetMaps(values []map[string]interface{}, limit int) []ScheduleEnterpriseSiteTarget { targets := make([]ScheduleEnterpriseSiteTarget, 0, len(values)) for _, value := range values { siteID := extractInt64FromTargetMap(value, "site_id", "connection_id", "id") categoryID := strings.TrimSpace(extractStringFromTargetMap(value, "category_id", "remote_id", "scode")) publishType := strings.TrimSpace(extractStringFromTargetMap(value, "publish_type", "status")) targets = append(targets, ScheduleEnterpriseSiteTarget{ SiteID: siteID, CategoryID: categoryID, PublishType: publishType, }) if limit > 0 && len(targets) >= limit { break } } return limitScheduleEnterpriseSiteTargets(NormalizeScheduleEnterpriseSiteTargets(targets), limit) } func limitScheduleEnterpriseSiteTargets(values []ScheduleEnterpriseSiteTarget, limit int) []ScheduleEnterpriseSiteTarget { if limit > 0 && len(values) > limit { return values[:limit] } return values } func extractInt64FromTargetMap(payload map[string]interface{}, keys ...string) int64 { for _, key := range keys { raw, ok := payload[key] if !ok || raw == nil { continue } switch typed := raw.(type) { case int64: return typed case int: return int64(typed) case int32: return int64(typed) case float64: return int64(typed) case json.Number: value, _ := typed.Int64() return value case string: value, _ := strconv.ParseInt(strings.TrimSpace(typed), 10, 64) return value default: text := strings.TrimSpace(fmt.Sprintf("%v", typed)) value, _ := strconv.ParseInt(text, 10, 64) return value } } return 0 } func extractStringFromTargetMap(payload map[string]interface{}, keys ...string) string { for _, key := range keys { raw, ok := payload[key] if !ok || raw == nil { continue } switch typed := raw.(type) { case string: return typed default: return fmt.Sprintf("%v", typed) } } return "" } func encodeScheduleEnterpriseSiteTargets(targets []ScheduleEnterpriseSiteTarget) ([]byte, error) { targets = NormalizeScheduleEnterpriseSiteTargets(targets) raw, err := json.Marshal(targets) if err != nil { return nil, response.ErrBadRequest(40029, "invalid_publish_enterprise_site_targets", "企业站点发布目标格式不正确") } return raw, nil }