feat(enterprise-site): add WordPress support and scheduled auto-publish to sites

Add WordPress as an enterprise-site CMS type alongside pbootcms, and let schedule tasks auto-publish generated articles to enterprise sites. - WordPress connection/publisher service and tests; register wordpress media platform and relax cms_type CHECK constraint - New publish_enterprise_site_targets JSONB column on schedule_tasks with array type constraint; thread targets through scheduler dispatch, prompt/KOL generation, and worker - Admin UI: enterprise-site targets in generate/schedule/publish flows, KOL package form, MediaView, and i18n strings Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-09 19:44:24 +08:00
parent 41f5623791
commit c7bad83496
30 changed files with 2946 additions and 696 deletions
@@ -0,0 +1,400 @@
+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
+}