package app import ( "context" "database/sql" "encoding/json" "errors" "fmt" "strings" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" goredis "github.com/redis/go-redis/v9" "go.uber.org/zap" "github.com/geo-platform/tenant-api/internal/shared/auth" "github.com/geo-platform/tenant-api/internal/shared/config" "github.com/geo-platform/tenant-api/internal/shared/response" ) type MediaSupplyService struct { pool *pgxpool.Pool redis *goredis.Client logger *zap.Logger cfg config.Provider client *MeijiequanClient } func NewMediaSupplyService(pool *pgxpool.Pool, redis *goredis.Client, logger *zap.Logger, cfg config.Provider) *MediaSupplyService { current := config.Config{} if cfg != nil && cfg.Current() != nil { current = *cfg.Current() } client := NewMeijiequanClient(redis, current.MediaSupply.Meijiequan) return &MediaSupplyService{ pool: pool, redis: redis, logger: logger, cfg: cfg, client: client, } } func (s *MediaSupplyService) mediaSupplyConfig() config.MediaSupplyConfig { if s != nil && s.cfg != nil && s.cfg.Current() != nil { cfg := s.cfg.Current().MediaSupply if s.client != nil { s.client.SetConfig(cfg.Meijiequan) } return cfg } var cfg config.MediaSupplyConfig config.NormalizeMediaSupplyConfig(&cfg) if s != nil && s.client != nil { s.client.SetConfig(cfg.Meijiequan) } return cfg } func (s *MediaSupplyService) ListResources(ctx context.Context, req ListSupplierMediaResourcesRequest) (*ListSupplierMediaResourcesResponse, error) { if s == nil || s.pool == nil { return nil, response.ErrServiceUnavailable(50360, "media_supply_store_unavailable", "媒体投稿服务暂不可用") } page, pageSize := normalizeMediaSupplyPagination(req.Page, req.PageSize) args := []any{mediaSupplySupplierMeijiequan} where := []string{"r.supplier = $1", "r.deleted_at IS NULL"} if !req.IncludeHidden { where = append(where, "r.customer_visible = TRUE") } cfg := s.mediaSupplyConfig() sellPriceSQL := mediaSupplySellPriceSQL(cfg) if req.ModelID > 0 { args = append(args, req.ModelID) where = append(where, fmt.Sprintf("r.model_id = $%d", len(args))) } if keyword := strings.TrimSpace(req.Keyword); keyword != "" { args = append(args, "%"+keyword+"%") where = append(where, fmt.Sprintf("r.name ILIKE $%d", len(args))) } if keyword := strings.TrimSpace(req.RemarkKeyword); keyword != "" { args = append(args, "%"+keyword+"%") where = append(where, fmt.Sprintf("r.resource_remark ILIKE $%d", len(args))) } if value := strings.TrimSpace(req.ChannelType); value != "" { args = append(args, value) where = append(where, fmt.Sprintf("r.channel_type = $%d", len(args))) } if value := strings.TrimSpace(req.Portal); value != "" { args = append(args, value) where = append(where, fmt.Sprintf("r.raw_json->>'zonghemenhu' = $%d", len(args))) } if value := strings.TrimSpace(req.Region); value != "" { args = append(args, value) where = append(where, fmt.Sprintf("r.region = $%d", len(args))) } if value := strings.TrimSpace(req.InclusionEffect); value != "" { args = append(args, value) where = append(where, fmt.Sprintf("r.inclusion_effect = $%d", len(args))) } if value := strings.TrimSpace(req.SpecialIndustry); value != "" { args = append(args, value) where = append(where, fmt.Sprintf("r.raw_json->>'tebiehangye' = $%d", len(args))) } if value := strings.TrimSpace(req.LinkType); value != "" { args = append(args, value) where = append(where, fmt.Sprintf("r.link_type = $%d", len(args))) } if value := strings.TrimSpace(req.DeliverySpeed); value != "" { args = append(args, value) where = append(where, fmt.Sprintf("r.delivery_speed = $%d", len(args))) } if value := strings.TrimSpace(req.AIInclude); value != "" && value != "全部" { args = append(args, value) where = append(where, fmt.Sprintf("r.raw_json->>'aiInclude' = $%d", len(args))) } if value := strings.TrimSpace(req.OtherOption); value != "" { args = append(args, value) where = append(where, fmt.Sprintf("r.raw_json->>'otherOption' = $%d", len(args))) } if req.MinPriceCents > 0 { args = append(args, req.MinPriceCents) where = append(where, fmt.Sprintf("(%s) >= $%d", sellPriceSQL, len(args))) } if req.MaxPriceCents > 0 { args = append(args, req.MaxPriceCents) where = append(where, fmt.Sprintf("(%s) <= $%d", sellPriceSQL, len(args))) } whereSQL := strings.Join(where, " AND ") orderSQL := mediaSupplyResourceOrderSQL(req.Sort, sellPriceSQL) var total int64 if err := s.pool.QueryRow(ctx, ` SELECT COUNT(*) FROM supplier_media_resources r LEFT JOIN supplier_media_price_overrides o ON o.resource_id = r.id AND o.price_type = 'price' WHERE `+whereSQL, args...).Scan(&total); err != nil { return nil, response.ErrInternal(50060, "media_supply_resource_count_failed", "媒体资源统计失败") } args = append(args, pageSize, (page-1)*pageSize) rows, err := s.pool.Query(ctx, ` SELECT r.id, r.supplier, r.supplier_resource_id, r.model_id, r.name, r.status, r.cost_price_cents, r.cost_prices_json, r.sale_price_label, r.resource_url, r.baidu_weight, r.resource_remark, r.customer_visible, r.channel_type, r.region, r.inclusion_effect, r.link_type, r.publish_rate, r.delivery_speed, r.supplier_updated_at, r.last_synced_at, r.raw_json, COALESCE(o.sell_price_cents, 0), COALESCE(o.enabled, false) FROM supplier_media_resources r LEFT JOIN supplier_media_price_overrides o ON o.resource_id = r.id AND o.price_type = 'price' WHERE `+whereSQL+` ORDER BY `+orderSQL+` LIMIT $`+fmt.Sprint(len(args)-1)+` OFFSET $`+fmt.Sprint(len(args))+` `, args...) if err != nil { return nil, response.ErrInternal(50061, "media_supply_resource_query_failed", "媒体资源读取失败") } defer rows.Close() items := make([]SupplierMediaResource, 0, pageSize) for rows.Next() { var item SupplierMediaResource var costPricesRaw []byte var raw []byte var overrideCents int64 var overrideEnabled bool if err := rows.Scan( &item.ID, &item.Supplier, &item.SupplierResourceID, &item.ModelID, &item.Name, &item.Status, &item.CostPriceCents, &costPricesRaw, &item.SalePriceLabel, &item.ResourceURL, &item.BaiduWeight, &item.ResourceRemark, &item.CustomerVisible, &item.ChannelType, &item.Region, &item.InclusionEffect, &item.LinkType, &item.PublishRate, &item.DeliverySpeed, &item.SupplierUpdatedAt, &item.LastSyncedAt, &raw, &overrideCents, &overrideEnabled, ); err != nil { return nil, response.ErrInternal(50062, "media_supply_resource_scan_failed", "媒体资源解析失败") } _ = json.Unmarshal(costPricesRaw, &item.CostPrices) if item.CostPrices == nil { item.CostPrices = map[string]int64{} } item.SellPriceCents = mediaSupplySellPrice(item.CostPriceCents, overrideCents, overrideEnabled, cfg.DefaultMarkupPercent, cfg.MinimumMarkupCents) item.Raw = json.RawMessage(raw) items = append(items, item) } return &ListSupplierMediaResourcesResponse{Items: items, Total: total, Page: page, PageSize: pageSize}, nil } func (s *MediaSupplyService) ListCustomerResources(ctx context.Context, req ListSupplierMediaResourcesRequest) (*ListCustomerSupplierMediaResourcesResponse, error) { req.IncludeHidden = false data, err := s.ListResources(ctx, req) if err != nil { return nil, err } items := make([]CustomerSupplierMediaResource, 0, len(data.Items)) for _, item := range data.Items { items = append(items, CustomerSupplierMediaResource{ ID: item.ID, SupplierResourceID: item.SupplierResourceID, ModelID: item.ModelID, Name: item.Name, Status: item.Status, SellPriceCents: item.SellPriceCents, ResourceURL: item.ResourceURL, BaiduWeight: item.BaiduWeight, ResourceRemark: item.ResourceRemark, ChannelType: item.ChannelType, Region: item.Region, InclusionEffect: item.InclusionEffect, LinkType: item.LinkType, PublishRate: item.PublishRate, DeliverySpeed: item.DeliverySpeed, LastSyncedAt: item.LastSyncedAt, }) } return &ListCustomerSupplierMediaResourcesResponse{ Items: items, Total: data.Total, Page: data.Page, PageSize: data.PageSize, }, nil } func (s *MediaSupplyService) ListCustomerResourcesByIDs(ctx context.Context, ids []int64, modelID int) (*ListCustomerSupplierMediaResourcesResponse, error) { if s == nil || s.pool == nil { return nil, response.ErrServiceUnavailable(50360, "media_supply_store_unavailable", "媒体投稿服务暂不可用") } ids = normalizeMediaSupplyResourceIDs(ids) if len(ids) == 0 { return &ListCustomerSupplierMediaResourcesResponse{ Items: []CustomerSupplierMediaResource{}, Total: 0, Page: 1, PageSize: 0, }, nil } args := []any{mediaSupplySupplierMeijiequan, ids} where := []string{ "r.supplier = $1", "r.id = ANY($2::BIGINT[])", "r.deleted_at IS NULL", "r.customer_visible = TRUE", } if modelID > 0 { args = append(args, modelID) where = append(where, fmt.Sprintf("r.model_id = $%d", len(args))) } cfg := s.mediaSupplyConfig() rows, err := s.pool.Query(ctx, ` SELECT r.id, r.supplier_resource_id, r.model_id, r.name, r.status, `+mediaSupplySellPriceSQL(cfg)+`, r.resource_url, r.baidu_weight, r.resource_remark, r.channel_type, r.region, r.inclusion_effect, r.link_type, r.publish_rate, r.delivery_speed, r.last_synced_at FROM supplier_media_resources r LEFT JOIN supplier_media_price_overrides o ON o.resource_id = r.id AND o.price_type = 'price' WHERE `+strings.Join(where, " AND ")+` ORDER BY array_position($2::BIGINT[], r.id) `, args...) if err != nil { return nil, response.ErrInternal(50061, "media_supply_resource_query_failed", "媒体资源读取失败") } defer rows.Close() items := make([]CustomerSupplierMediaResource, 0, len(ids)) for rows.Next() { var item CustomerSupplierMediaResource if err := rows.Scan( &item.ID, &item.SupplierResourceID, &item.ModelID, &item.Name, &item.Status, &item.SellPriceCents, &item.ResourceURL, &item.BaiduWeight, &item.ResourceRemark, &item.ChannelType, &item.Region, &item.InclusionEffect, &item.LinkType, &item.PublishRate, &item.DeliverySpeed, &item.LastSyncedAt, ); err != nil { return nil, response.ErrInternal(50062, "media_supply_resource_scan_failed", "媒体资源解析失败") } items = append(items, item) } if err := rows.Err(); err != nil { return nil, response.ErrInternal(50062, "media_supply_resource_scan_failed", "媒体资源解析失败") } return &ListCustomerSupplierMediaResourcesResponse{ Items: items, Total: int64(len(items)), Page: 1, PageSize: len(items), }, nil } func (s *MediaSupplyService) QueueSync(ctx context.Context, modelID int) (int64, error) { actor := auth.MustActor(ctx) if !isTenantAdmin(actor) { return 0, response.ErrForbidden(40360, "media_supply_admin_required", "需要租户管理员权限") } if modelID <= 0 { modelID = 1 } if !isKnownMeijiequanModel(modelID) { return 0, response.ErrBadRequest(40060, "invalid_media_supply_model", "不支持的媒体资源类型") } var id int64 if err := s.pool.QueryRow(ctx, ` WITH existing AS ( SELECT id FROM media_supply_sync_jobs WHERE supplier = $1 AND status IN ($4::varchar, $5::varchar) ORDER BY created_at ASC LIMIT 1 ), inserted AS ( INSERT INTO media_supply_sync_jobs (supplier, model_id, requested_by) SELECT $1, $2, $3 WHERE NOT EXISTS (SELECT 1 FROM existing) RETURNING id ) SELECT id FROM inserted UNION ALL SELECT id FROM existing LIMIT 1 `, mediaSupplySupplierMeijiequan, modelID, actor.UserID, mediaSupplySyncStatusQueued, mediaSupplySyncStatusRunning).Scan(&id); err != nil { return 0, response.ErrInternal(50063, "media_supply_sync_queue_failed", "媒体资源同步排队失败") } return id, nil } func (s *MediaSupplyService) SearchOptions(ctx context.Context, modelID int) (*MediaSupplySearchOptionsResponse, error) { if modelID <= 0 { modelID = 1 } if !isKnownMeijiequanModel(modelID) { return nil, response.ErrBadRequest(40060, "invalid_media_supply_model", "不支持的媒体资源类型") } _ = s.mediaSupplyConfig() data, err := s.client.SearchOptions(ctx, modelID) if err != nil { return nil, mediaSupplyClientError(err) } return data, nil } func (s *MediaSupplyService) ImportSession(ctx context.Context, req ImportMeijiequanSessionRequest) (*MeijiequanSessionStatus, error) { if !isTenantAdmin(auth.MustActor(ctx)) { return nil, response.ErrForbidden(40360, "media_supply_admin_required", "需要租户管理员权限") } _ = s.mediaSupplyConfig() status, err := s.client.ImportSession(ctx, req) if err != nil { return nil, mediaSupplyClientError(err) } return status, nil } func (s *MediaSupplyService) LoginConfiguredAccount(ctx context.Context) (*MeijiequanSessionStatus, error) { if !isTenantAdmin(auth.MustActor(ctx)) { return nil, response.ErrForbidden(40360, "media_supply_admin_required", "需要租户管理员权限") } _ = s.mediaSupplyConfig() status, err := s.client.LoginWithConfiguredAccount(ctx) if err != nil { return nil, mediaSupplyClientError(err) } return status, nil } func (s *MediaSupplyService) SessionStatus(ctx context.Context) (*MeijiequanSessionStatus, error) { if !isTenantAdmin(auth.MustActor(ctx)) { return nil, response.ErrForbidden(40360, "media_supply_admin_required", "需要租户管理员权限") } _ = s.mediaSupplyConfig() status, err := s.client.SessionStatus(ctx) if err != nil { return nil, mediaSupplyClientError(err) } return status, nil } func (s *MediaSupplyService) SetResourcePrice(ctx context.Context, resourceID int64, req SetSupplierMediaPriceRequest) error { if actor, ok := auth.ActorFromCtx(ctx); ok && !isTenantAdmin(actor) { return response.ErrForbidden(40360, "media_supply_admin_required", "需要租户管理员权限") } priceType := normalizeMediaSupplyPriceType(req.PriceType) if req.SellPriceCents < 0 { return response.ErrBadRequest(40061, "invalid_media_supply_price", "媒体售价不能为负数") } cost, err := s.resourceCostForPriceType(ctx, resourceID, priceType) if err != nil { return err } sellPriceCents := mediaSupplyRoundUpToYuan(req.SellPriceCents) if sellPriceCents < cost { return response.ErrConflict(40960, "media_supply_price_below_cost", "媒体售价不能低于成本价") } enabled := true if req.Enabled != nil { enabled = *req.Enabled } var updatedBy any if actor, ok := auth.ActorFromCtx(ctx); ok { updatedBy = actor.UserID } if _, err := s.pool.Exec(ctx, ` INSERT INTO supplier_media_price_overrides (resource_id, price_type, sell_price_cents, enabled, updated_by) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (resource_id, price_type) DO UPDATE SET sell_price_cents = EXCLUDED.sell_price_cents, enabled = EXCLUDED.enabled, updated_by = EXCLUDED.updated_by, updated_at = NOW() `, resourceID, priceType, sellPriceCents, enabled, updatedBy); err != nil { return response.ErrInternal(50064, "media_supply_price_update_failed", "媒体价格更新失败") } return nil } func (s *MediaSupplyService) SetResourceVisibility(ctx context.Context, resourceID int64, req SetSupplierMediaVisibilityRequest) error { if actor, ok := auth.ActorFromCtx(ctx); ok && !isTenantAdmin(actor) { return response.ErrForbidden(40360, "media_supply_admin_required", "需要租户管理员权限") } tag, err := s.pool.Exec(ctx, ` UPDATE supplier_media_resources SET customer_visible = $3, updated_at = NOW() WHERE id = $1 AND supplier = $2 AND deleted_at IS NULL `, resourceID, mediaSupplySupplierMeijiequan, req.CustomerVisible) if err != nil { return response.ErrInternal(50087, "media_supply_visibility_update_failed", "媒体展示状态更新失败") } if tag.RowsAffected() == 0 { return response.ErrNotFound(40460, "media_supply_resource_not_found", "媒体资源不存在或已下架") } return nil } func (s *MediaSupplyService) CreateOrder(ctx context.Context, req CreateMediaSupplyOrderRequest) (*CreateMediaSupplyOrderResponse, error) { actor := auth.MustActor(ctx) brandID, err := requireCurrentBrandID(ctx) if err != nil { return nil, err } if req.ModelID <= 0 || !isKnownMeijiequanModel(req.ModelID) { return nil, response.ErrBadRequest(40060, "invalid_media_supply_model", "不支持的媒体资源类型") } if len(req.Items) == 0 { return nil, response.ErrBadRequest(40062, "media_supply_order_items_required", "请至少选择一个媒体资源") } if len(req.Items) > 100 { return nil, response.ErrBadRequest(40063, "media_supply_order_items_too_many", "一次投稿选择的媒体资源过多") } title := strings.TrimSpace(req.Title) content := strings.TrimSpace(req.Content) if req.ArticleID != nil && *req.ArticleID > 0 { articleTitle, articleContent, loadErr := s.loadArticleContent(ctx, actor.TenantID, brandID, *req.ArticleID) if loadErr != nil { return nil, loadErr } if title == "" { title = articleTitle } if content == "" { content = articleContent } } if title == "" { return nil, response.ErrBadRequest(40064, "media_supply_order_title_required", "请输入投稿标题") } if content == "" { return nil, response.ErrBadRequest(40065, "media_supply_order_content_required", "请输入投稿内容") } tx, err := s.pool.Begin(ctx) if err != nil { return nil, response.ErrInternal(50065, "media_supply_order_begin_failed", "媒体投稿创建失败") } defer tx.Rollback(ctx) cfg := s.mediaSupplyConfig() seen := make(map[int64]string, len(req.Items)) orderItems := make([]MediaSupplyOrderItem, 0, len(req.Items)) var costTotal, sellTotal int64 for _, requested := range req.Items { if requested.ResourceID <= 0 { return nil, response.ErrBadRequest(40066, "invalid_media_supply_resource", "媒体资源参数不正确") } priceType := normalizeMediaSupplyPriceType(requested.PriceType) if existing, exists := seen[requested.ResourceID]; exists && existing == priceType { continue } seen[requested.ResourceID] = priceType resource, overrideCents, overrideEnabled, loadErr := s.loadResourceForOrder(ctx, tx, requested.ResourceID, req.ModelID, priceType) if loadErr != nil { return nil, loadErr } cost := costForResourcePriceType(resource, priceType) if cost <= 0 { return nil, response.ErrConflict(40961, "media_supply_cost_missing", "部分媒体资源成本价缺失,请联系管理员处理") } sell := mediaSupplySellPrice(cost, overrideCents, overrideEnabled, cfg.DefaultMarkupPercent, cfg.MinimumMarkupCents) if sell < cost { return nil, response.ErrConflict(40960, "media_supply_price_below_cost", "媒体售价不能低于成本价") } costTotal += cost sellTotal += sell orderItems = append(orderItems, MediaSupplyOrderItem{ ResourceID: resource.ID, SupplierResourceID: resource.SupplierResourceID, PriceType: priceType, ResourceNameSnapshot: resource.Name, LockedCostPriceCents: cost, LockedSellPriceCents: sell, }) } if len(orderItems) == 0 { return nil, response.ErrBadRequest(40062, "media_supply_order_items_required", "请至少选择一个媒体资源") } requestPayload, _ := json.Marshal(map[string]any{ "article_id": req.ArticleID, "model_id": req.ModelID, "title": title, "remark": strings.TrimSpace(req.Remark), "order_brand": strings.TrimSpace(req.OrderBrand), "items": req.Items, "extra": req.Extra, }) var orderID int64 if err := tx.QueryRow(ctx, ` INSERT INTO media_supply_orders ( tenant_id, workspace_id, brand_id, user_id, article_id, supplier, model_id, status, title, content_snapshot, remark, order_brand, cost_total_cents, sell_total_cents, wallet_debit_cents, wallet_debited_at, request_payload_json ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, NOW(), $16) RETURNING id `, actor.TenantID, actor.PrimaryWorkspaceID, brandID, actor.UserID, nullableInt64Ptr(req.ArticleID), mediaSupplySupplierMeijiequan, req.ModelID, mediaSupplyOrderStatusQueued, title, content, nullableTrimmedString(req.Remark), nullableTrimmedString(req.OrderBrand), costTotal, sellTotal, sellTotal, requestPayload, ).Scan(&orderID); err != nil { return nil, response.ErrInternal(50066, "media_supply_order_create_failed", "媒体投稿创建失败") } balanceAfter, err := s.debitMediaSupplyWallet(ctx, tx, actor.TenantID, actor.UserID, sellTotal, orderID, title, actor.UserID) if err != nil { return nil, err } for _, item := range orderItems { if _, err := tx.Exec(ctx, ` INSERT INTO media_supply_order_items ( order_id, resource_id, supplier_resource_id, price_type, resource_name_snapshot, locked_cost_price_cents, locked_sell_price_cents ) VALUES ($1, $2, $3, $4, $5, $6, $7) `, orderID, item.ResourceID, item.SupplierResourceID, item.PriceType, item.ResourceNameSnapshot, item.LockedCostPriceCents, item.LockedSellPriceCents); err != nil { return nil, response.ErrInternal(50067, "media_supply_order_item_create_failed", "媒体投稿明细创建失败") } } if err := tx.Commit(ctx); err != nil { return nil, response.ErrInternal(50068, "media_supply_order_commit_failed", "媒体投稿创建失败") } return &CreateMediaSupplyOrderResponse{ OrderID: orderID, Status: mediaSupplyOrderStatusQueued, CostTotalCents: costTotal, SellTotalCents: sellTotal, BalanceAfterCents: balanceAfter, }, nil } func (s *MediaSupplyService) WalletStatus(ctx context.Context) (*MediaSupplyWalletStatus, error) { actor := auth.MustActor(ctx) if s == nil || s.pool == nil { return nil, response.ErrServiceUnavailable(50360, "media_supply_store_unavailable", "媒体投稿服务暂不可用") } if err := s.ensureWalletRow(ctx, s.pool, actor.TenantID, actor.UserID); err != nil { return nil, response.ErrInternal(50077, "media_supply_wallet_prepare_failed", "媒体投稿余额初始化失败") } var status MediaSupplyWalletStatus var ledgerTotal sql.NullInt64 var lastLedgerAt sql.NullTime if err := s.pool.QueryRow(ctx, ` SELECT w.tenant_id, w.user_id, w.balance_cents, w.updated_at, COUNT(l.id), MAX(l.created_at) FROM media_supply_user_wallets w LEFT JOIN media_supply_wallet_ledgers l ON l.tenant_id = w.tenant_id AND l.user_id = w.user_id WHERE w.tenant_id = $1 AND w.user_id = $2 GROUP BY w.tenant_id, w.user_id, w.balance_cents, w.updated_at `, actor.TenantID, actor.UserID).Scan( &status.TenantID, &status.UserID, &status.BalanceCents, &status.UpdatedAt, &ledgerTotal, &lastLedgerAt, ); err != nil { return nil, response.ErrInternal(50078, "media_supply_wallet_lookup_failed", "媒体投稿余额读取失败") } if ledgerTotal.Valid { status.LedgerTotal = &ledgerTotal.Int64 } if lastLedgerAt.Valid { status.LastLedgerAt = &lastLedgerAt.Time } return &status, nil } func (s *MediaSupplyService) ListWalletLedgers(ctx context.Context, req ListMediaSupplyWalletLedgersRequest) (*ListMediaSupplyWalletLedgersResponse, error) { actor := auth.MustActor(ctx) page, pageSize := normalizeMediaSupplyPagination(req.Page, req.PageSize) userID := actor.UserID if req.UserID > 0 && req.UserID != actor.UserID { if !isTenantAdmin(actor) { return nil, response.ErrForbidden(40360, "media_supply_admin_required", "需要租户管理员权限") } if err := s.ensureWalletUserInTenant(ctx, actor.TenantID, req.UserID); err != nil { return nil, err } userID = req.UserID } args := []any{actor.TenantID, userID} where := []string{"l.tenant_id = $1", "l.user_id = $2"} if req.CreatedFrom != nil { args = append(args, *req.CreatedFrom) where = append(where, fmt.Sprintf("l.created_at >= $%d", len(args))) } if req.CreatedTo != nil { args = append(args, *req.CreatedTo) where = append(where, fmt.Sprintf("l.created_at < $%d", len(args))) } whereSQL := strings.Join(where, " AND ") var total int64 if err := s.pool.QueryRow(ctx, `SELECT COUNT(*) FROM media_supply_wallet_ledgers l WHERE `+whereSQL, args...).Scan(&total); err != nil { return nil, response.ErrInternal(50079, "media_supply_wallet_ledger_count_failed", "媒体投稿账单统计失败") } args = append(args, pageSize, (page-1)*pageSize) rows, err := s.pool.Query(ctx, ` SELECT l.id, l.tenant_id, l.user_id, l.order_id, o.title, l.delta_cents, l.balance_after_cents, l.reason, l.note, l.created_by, l.created_at FROM media_supply_wallet_ledgers l LEFT JOIN media_supply_orders o ON o.id = l.order_id WHERE `+whereSQL+` ORDER BY l.created_at DESC, l.id DESC LIMIT $`+fmt.Sprint(len(args)-1)+` OFFSET $`+fmt.Sprint(len(args))+` `, args...) if err != nil { return nil, response.ErrInternal(50080, "media_supply_wallet_ledger_query_failed", "媒体投稿账单读取失败") } defer rows.Close() items := make([]MediaSupplyWalletLedger, 0, pageSize) for rows.Next() { var item MediaSupplyWalletLedger var orderID sql.NullInt64 var orderTitle sql.NullString var note sql.NullString var createdBy sql.NullInt64 if err := rows.Scan( &item.ID, &item.TenantID, &item.UserID, &orderID, &orderTitle, &item.DeltaCents, &item.BalanceAfterCents, &item.Reason, ¬e, &createdBy, &item.CreatedAt, ); err != nil { return nil, response.ErrInternal(50081, "media_supply_wallet_ledger_scan_failed", "媒体投稿账单解析失败") } if orderID.Valid { item.OrderID = &orderID.Int64 } if orderTitle.Valid { item.OrderTitle = &orderTitle.String } if note.Valid { noteText := mediaSupplyPublicGenericErrorMessage(note.String) item.Note = ¬eText } if createdBy.Valid { item.CreatedBy = &createdBy.Int64 } items = append(items, item) } return &ListMediaSupplyWalletLedgersResponse{Items: items, Total: total, Page: page, PageSize: pageSize}, nil } func (s *MediaSupplyService) AdjustWallet(ctx context.Context, req AdjustMediaSupplyWalletRequest) (*MediaSupplyWalletStatus, error) { actor := auth.MustActor(ctx) if !isTenantAdmin(actor) { return nil, response.ErrForbidden(40360, "media_supply_admin_required", "需要租户管理员权限") } if req.UserID <= 0 { return nil, response.ErrBadRequest(40067, "invalid_media_supply_wallet_user", "用户参数不正确") } if req.DeltaCents == 0 { return nil, response.ErrBadRequest(40068, "invalid_media_supply_wallet_delta", "调整金额不能为 0") } if err := s.ensureWalletUserInTenant(ctx, actor.TenantID, req.UserID); err != nil { return nil, err } tx, err := s.pool.Begin(ctx) if err != nil { return nil, response.ErrInternal(50082, "media_supply_wallet_begin_failed", "媒体投稿余额调整失败") } defer tx.Rollback(ctx) balance, err := s.lockMediaSupplyWallet(ctx, tx, actor.TenantID, req.UserID) if err != nil { return nil, err } balanceAfter := balance + req.DeltaCents if balanceAfter < 0 { return nil, response.ErrConflict(40965, "media_supply_balance_insufficient", "媒体投稿余额不足") } if _, err := tx.Exec(ctx, ` UPDATE media_supply_user_wallets SET balance_cents = $3, updated_at = NOW() WHERE tenant_id = $1 AND user_id = $2 `, actor.TenantID, req.UserID, balanceAfter); err != nil { return nil, response.ErrInternal(50083, "media_supply_wallet_update_failed", "媒体投稿余额更新失败") } reason := "adjustment" if req.DeltaCents > 0 { reason = "recharge" } if _, err := tx.Exec(ctx, ` INSERT INTO media_supply_wallet_ledgers ( tenant_id, user_id, delta_cents, balance_after_cents, reason, note, created_by ) VALUES ($1, $2, $3, $4, $5, $6, $7) `, actor.TenantID, req.UserID, req.DeltaCents, balanceAfter, reason, nullableTrimmedString(req.Note), actor.UserID); err != nil { return nil, response.ErrInternal(50084, "media_supply_wallet_ledger_create_failed", "媒体投稿账单创建失败") } if err := tx.Commit(ctx); err != nil { return nil, response.ErrInternal(50085, "media_supply_wallet_commit_failed", "媒体投稿余额调整失败") } return &MediaSupplyWalletStatus{ TenantID: actor.TenantID, UserID: req.UserID, BalanceCents: balanceAfter, UpdatedAt: time.Now().UTC(), }, nil } type mediaSupplyWalletExecutor interface { Exec(context.Context, string, ...any) (pgconn.CommandTag, error) } func (s *MediaSupplyService) ensureWalletRow(ctx context.Context, q mediaSupplyWalletExecutor, tenantID, userID int64) error { _, err := q.Exec(ctx, ` INSERT INTO media_supply_user_wallets (tenant_id, user_id, balance_cents) VALUES ($1, $2, 0) ON CONFLICT (tenant_id, user_id) DO NOTHING `, tenantID, userID) return err } func (s *MediaSupplyService) ensureWalletUserInTenant(ctx context.Context, tenantID, userID int64) error { var exists bool if err := s.pool.QueryRow(ctx, ` SELECT EXISTS ( SELECT 1 FROM tenant_memberships WHERE tenant_id = $1 AND user_id = $2 AND deleted_at IS NULL ) `, tenantID, userID).Scan(&exists); err != nil { return response.ErrInternal(50086, "media_supply_wallet_user_lookup_failed", "媒体投稿余额用户校验失败") } if !exists { return response.ErrNotFound(40463, "media_supply_wallet_user_not_found", "用户不存在或不属于当前租户") } return nil } func (s *MediaSupplyService) lockMediaSupplyWallet(ctx context.Context, tx pgx.Tx, tenantID, userID int64) (int64, error) { if err := s.ensureWalletRow(ctx, tx, tenantID, userID); err != nil { return 0, response.ErrInternal(50077, "media_supply_wallet_prepare_failed", "媒体投稿余额初始化失败") } var balance int64 if err := tx.QueryRow(ctx, ` SELECT balance_cents FROM media_supply_user_wallets WHERE tenant_id = $1 AND user_id = $2 FOR UPDATE `, tenantID, userID).Scan(&balance); err != nil { return 0, response.ErrInternal(50078, "media_supply_wallet_lookup_failed", "媒体投稿余额读取失败") } return balance, nil } func (s *MediaSupplyService) debitMediaSupplyWallet(ctx context.Context, tx pgx.Tx, tenantID, userID, amountCents, orderID int64, orderTitle string, createdBy int64) (int64, error) { if amountCents <= 0 { balance, err := s.lockMediaSupplyWallet(ctx, tx, tenantID, userID) return balance, err } balance, err := s.lockMediaSupplyWallet(ctx, tx, tenantID, userID) if err != nil { return 0, err } if balance < amountCents { return 0, response.ErrConflict(40965, "media_supply_balance_insufficient", "媒体投稿余额不足") } balanceAfter := balance - amountCents if _, err := tx.Exec(ctx, ` UPDATE media_supply_user_wallets SET balance_cents = $3, updated_at = NOW() WHERE tenant_id = $1 AND user_id = $2 `, tenantID, userID, balanceAfter); err != nil { return 0, response.ErrInternal(50083, "media_supply_wallet_update_failed", "媒体投稿余额更新失败") } if _, err := tx.Exec(ctx, ` INSERT INTO media_supply_wallet_ledgers ( tenant_id, user_id, order_id, delta_cents, balance_after_cents, reason, note, created_by ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) `, tenantID, userID, orderID, -amountCents, balanceAfter, "order_debit", nullableTrimmedString(orderTitle), createdBy); err != nil { return 0, response.ErrInternal(50084, "media_supply_wallet_ledger_create_failed", "媒体投稿账单创建失败") } return balanceAfter, nil } func (s *MediaSupplyService) refundMediaSupplyWalletForOrder(ctx context.Context, tx pgx.Tx, orderID int64, note string) error { var tenantID, userID, debitCents int64 var refundedAt sql.NullTime var title string if err := tx.QueryRow(ctx, ` SELECT tenant_id, user_id, wallet_debit_cents, wallet_refunded_at, title FROM media_supply_orders WHERE id = $1 FOR UPDATE `, orderID).Scan(&tenantID, &userID, &debitCents, &refundedAt, &title); err != nil { if errors.Is(err, pgx.ErrNoRows) { return response.ErrNotFound(40462, "media_supply_order_not_found", "媒体投稿订单不存在") } return response.ErrInternal(50076, "media_supply_order_scan_failed", "媒体投稿订单解析失败") } if refundedAt.Valid || debitCents <= 0 { return nil } balance, err := s.lockMediaSupplyWallet(ctx, tx, tenantID, userID) if err != nil { return err } balanceAfter := balance + debitCents if _, err := tx.Exec(ctx, ` UPDATE media_supply_user_wallets SET balance_cents = $3, updated_at = NOW() WHERE tenant_id = $1 AND user_id = $2 `, tenantID, userID, balanceAfter); err != nil { return response.ErrInternal(50083, "media_supply_wallet_update_failed", "媒体投稿余额更新失败") } if _, err := tx.Exec(ctx, ` UPDATE media_supply_orders SET wallet_refunded_at = NOW(), updated_at = NOW() WHERE id = $1 AND wallet_refunded_at IS NULL `, orderID); err != nil { return response.ErrInternal(50066, "media_supply_order_create_failed", "媒体投稿退款状态更新失败") } ledgerNote := strings.TrimSpace(note) if ledgerNote == "" { ledgerNote = title } if _, err := tx.Exec(ctx, ` INSERT INTO media_supply_wallet_ledgers ( tenant_id, user_id, order_id, delta_cents, balance_after_cents, reason, note, created_by ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) `, tenantID, userID, orderID, debitCents, balanceAfter, "order_refund", nullableTrimmedString(ledgerNote), userID); err != nil { return response.ErrInternal(50084, "media_supply_wallet_ledger_create_failed", "媒体投稿账单创建失败") } return nil } func (s *MediaSupplyService) ListOrders(ctx context.Context, req ListMediaSupplyOrdersRequest) (*ListMediaSupplyOrdersResponse, error) { actor := auth.MustActor(ctx) page, pageSize := normalizeMediaSupplyPagination(req.Page, req.PageSize) args := []any{actor.TenantID, actor.UserID} where := []string{"tenant_id = $1", "user_id = $2", "deleted_at IS NULL"} if status := strings.TrimSpace(req.Status); status != "" { args = append(args, status) where = append(where, fmt.Sprintf("status = $%d", len(args))) } if req.ArticleID != nil && *req.ArticleID > 0 { args = append(args, *req.ArticleID) where = append(where, fmt.Sprintf("article_id = $%d", len(args))) } whereSQL := strings.Join(where, " AND ") var total int64 if err := s.pool.QueryRow(ctx, `SELECT COUNT(*) FROM media_supply_orders WHERE `+whereSQL, args...).Scan(&total); err != nil { return nil, response.ErrInternal(50069, "media_supply_order_count_failed", "媒体投稿订单统计失败") } args = append(args, pageSize, (page-1)*pageSize) rows, err := s.pool.Query(ctx, ` SELECT id, tenant_id, workspace_id, brand_id, user_id, article_id, supplier, model_id, status, title, remark, order_brand, external_order_id, external_order_code, supplier_status, cost_total_cents, sell_total_cents, wallet_debit_cents, wallet_debited_at, wallet_refunded_at, error_message, attempt_count, queued_at, submitted_at, completed_at, created_at, updated_at FROM media_supply_orders WHERE `+whereSQL+` ORDER BY created_at DESC, id DESC LIMIT $`+fmt.Sprint(len(args)-1)+` OFFSET $`+fmt.Sprint(len(args))+` `, args...) if err != nil { return nil, response.ErrInternal(50070, "media_supply_order_query_failed", "媒体投稿订单读取失败") } defer rows.Close() items := make([]MediaSupplyOrderDetail, 0, pageSize) for rows.Next() { item, scanErr := scanMediaSupplyOrder(rows) if scanErr != nil { return nil, scanErr } items = append(items, item) } if err := s.attachOrderItems(ctx, items); err != nil { return nil, err } return &ListMediaSupplyOrdersResponse{Items: items, Total: total, Page: page, PageSize: pageSize}, nil } func (s *MediaSupplyService) GetOrder(ctx context.Context, orderID int64) (*MediaSupplyOrderDetail, error) { actor := auth.MustActor(ctx) row := s.pool.QueryRow(ctx, ` SELECT id, tenant_id, workspace_id, brand_id, user_id, article_id, supplier, model_id, status, title, remark, order_brand, external_order_id, external_order_code, supplier_status, cost_total_cents, sell_total_cents, wallet_debit_cents, wallet_debited_at, wallet_refunded_at, error_message, attempt_count, queued_at, submitted_at, completed_at, created_at, updated_at FROM media_supply_orders WHERE id = $1 AND tenant_id = $2 AND user_id = $3 AND deleted_at IS NULL `, orderID, actor.TenantID, actor.UserID) item, err := scanMediaSupplyOrder(row) if err != nil { return nil, err } children, err := s.listOrderItems(ctx, orderID) if err != nil { return nil, err } item.Items = children return &item, nil } type mediaResourceSyncStats struct { Synced int RemovedPriceOverride int } func (s *MediaSupplyService) UpsertSyncedResources(ctx context.Context, resources []UpsertSupplierMediaResourceInput) (int, error) { stats, err := s.upsertSyncedResourcesWithStats(ctx, resources) return stats.Synced, err } func (s *MediaSupplyService) upsertSyncedResourcesWithStats(ctx context.Context, resources []UpsertSupplierMediaResourceInput) (mediaResourceSyncStats, error) { if len(resources) == 0 { return mediaResourceSyncStats{}, nil } tx, err := s.pool.Begin(ctx) if err != nil { return mediaResourceSyncStats{}, err } defer tx.Rollback(ctx) stats := mediaResourceSyncStats{} for _, resource := range resources { raw := compactJSON(resource.Raw) costPrices, _ := json.Marshal(resource.CostPrices) if len(costPrices) == 0 { costPrices = []byte(`{}`) } var resourceID int64 if err := tx.QueryRow(ctx, ` INSERT INTO supplier_media_resources ( supplier, supplier_resource_id, model_id, name, status, cost_price_cents, cost_prices_json, sale_price_label, resource_url, baidu_weight, resource_remark, channel_type, region, inclusion_effect, link_type, publish_rate, delivery_speed, supplier_updated_at, last_synced_at, raw_json ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, NOW(), $19) ON CONFLICT (supplier, supplier_resource_id) DO UPDATE SET model_id = EXCLUDED.model_id, name = EXCLUDED.name, status = EXCLUDED.status, cost_price_cents = EXCLUDED.cost_price_cents, cost_prices_json = EXCLUDED.cost_prices_json, sale_price_label = EXCLUDED.sale_price_label, resource_url = EXCLUDED.resource_url, baidu_weight = EXCLUDED.baidu_weight, resource_remark = EXCLUDED.resource_remark, channel_type = EXCLUDED.channel_type, region = EXCLUDED.region, inclusion_effect = EXCLUDED.inclusion_effect, link_type = EXCLUDED.link_type, publish_rate = EXCLUDED.publish_rate, delivery_speed = EXCLUDED.delivery_speed, supplier_updated_at = EXCLUDED.supplier_updated_at, last_synced_at = NOW(), raw_json = EXCLUDED.raw_json, updated_at = NOW(), deleted_at = NULL RETURNING id `, resource.Supplier, resource.SupplierResourceID, resource.ModelID, resource.Name, resource.Status, resource.CostPriceCents, costPrices, nullableTrimmedString(resource.SalePriceLabel), nullableTrimmedString(resource.ResourceURL), resource.BaiduWeight, nullableTrimmedString(resource.ResourceRemark), nullableTrimmedString(resource.ChannelType), nullableTrimmedString(resource.Region), nullableTrimmedString(resource.InclusionEffect), nullableTrimmedString(resource.LinkType), nullableTrimmedString(resource.PublishRate), nullableTrimmedString(resource.DeliverySpeed), resource.SupplierUpdatedAt, raw).Scan(&resourceID); err != nil { return stats, err } removed, err := s.deletePriceOverridesBelowCost(ctx, tx, resourceID) if err != nil { return stats, err } stats.RemovedPriceOverride += removed stats.Synced++ } if err := tx.Commit(ctx); err != nil { return stats, err } return stats, nil } func (s *MediaSupplyService) deletePriceOverridesBelowCost(ctx context.Context, q interface { Exec(context.Context, string, ...any) (pgconn.CommandTag, error) }, resourceID int64) (int, error) { if s == nil || q == nil || resourceID <= 0 { return 0, nil } tag, err := q.Exec(ctx, ` DELETE FROM supplier_media_price_overrides o USING supplier_media_resources r WHERE o.resource_id = r.id AND r.id = $1 AND r.supplier = $2 AND r.deleted_at IS NULL AND o.enabled = TRUE AND o.sell_price_cents < COALESCE( CASE WHEN o.price_type = 'price' THEN COALESCE(NULLIF(r.cost_prices_json->>'price', '')::bigint, NULLIF(r.cost_prices_json->>'sale_price', '')::bigint) ELSE NULLIF(r.cost_prices_json->>o.price_type, '')::bigint END, r.cost_price_cents ) `, resourceID, mediaSupplySupplierMeijiequan) return int(tag.RowsAffected()), err } func (s *MediaSupplyService) RunMediaResourceSyncOnce(ctx context.Context, modelID int) (map[string]any, error) { if s == nil || s.pool == nil || s.client == nil { return nil, response.ErrServiceUnavailable(50360, "media_supply_store_unavailable", "媒体投稿服务暂不可用") } cfg := s.mediaSupplyConfig() if !cfg.Enabled { return map[string]any{"synced": 0, "model_id": modelID, "disabled": true}, nil } if modelID <= 0 { modelID = 1 } if status, statusErr := s.client.SessionStatus(ctx); statusErr != nil || status == nil || !status.Available { if loginStatus, loginErr := s.client.LoginWithConfiguredAccount(ctx); loginErr != nil { return nil, mediaSupplyClientError(loginErr) } else if loginStatus == nil || !loginStatus.Available { return nil, mediaSupplyClientError(errMeijiequanSessionMissing) } } stats, err := s.syncMediaResourceModel(ctx, modelID, cfg.Meijiequan.SyncPageSize, cfg.Meijiequan.MaxSyncPages, cfg.Meijiequan.SyncPageDelay) if err != nil { return nil, mediaSupplyClientError(err) } return map[string]any{ "model_id": modelID, "synced": stats.Synced, "removed_price_overrides": stats.RemovedPriceOverride, }, nil } func (s *MediaSupplyService) syncMediaResourceModel(ctx context.Context, modelID, pageSize, maxPages int, pageDelay time.Duration) (mediaResourceSyncStats, error) { var total mediaResourceSyncStats for page := 1; page <= maxPages; page++ { if page > 1 && pageDelay > 0 { select { case <-ctx.Done(): return total, ctx.Err() case <-time.After(pageDelay): } } mediaPage, err := s.client.ListMedia(ctx, modelID, page, pageSize) if err != nil { return total, err } stats, err := s.upsertSyncedResourcesWithStats(ctx, mediaPage.Items) if err != nil { return total, err } total.Synced += stats.Synced total.RemovedPriceOverride += stats.RemovedPriceOverride if len(mediaPage.Items) == 0 { break } if mediaPage.AllPage > 0 && page >= mediaPage.AllPage { break } } return total, nil } func (s *MediaSupplyService) resourceCostForPriceType(ctx context.Context, resourceID int64, priceType string) (int64, error) { var costPrice int64 var costPricesRaw []byte if err := s.pool.QueryRow(ctx, ` SELECT cost_price_cents, cost_prices_json FROM supplier_media_resources WHERE id = $1 AND supplier = $2 AND deleted_at IS NULL `, resourceID, mediaSupplySupplierMeijiequan).Scan(&costPrice, &costPricesRaw); err != nil { if errors.Is(err, pgx.ErrNoRows) { return 0, response.ErrNotFound(40460, "media_supply_resource_not_found", "媒体资源不存在或已下架") } return 0, response.ErrInternal(50071, "media_supply_resource_lookup_failed", "媒体资源读取失败") } var costPrices map[string]int64 _ = json.Unmarshal(costPricesRaw, &costPrices) if priceTypeCost := costPrices[priceType]; priceTypeCost > 0 { return priceTypeCost, nil } return costPrice, nil } func (s *MediaSupplyService) loadArticleContent(ctx context.Context, tenantID, brandID, articleID int64) (string, string, error) { var title sql.NullString var html sql.NullString var markdown sql.NullString if err := s.pool.QueryRow(ctx, ` SELECT av.title, av.html_content, av.markdown_content FROM articles a JOIN article_versions av ON av.id = a.current_version_id WHERE a.id = $1 AND a.tenant_id = $2 AND a.brand_id = $3 AND a.deleted_at IS NULL `, articleID, tenantID, brandID).Scan(&title, &html, &markdown); err != nil { if errors.Is(err, pgx.ErrNoRows) { return "", "", response.ErrNotFound(40461, "article_not_found", "文章不存在或已删除") } return "", "", response.ErrInternal(50072, "article_lookup_failed", "文章读取失败") } content := strings.TrimSpace(markdown.String) if content == "" { content = strings.TrimSpace(html.String) } return strings.TrimSpace(title.String), content, nil } func (s *MediaSupplyService) loadResourceForOrder(ctx context.Context, q interface { QueryRow(context.Context, string, ...any) pgx.Row }, resourceID int64, modelID int, priceType string) (SupplierMediaResource, int64, bool, error) { var item SupplierMediaResource var costPricesRaw []byte var overrideCents int64 var overrideEnabled bool if err := q.QueryRow(ctx, ` SELECT r.id, r.supplier, r.supplier_resource_id, r.model_id, r.name, r.status, r.cost_price_cents, r.cost_prices_json, COALESCE(o.sell_price_cents, 0), COALESCE(o.enabled, false) FROM supplier_media_resources r LEFT JOIN supplier_media_price_overrides o ON o.resource_id = r.id AND o.price_type = $4 WHERE r.id = $1 AND r.supplier = $2 AND r.model_id = $3 AND r.deleted_at IS NULL AND r.customer_visible = TRUE FOR UPDATE OF r `, resourceID, mediaSupplySupplierMeijiequan, modelID, priceType).Scan( &item.ID, &item.Supplier, &item.SupplierResourceID, &item.ModelID, &item.Name, &item.Status, &item.CostPriceCents, &costPricesRaw, &overrideCents, &overrideEnabled, ); err != nil { if errors.Is(err, pgx.ErrNoRows) { return item, 0, false, response.ErrNotFound(40460, "media_supply_resource_not_found", "媒体资源不存在或已下架") } return item, 0, false, response.ErrInternal(50071, "media_supply_resource_lookup_failed", "媒体资源读取失败") } _ = json.Unmarshal(costPricesRaw, &item.CostPrices) if item.CostPrices == nil { item.CostPrices = map[string]int64{} } return item, overrideCents, overrideEnabled, nil } func (s *MediaSupplyService) listOrderItems(ctx context.Context, orderID int64) ([]MediaSupplyOrderItem, error) { rows, err := s.pool.Query(ctx, ` SELECT id, order_id, resource_id, supplier_resource_id, price_type, resource_name_snapshot, locked_cost_price_cents, locked_sell_price_cents, status, external_article_url FROM media_supply_order_items WHERE order_id = $1 ORDER BY id ASC `, orderID) if err != nil { return nil, response.ErrInternal(50073, "media_supply_order_item_query_failed", "媒体投稿明细读取失败") } defer rows.Close() items := make([]MediaSupplyOrderItem, 0) for rows.Next() { var item MediaSupplyOrderItem if err := rows.Scan( &item.ID, &item.OrderID, &item.ResourceID, &item.SupplierResourceID, &item.PriceType, &item.ResourceNameSnapshot, &item.LockedCostPriceCents, &item.LockedSellPriceCents, &item.Status, &item.ExternalArticleURL, ); err != nil { return nil, response.ErrInternal(50074, "media_supply_order_item_scan_failed", "媒体投稿明细解析失败") } if item.ExternalArticleURL != nil { if cleaned := publicMediaSupplyArticleURL(*item.ExternalArticleURL); cleaned != "" { item.ExternalArticleURL = &cleaned } else { item.ExternalArticleURL = nil } } items = append(items, item) } return items, nil } func (s *MediaSupplyService) attachOrderItems(ctx context.Context, orders []MediaSupplyOrderDetail) error { if len(orders) == 0 { return nil } orderIDs := make([]int64, 0, len(orders)) index := make(map[int64]int, len(orders)) for idx := range orders { orderIDs = append(orderIDs, orders[idx].ID) index[orders[idx].ID] = idx } rows, err := s.pool.Query(ctx, ` SELECT id, order_id, resource_id, supplier_resource_id, price_type, resource_name_snapshot, locked_cost_price_cents, locked_sell_price_cents, status, external_article_url FROM media_supply_order_items WHERE order_id = ANY($1::bigint[]) ORDER BY order_id ASC, id ASC `, orderIDs) if err != nil { return response.ErrInternal(50073, "media_supply_order_item_query_failed", "媒体投稿明细读取失败") } defer rows.Close() for rows.Next() { var item MediaSupplyOrderItem if err := rows.Scan( &item.ID, &item.OrderID, &item.ResourceID, &item.SupplierResourceID, &item.PriceType, &item.ResourceNameSnapshot, &item.LockedCostPriceCents, &item.LockedSellPriceCents, &item.Status, &item.ExternalArticleURL, ); err != nil { return response.ErrInternal(50074, "media_supply_order_item_scan_failed", "媒体投稿明细解析失败") } if item.ExternalArticleURL != nil { if cleaned := publicMediaSupplyArticleURL(*item.ExternalArticleURL); cleaned != "" { item.ExternalArticleURL = &cleaned } else { item.ExternalArticleURL = nil } } if idx, ok := index[item.OrderID]; ok { orders[idx].Items = append(orders[idx].Items, item) } } if err := rows.Err(); err != nil { return response.ErrInternal(50074, "media_supply_order_item_scan_failed", "媒体投稿明细解析失败") } return nil } func isKnownMeijiequanModel(modelID int) bool { for _, known := range meijiequanModelIDs { if modelID == known { return true } } return false } func isTenantAdmin(actor auth.Actor) bool { return strings.EqualFold(strings.TrimSpace(actor.Role), "tenant_admin") } func nullableInt(value int) any { if value <= 0 { return nil } return value } func nullableInt64Ptr(value *int64) any { if value == nil || *value <= 0 { return nil } return *value } func nullableTrimmedString(value string) any { value = strings.TrimSpace(value) if value == "" { return nil } return value } func costForResourcePriceType(resource SupplierMediaResource, priceType string) int64 { if resource.CostPrices != nil { if value := resource.CostPrices[priceType]; value > 0 { return value } if priceType == "price" { if value := resource.CostPrices["sale_price"]; value > 0 { return value } } } return resource.CostPriceCents } func mediaSupplyOverrideBelowCost(resource SupplierMediaResource, priceType string, overrideCents int64, overrideEnabled bool) bool { if !overrideEnabled { return false } cost := costForResourcePriceType(resource, normalizeMediaSupplyPriceType(priceType)) return cost > 0 && overrideCents < cost } func mediaSupplyClientError(err error) error { switch { case errors.Is(err, errMeijiequanCredentialsMissing): return response.ErrConflict(40964, "media_supply_credentials_required", "媒体供应商账号未配置,请联系管理员处理") case errors.Is(err, errMeijiequanSessionMissing), errors.Is(err, errMeijiequanSessionExpired), errors.Is(err, errMeijiequanAuthRequired): return response.ErrConflict(40962, "media_supply_session_required", "媒体供应商登录状态已失效,请联系管理员处理") case errors.Is(err, errMeijiequanChallengeRequired): return response.ErrConflict(40963, "media_supply_challenge_required", "媒体供应商登录验证失败,请联系管理员处理") default: return response.ErrInternal(50075, "media_supply_request_failed", mediaSupplyPublicErrorMessage(err)) } } func mediaSupplySellPriceSQL(cfg config.MediaSupplyConfig) string { floorSQL := "r.cost_price_cents" if cfg.DefaultMarkupPercent > 0 { multiplier := 1 + cfg.DefaultMarkupPercent/100 floorSQL = fmt.Sprintf("CEIL(r.cost_price_cents * %.8f)::BIGINT", multiplier) } if cfg.MinimumMarkupCents > 0 { floorSQL = fmt.Sprintf("GREATEST(%s, r.cost_price_cents + %d)", floorSQL, cfg.MinimumMarkupCents) } rawPriceSQL := fmt.Sprintf("CASE WHEN COALESCE(o.enabled, false) AND COALESCE(o.sell_price_cents, 0) >= r.cost_price_cents THEN o.sell_price_cents ELSE (%s) END", floorSQL) return fmt.Sprintf("CEIL((%s) / 100.0)::BIGINT * 100", rawPriceSQL) } func mediaSupplyResourceOrderSQL(sort string, sellPriceSQL string) string { switch strings.ToLower(strings.TrimSpace(sort)) { case "price": return fmt.Sprintf("(%s) ASC, r.id DESC", sellPriceSQL) case "weight": return "COALESCE(r.baidu_weight, 0) DESC, r.id DESC" default: return "r.updated_at DESC, r.id DESC" } } type mediaSupplyOrderScanner interface { Scan(...any) error } func scanMediaSupplyOrder(row mediaSupplyOrderScanner) (MediaSupplyOrderDetail, error) { var item MediaSupplyOrderDetail if err := row.Scan( &item.ID, &item.TenantID, &item.WorkspaceID, &item.BrandID, &item.UserID, &item.ArticleID, &item.Supplier, &item.ModelID, &item.Status, &item.Title, &item.Remark, &item.OrderBrand, &item.ExternalOrderID, &item.ExternalOrderCode, &item.SupplierStatus, &item.CostTotalCents, &item.SellTotalCents, &item.WalletDebitCents, &item.WalletDebitedAt, &item.WalletRefundedAt, &item.ErrorMessage, &item.AttemptCount, &item.QueuedAt, &item.SubmittedAt, &item.CompletedAt, &item.CreatedAt, &item.UpdatedAt, ); err != nil { if errors.Is(err, pgx.ErrNoRows) { return item, response.ErrNotFound(40462, "media_supply_order_not_found", "媒体投稿订单不存在") } return item, response.ErrInternal(50076, "media_supply_order_scan_failed", "媒体投稿订单解析失败") } item.Supplier = publicMediaSupplySupplierCode(item.Supplier) if item.ErrorMessage != nil { message := mediaSupplyPublicGenericErrorMessage(*item.ErrorMessage) item.ErrorMessage = &message } item.ExternalOrderID = nil if item.ExternalOrderCode != nil { code := strings.ToUpper(strings.TrimSpace(*item.ExternalOrderCode)) if !isLikelyMeijiequanOrderCode(code) { item.ExternalOrderCode = nil } else { item.ExternalOrderCode = &code } } return item, nil }