438 lines
14 KiB
Go
438 lines
14 KiB
Go
// 接入方只需要填自己的业务数据，其余交给 SDK：
|
||
//
|
||
// - data_date 留空自动填当天 UTC 日期（回补历史数据时显式指定）
|
||
// - company 留空自动填「瑞升」
|
||
// - role_id 留空自动补 []（不是 null）
|
||
// - 发送前做必填校验和日期格式校验，问题一次全列出来
|
||
// - JSON 编码、带 API Key 的 POST、超时控制
|
||
//
|
||
// 最小用法：
|
||
//
|
||
// client := datacenter.New(os.Getenv("STUDIO_API_KEY"))
|
||
//
|
||
// resp, err := client.Send(ctx, datacenter.Record{
|
||
// CompanyPlatform: "黄果短剧", // 必填：所属产品/项目名
|
||
// PublishedData: "01/09/2026", // 必填：视频上线时间
|
||
// ScriptName: "剧本A", // 必填：剧本名称
|
||
// DramaName: "短剧A", // 必填：短剧名称
|
||
// DramaNo: "第1集", // 必填：集数
|
||
// VideoURL: "https://cdn.x/a.mp4", // 必填：成片下载链接
|
||
// PlatformURL: "https://x.com/a", // 必填：产品内播放页链接
|
||
//
|
||
// VideoID: "12345", // 选填但建议填
|
||
// Views: 10000,
|
||
// Clicks: 8200,
|
||
// Likes: 300,
|
||
// })
|
||
//
|
||
// 一次可以传多条：client.Send(ctx, r1, r2, r3)。
|
||
package datacenter
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"log"
|
||
"net/http"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
)
|
||
|
||
const (
|
||
// DefaultEndpoint 是数据回流接口地址，Client.Endpoint 留空时用它。
|
||
DefaultEndpoint = "https://ldschool.up.railway.app/api/v1/video-performance/daily"
|
||
|
||
// DefaultCompany 是所属公司的默认值，接入方不用填。
|
||
DefaultCompany = "瑞升"
|
||
|
||
// DateLayout 是接口要求的日期格式：DD/MM/YYYY，对应 Go 的 "02/01/2006"。
|
||
// 注意和 "2006-01-02" 的区别，写成后者会被本地校验拦下。
|
||
DateLayout = "02/01/2006"
|
||
)
|
||
|
||
// Record 是一条视频在某一天的数据。一次请求可以带多条。
|
||
//
|
||
// 字段分三类：
|
||
//
|
||
// 1. SDK 自动填：DataDate、Company、RoleID，接入方留空即可。
|
||
// 2. 必填：CompanyPlatform、PublishedData、ScriptName、DramaName、DramaNo、
|
||
// VideoURL、PlatformURL。留空会在本地就报错，不会白跑一次请求。
|
||
// 3. 选填：其余字段。带 omitempty 的字段留空就整个不出现在请求体里，
|
||
// 不会用空字符串污染服务端数据。
|
||
type Record struct {
|
||
// ---- 由 SDK 自动填，接入方留空即可 ----
|
||
|
||
// DataDate 数据所属日期，格式 DD/MM/YYYY。
|
||
// 留空 = 当天 UTC 日期；要回补历史数据就显式写，如 "21/09/2026"。
|
||
DataDate string `json:"data_date"`
|
||
|
||
// Company 所属公司，留空 = 瑞升。
|
||
Company string `json:"company"`
|
||
|
||
// RoleID 角色 ID 数组，留空会发送 []（而不是 null）。
|
||
RoleID []string `json:"role_id"`
|
||
|
||
// ---- 必填 ----
|
||
|
||
// CompanyPlatform 所属项目/产品名称，如「黄果短剧」。
|
||
CompanyPlatform string `json:"company_platform"`
|
||
|
||
// PublishedData 视频首次上线时间，格式 DD/MM/YYYY。
|
||
PublishedData string `json:"published_data"`
|
||
|
||
// ScriptName 剧本名称，需与剧本 ID 对应一致。
|
||
ScriptName string `json:"script_name"`
|
||
|
||
// DramaName 短剧名称，需与集数保持一致。
|
||
DramaName string `json:"drama_name"`
|
||
|
||
// DramaNo 集数，如「第1集」。
|
||
DramaNo string `json:"drama_no"`
|
||
|
||
// VideoURL 可访问的成片下载/存储链接。
|
||
VideoURL string `json:"video_url"`
|
||
|
||
// PlatformURL 视频在产品内的播放页面链接。
|
||
PlatformURL string `json:"platform_url"`
|
||
|
||
// ---- 选填 ----
|
||
|
||
// VideoID 视频唯一标识。选填，但建议填，便于中台对齐同一条视频的历史数据。
|
||
VideoID string `json:"video_id,omitempty"`
|
||
|
||
// ScriptID 剧本唯一编号。
|
||
ScriptID string `json:"script_id,omitempty"`
|
||
|
||
// PlatformVideoID 视频在产品平台内的唯一 ID。
|
||
PlatformVideoID string `json:"platform_video_id,omitempty"`
|
||
|
||
// Views 浏览量：统计周期内的视频曝光/浏览次数。
|
||
Views int64 `json:"views"`
|
||
|
||
// Clicks 播放量：统计周期内的有效播放次数。
|
||
Clicks int64 `json:"clicks"`
|
||
|
||
// Likes 点赞量：统计周期内的累计点赞次数。
|
||
Likes int64 `json:"likes"`
|
||
|
||
// LikeRate 点赞率，字符串带百分号，如 "3.66%"。
|
||
// 计算公式：点赞量 ÷ 播放量 × 100%，保留两位小数。
|
||
LikeRate string `json:"like_rate,omitempty"`
|
||
|
||
// CompletionRate 完播率，如 "43.00%"。
|
||
// 没有单条视频数据时，可以填剧本整体的完播率。
|
||
CompletionRate string `json:"completion_rate,omitempty"`
|
||
|
||
// PaidRate 付费率，如 "0.00%"。
|
||
// 计算公式：付费用户 ÷ 播放用户 × 100%；免费短剧统一填 "0.00%"。
|
||
PaidRate string `json:"paid_rate,omitempty"`
|
||
|
||
// Comments 评论量。用指针是为了区分「没填」和「填了 0」：
|
||
// 填 datacenter.Int(0) 会真的发出 0，留 nil 则整个字段省略。
|
||
Comments *int64 `json:"comments,omitempty"`
|
||
|
||
// Shares 分享量，和 Comments 同理。
|
||
Shares *int64 `json:"shares,omitempty"`
|
||
|
||
// CompletionRateScope 完播率口径，取值 video / script / unknown。
|
||
CompletionRateScope string `json:"completion_rate_scope,omitempty"`
|
||
}
|
||
|
||
// Int 返回指向 v 的指针，用来给 Comments / Shares 填 0。
|
||
//
|
||
// Comments: datacenter.Int(0) // 发出 "comments": 0
|
||
// Comments: nil // 不发送 comments 字段
|
||
func Int(v int64) *int64 { return &v }
|
||
|
||
// request 是接口的请求体，数据都包在 records 数组里。
|
||
type request struct {
|
||
Records []Record `json:"records"`
|
||
}
|
||
|
||
// Response 是接口返回，方便直接打日志。
|
||
type Response struct {
|
||
StatusCode int // HTTP 状态码
|
||
Body []byte // 响应体原文，最多读 1MB
|
||
}
|
||
|
||
// OK 表示接口返回 200。
|
||
func (r *Response) OK() bool { return r != nil && r.StatusCode == http.StatusOK }
|
||
|
||
// ValidationError 表示本地校验没过。Problems 里逐条说明是第几条、缺什么，
|
||
// 一次列全，不用改一个跑一次。
|
||
type ValidationError struct {
|
||
Problems []string // 例：`records[0]: 缺少必填字段: script_name, video_url`
|
||
}
|
||
|
||
func (e *ValidationError) Error() string {
|
||
return "datacenter: 数据校验失败:\n " + strings.Join(e.Problems, "\n ")
|
||
}
|
||
|
||
// Client 是回流客户端。创建一次就能一直复用，多个 goroutine 同时用也没问题。
|
||
type Client struct {
|
||
APIKey string // 领取时展示的完整 Key，以 ld_ 开头；建议从环境变量读，别写进代码
|
||
Endpoint string // 接口地址，留空用 DefaultEndpoint
|
||
HTTPClient *http.Client // 留空用 30 秒超时的默认 client
|
||
|
||
// OnError 是异步上报失败时的回调，可选。不设置就打到 stderr。
|
||
// 校验失败、网络失败、队列满被丢弃，都会走这里。
|
||
// 回调在后台 goroutine 里执行，里面别做耗时操作。
|
||
OnError func(error)
|
||
|
||
// QueueSize 是异步队列能缓存的批次数，留空用 DefaultQueueSize。
|
||
// 队列满了会丢弃新数据（不阻塞主流程）并回调 OnError。
|
||
QueueSize int
|
||
|
||
// 以下为异步内部状态，接入方不用管。
|
||
mu sync.Mutex
|
||
queue chan []Record
|
||
done chan struct{}
|
||
started bool
|
||
closed bool
|
||
}
|
||
|
||
// DefaultQueueSize 是异步队列的默认长度（批次数）。
|
||
const DefaultQueueSize = 1000
|
||
|
||
// asyncTimeout 是后台单次请求的超时，比 HTTPClient 的 30 秒留一点余量。
|
||
const asyncTimeout = 60 * time.Second
|
||
|
||
// New 用 API Key 创建客户端，地址和 HTTP client 都用默认值。
|
||
func New(apiKey string) *Client {
|
||
return &Client{
|
||
APIKey: strings.TrimSpace(apiKey),
|
||
Endpoint: DefaultEndpoint,
|
||
HTTPClient: &http.Client{Timeout: 30 * time.Second},
|
||
QueueSize: DefaultQueueSize,
|
||
}
|
||
}
|
||
|
||
// Send 上报一批数据：先补齐默认值（日期/公司/角色数组），再校验，最后整体 POST。
|
||
//
|
||
// 校验不通过返回 *ValidationError（不发请求）；服务端返回非 200 时同时返回
|
||
// Response 和 error，Response 里带着状态码和响应体，方便排查。
|
||
//
|
||
// 只打印不发请求的话用 JSON()。
|
||
func (c *Client) Send(ctx context.Context, records ...Record) (*Response, error) {
|
||
if c.APIKey == "" {
|
||
return nil, fmt.Errorf("datacenter: APIKey 为空")
|
||
}
|
||
if !strings.HasPrefix(c.APIKey, "ld_") {
|
||
return nil, fmt.Errorf("datacenter: APIKey 需要是领取时展示的完整 Key，以 ld_ 开头")
|
||
}
|
||
if len(records) == 0 {
|
||
return nil, fmt.Errorf("datacenter: 没有要发送的数据")
|
||
}
|
||
|
||
// 复制一份再动，不改调用方传进来的切片。
|
||
payload := make([]Record, len(records))
|
||
copy(payload, records)
|
||
|
||
// 校验全部走一遍再返回，把问题攒一起报。
|
||
var problems []string
|
||
for i := range payload {
|
||
payload[i].fillDefaults()
|
||
if err := payload[i].validate(); err != nil {
|
||
problems = append(problems, fmt.Sprintf("records[%d]: %v", i, err))
|
||
}
|
||
}
|
||
if len(problems) > 0 {
|
||
return nil, &ValidationError{Problems: problems}
|
||
}
|
||
|
||
body, err := json.Marshal(request{Records: payload})
|
||
if err != nil {
|
||
return nil, fmt.Errorf("datacenter: 编码请求失败: %w", err)
|
||
}
|
||
|
||
endpoint := c.Endpoint
|
||
if endpoint == "" {
|
||
endpoint = DefaultEndpoint
|
||
}
|
||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("datacenter: 构造请求失败: %w", err)
|
||
}
|
||
req.Header.Set("Authorization", "Bearer "+c.APIKey)
|
||
req.Header.Set("Content-Type", "application/json")
|
||
|
||
client := c.HTTPClient
|
||
if client == nil {
|
||
client = &http.Client{Timeout: 30 * time.Second}
|
||
}
|
||
resp, err := client.Do(req)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("datacenter: 请求失败: %w", err)
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
// 只读 1MB，防止对面返回异常大的内容把内存吃掉。
|
||
respBody, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("datacenter: 读取响应失败: %w", err)
|
||
}
|
||
out := &Response{StatusCode: resp.StatusCode, Body: respBody}
|
||
if !out.OK() {
|
||
return out, fmt.Errorf("datacenter: HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody)))
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// SendAsync 异步上报一批数据，立即返回，不会阻塞调用方。
|
||
//
|
||
// 数据先进内存队列，由后台 goroutine 依次发送；同一个 client 的发送顺序
|
||
// 和调用顺序一致。适合放在主流程里（定时任务、接口 handler）用。
|
||
//
|
||
// 三点必须知道：
|
||
//
|
||
// 1. 失败不会返回给调用方，通过 c.OnError 回调（不设置就打到 stderr）。
|
||
// 2. 队列满了会丢弃新数据并回调，宁丢不堵——主流程优先。
|
||
// 3. 进程退出前必须调 c.Close() 等数据发完，否则队列里的会丢。
|
||
//
|
||
// 需要拿到结果、或者只是短脚本发一次就退出，直接用同步的 Send。
|
||
func (c *Client) SendAsync(records ...Record) {
|
||
if len(records) == 0 {
|
||
return
|
||
}
|
||
|
||
var err error
|
||
c.mu.Lock()
|
||
switch {
|
||
case c.closed:
|
||
err = fmt.Errorf("datacenter: 客户端已 Close，丢弃 %d 条数据", len(records))
|
||
default:
|
||
if !c.started {
|
||
size := c.QueueSize
|
||
if size <= 0 {
|
||
size = DefaultQueueSize
|
||
}
|
||
c.queue = make(chan []Record, size)
|
||
c.done = make(chan struct{})
|
||
c.started = true
|
||
go c.worker()
|
||
}
|
||
select {
|
||
case c.queue <- records:
|
||
default:
|
||
err = fmt.Errorf("datacenter: 异步队列已满（%d 批），丢弃 %d 条数据", cap(c.queue), len(records))
|
||
}
|
||
}
|
||
c.mu.Unlock()
|
||
|
||
// 回调放在锁外，这样 OnError 里再调 SendAsync 也不会死锁。
|
||
if err != nil {
|
||
c.reportErr(err)
|
||
}
|
||
}
|
||
|
||
// worker 是后台发送协程，按队列顺序逐批发，发完一批再取下一批。
|
||
func (c *Client) worker() {
|
||
defer close(c.done)
|
||
for batch := range c.queue {
|
||
ctx, cancel := context.WithTimeout(context.Background(), asyncTimeout)
|
||
_, err := c.Send(ctx, batch...)
|
||
cancel()
|
||
if err != nil {
|
||
c.reportErr(err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// Close 停止接收新的异步数据，并等队列里剩下的发完。
|
||
//
|
||
// 进程/任务退出前一定要调（defer 或任务末尾）。没调过 SendAsync 的话是空操作，
|
||
// 重复调也安全。Close 之后还调 SendAsync 不会 panic，数据会被丢弃并回调 OnError。
|
||
func (c *Client) Close() {
|
||
c.mu.Lock()
|
||
c.closed = true
|
||
if c.started {
|
||
close(c.queue)
|
||
}
|
||
done := c.done
|
||
c.mu.Unlock()
|
||
|
||
if done != nil {
|
||
<-done // 等 worker 把队列清空
|
||
}
|
||
}
|
||
|
||
// reportErr 把异步过程中的错误交给 OnError，没设置就打到 stderr。
|
||
func (c *Client) reportErr(err error) {
|
||
if c.OnError != nil {
|
||
c.OnError(err)
|
||
return
|
||
}
|
||
log.Println(err)
|
||
}
|
||
|
||
// JSON 返回这批数据最终会发出去的 JSON（同样会补默认值），用来本地核对，
|
||
// 不发任何请求。配合 Send 使用可以做到「先看一眼，再真发」。
|
||
func JSON(records ...Record) ([]byte, error) {
|
||
payload := make([]Record, len(records))
|
||
copy(payload, records)
|
||
for i := range payload {
|
||
payload[i].fillDefaults()
|
||
}
|
||
return json.MarshalIndent(request{Records: payload}, "", " ")
|
||
}
|
||
|
||
// fillDefaults 补齐 SDK 负责的三个字段。显式填过的值不会被覆盖。
|
||
func (r *Record) fillDefaults() {
|
||
if strings.TrimSpace(r.DataDate) == "" {
|
||
r.DataDate = time.Now().UTC().Format(DateLayout)
|
||
}
|
||
if strings.TrimSpace(r.Company) == "" {
|
||
r.Company = DefaultCompany
|
||
}
|
||
if r.RoleID == nil {
|
||
// 用空数组而不是 nil，避免 JSON 里出现 "role_id": null。
|
||
r.RoleID = []string{}
|
||
}
|
||
}
|
||
|
||
// validate 检查必填字段和日期格式，返回一条汇总的错误信息。
|
||
func (r *Record) validate() error {
|
||
// 必填字段，顺序和《参数说明》表格一致。
|
||
required := []struct {
|
||
name string // 报错时显示的字段名，用 JSON 里的名字方便对照
|
||
value string
|
||
}{
|
||
{"company_platform", r.CompanyPlatform},
|
||
{"published_data", r.PublishedData},
|
||
{"script_name", r.ScriptName},
|
||
{"drama_name", r.DramaName},
|
||
{"drama_no", r.DramaNo},
|
||
{"video_url", r.VideoURL},
|
||
{"platform_url", r.PlatformURL},
|
||
}
|
||
var missing []string
|
||
for _, f := range required {
|
||
if strings.TrimSpace(f.value) == "" {
|
||
missing = append(missing, f.name)
|
||
}
|
||
}
|
||
|
||
// 从示例代码抄下来忘了改的占位符，也拦一下。
|
||
if strings.HasPrefix(r.VideoID, "REPLACE_") {
|
||
missing = append(missing, "video_id")
|
||
}
|
||
|
||
// 日期必须是 DD/MM/YYYY。写成 2026-09-21 这种会在这里被拦下。
|
||
for _, d := range []struct{ name, value string }{
|
||
{"data_date", r.DataDate},
|
||
{"published_data", r.PublishedData},
|
||
} {
|
||
if _, err := time.Parse(DateLayout, d.value); err != nil {
|
||
missing = append(missing, fmt.Sprintf("%s 日期格式应为 DD/MM/YYYY，当前为 %q", d.name, d.value))
|
||
}
|
||
}
|
||
|
||
if len(missing) == 0 {
|
||
return nil
|
||
}
|
||
return fmt.Errorf("缺少必填字段: %s", strings.Join(missing, ", "))
|
||
}
|