feat(dingtalk): deliver card images through DingTalk media ids
AI card bodies render only DingTalk-hosted media, so published image URLs are rewritten into media ids inside card content (streaming frames never block, final frame waits briefly), while the markdown fallback keeps external URLs.
This commit is contained in:
姚章浩
@@ -307,6 +307,7 @@ func main() {
|
||||
}
|
||||
cardSender = cs
|
||||
}
|
||||
cardImageMedia := buildCardImageMedia(outboundImages.publicURLPrefix, dingCfg.ClientID, dingCfg.ClientSecret, address)
|
||||
handler, err := dingtalk.NewHandlerWithOptions(
|
||||
dingtalk.TenantConfig{TenantKey: dingCfg.TenantKey, Channel: "dingtalk"},
|
||||
qaService,
|
||||
@@ -317,6 +318,7 @@ func main() {
|
||||
CardSender: cardSender,
|
||||
Directory: directoryResolver,
|
||||
OutboundRewriter: outboundImages.rewriter,
|
||||
CardImageMedia: cardImageMedia,
|
||||
SettingsRouter: &dingTalkSettingsRouter{controller: settingsRuntime.controller},
|
||||
ReasoningRouter: &dingTalkReasoningRouter{service: reasoningService},
|
||||
LanguageResolver: settingsRuntime.repository,
|
||||
|
||||
@@ -2,7 +2,10 @@ package main
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"log"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -32,6 +35,9 @@ type outboundImageRuntime struct {
|
||||
rewriter dingtalk.OutboundContentRewriter
|
||||
publisher *outboundimage.StableHashPublisher
|
||||
handler http.Handler
|
||||
// publicURLPrefix is the prefix of every published image URL, used by the
|
||||
// DingTalk card image uploader to recognise (and refetch) our own images.
|
||||
publicURLPrefix string
|
||||
}
|
||||
|
||||
func buildOutboundImageRuntime(cfg outboundImageConfig) (outboundImageRuntime, error) {
|
||||
@@ -81,9 +87,10 @@ func buildOutboundImageRuntime(cfg outboundImageConfig) (outboundImageRuntime, e
|
||||
return outboundImageRuntime{}, err
|
||||
}
|
||||
return outboundImageRuntime{
|
||||
rewriter: outboundimage.NewRewriter(fetcher, publisher, outboundImageMaxCount),
|
||||
publisher: publisher,
|
||||
handler: handler,
|
||||
rewriter: outboundimage.NewRewriter(fetcher, publisher, outboundImageMaxCount),
|
||||
publisher: publisher,
|
||||
handler: handler,
|
||||
publicURLPrefix: strings.TrimRight(cfg.PublicBaseURL, "/") + "/",
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -91,3 +98,49 @@ func buildOutboundImageRewriter(cfg outboundImageConfig) (dingtalk.OutboundConte
|
||||
runtime, err := buildOutboundImageRuntime(cfg)
|
||||
return runtime.rewriter, err
|
||||
}
|
||||
|
||||
// buildCardImageMedia wires the DingTalk media-id rewriter used for AI card
|
||||
// bodies. It returns nil when the feature is switched off or not configured, so
|
||||
// a disabled uploader simply keeps external image URLs in place.
|
||||
func buildCardImageMedia(publicImagePrefix, clientID, clientSecret, listenAddr string) dingtalk.CardImageMediaRewriter {
|
||||
switch strings.ToLower(strings.TrimSpace(os.Getenv("DINGTALK_CARD_IMAGE_MEDIA"))) {
|
||||
case "off", "0", "false", "no":
|
||||
log.Printf("DingTalk card image media ids disabled by configuration")
|
||||
return nil
|
||||
}
|
||||
resolver, err := dingtalk.NewCardImageMediaRewriter(dingtalk.CardImageMediaOptions{
|
||||
ClientID: clientID,
|
||||
ClientSecret: clientSecret,
|
||||
ImageURLPrefix: publicImagePrefix,
|
||||
FetchURLPrefix: cardImageFetchPrefix(listenAddr),
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("DingTalk card image media ids disabled: %v", err)
|
||||
return nil
|
||||
}
|
||||
if resolver == nil {
|
||||
return nil
|
||||
}
|
||||
log.Printf("DingTalk card images will be delivered through DingTalk media ids")
|
||||
return resolver
|
||||
}
|
||||
|
||||
// cardImageFetchPrefix derives the prefix used to download our own published
|
||||
// images before uploading them to DingTalk. An explicit
|
||||
// DINGTALK_CARD_IMAGE_FETCH_BASE_URL (host only, without the image route) wins;
|
||||
// otherwise the local listen port is reused so the fetch stays on loopback.
|
||||
func cardImageFetchPrefix(listenAddr string) string {
|
||||
base := strings.TrimSpace(os.Getenv("DINGTALK_CARD_IMAGE_FETCH_BASE_URL"))
|
||||
if base == "" {
|
||||
addr := strings.TrimSpace(listenAddr)
|
||||
if addr == "" {
|
||||
return ""
|
||||
}
|
||||
_, port, err := net.SplitHostPort(addr)
|
||||
if err != nil || port == "" {
|
||||
return ""
|
||||
}
|
||||
base = "http://127.0.0.1:" + port
|
||||
}
|
||||
return strings.TrimRight(base, "/") + outboundImageRoute
|
||||
}
|
||||
|
||||
@@ -0,0 +1,437 @@
|
||||
package dingtalk
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"mime/multipart"
|
||||
"net/http"
|
||||
"net/textproto"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DingTalk AI cards only render images that live in DingTalk's own media store:
|
||||
// an external HTTPS image URL shows up in a normal markdown message but is
|
||||
// dropped inside a streaming/short-card body (verified on both desktop and
|
||||
// mobile, 2026-09-17). The card content therefore rewrites our own published
|
||||
// image URLs into DingTalk media_id values obtained through the legacy
|
||||
// oapi media upload endpoint, which both clients render.
|
||||
const (
|
||||
defaultCardImageTokenURL = "https://oapi.dingtalk.com/gettoken"
|
||||
defaultCardImageUploadURL = "https://oapi.dingtalk.com/media/upload"
|
||||
defaultCardImageMediaTTL = 12 * time.Hour
|
||||
defaultCardImageWaitTimeout = 5 * time.Second
|
||||
defaultCardImageFetchTimeout = 15 * time.Second
|
||||
defaultCardImageUploadTimeout = 20 * time.Second
|
||||
defaultCardImageMaxBytes = 8 << 20
|
||||
cardImageTokenRefreshMargin = 5 * time.Minute
|
||||
maxConcurrentCardImageUploads = 3
|
||||
)
|
||||
|
||||
// cardImageStableIDRE matches the deterministic image id produced by
|
||||
// outboundimage.StableHashPublisher (UUIDv5 shaped, lowercase hex).
|
||||
var cardImageStableIDRE = regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-5[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$`)
|
||||
|
||||
// cardMarkdownImageRE captures the URL of a complete markdown image, including
|
||||
// the optional title slot. Incomplete image syntax (still streaming) is not
|
||||
// matched, so it is never rewritten.
|
||||
var cardMarkdownImageRE = regexp.MustCompile(`!\[[^\]]*\]\(([^\s)\r\n]+)(?:\s+(?:"[^"\r\n]*"|'[^'\r\n]*'|\([^)\r\n]*\)))?\)`)
|
||||
|
||||
// CardImageMediaRewriter converts card-body image references into DingTalk
|
||||
// media ids. Implementations must be safe for concurrent use and must never
|
||||
// fail a card update: on any error the original reference is kept.
|
||||
type CardImageMediaRewriter interface {
|
||||
RewriteForCard(ctx context.Context, content string, wait bool) string
|
||||
}
|
||||
|
||||
// CardImageMediaOptions configures the resolver. ClientID/ClientSecret are the
|
||||
// same Robot app credentials used for access tokens, and ImageURLPrefix is the
|
||||
// public prefix of our own published images.
|
||||
type CardImageMediaOptions struct {
|
||||
ClientID string
|
||||
ClientSecret string
|
||||
ImageURLPrefix string
|
||||
// FetchURLPrefix optionally replaces ImageURLPrefix when downloading the
|
||||
// bytes back (for example a loopback address), so the upload never depends
|
||||
// on the container being able to reach its own public host name.
|
||||
FetchURLPrefix string
|
||||
TokenURL string
|
||||
UploadURL string
|
||||
HTTPClient *http.Client
|
||||
TTL time.Duration
|
||||
WaitTimeout time.Duration
|
||||
MaxBytes int64
|
||||
}
|
||||
|
||||
type cardImageMediaEntry struct {
|
||||
mediaID string
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
type cardImageMediaJob struct {
|
||||
done chan struct{}
|
||||
mediaID string
|
||||
err error
|
||||
}
|
||||
|
||||
type cardImageMediaResolver struct {
|
||||
clientID string
|
||||
clientSecret string
|
||||
imageURLPrefix string
|
||||
fetchURLPrefix string
|
||||
tokenURL string
|
||||
uploadURL string
|
||||
client *http.Client
|
||||
ttl time.Duration
|
||||
waitTimeout time.Duration
|
||||
maxBytes int64
|
||||
sem chan struct{}
|
||||
now func() time.Time
|
||||
|
||||
mu sync.Mutex
|
||||
entries map[string]cardImageMediaEntry
|
||||
inflight map[string]*cardImageMediaJob
|
||||
token string
|
||||
tokenExpiry time.Time
|
||||
}
|
||||
|
||||
// NewCardImageMediaRewriter builds the resolver. It returns a nil rewriter and
|
||||
// no error when the feature is not configured, which keeps the caller simple.
|
||||
func NewCardImageMediaRewriter(opts CardImageMediaOptions) (CardImageMediaRewriter, error) {
|
||||
clientID := strings.TrimSpace(opts.ClientID)
|
||||
clientSecret := strings.TrimSpace(opts.ClientSecret)
|
||||
prefix := strings.TrimSpace(opts.ImageURLPrefix)
|
||||
if clientID == "" || clientSecret == "" || prefix == "" {
|
||||
return nil, nil
|
||||
}
|
||||
if !strings.HasSuffix(prefix, "/") {
|
||||
prefix += "/"
|
||||
}
|
||||
tokenURL := strings.TrimSpace(opts.TokenURL)
|
||||
if tokenURL == "" {
|
||||
tokenURL = defaultCardImageTokenURL
|
||||
}
|
||||
uploadURL := strings.TrimSpace(opts.UploadURL)
|
||||
if uploadURL == "" {
|
||||
uploadURL = defaultCardImageUploadURL
|
||||
}
|
||||
ttl := opts.TTL
|
||||
if ttl <= 0 {
|
||||
ttl = defaultCardImageMediaTTL
|
||||
}
|
||||
waitTimeout := opts.WaitTimeout
|
||||
if waitTimeout <= 0 {
|
||||
waitTimeout = defaultCardImageWaitTimeout
|
||||
}
|
||||
maxBytes := opts.MaxBytes
|
||||
if maxBytes <= 0 {
|
||||
maxBytes = defaultCardImageMaxBytes
|
||||
}
|
||||
client := opts.HTTPClient
|
||||
if client == nil {
|
||||
transport := http.DefaultTransport.(*http.Transport).Clone()
|
||||
client = &http.Client{Timeout: defaultCardImageFetchTimeout, Transport: transport}
|
||||
}
|
||||
return &cardImageMediaResolver{
|
||||
clientID: clientID,
|
||||
clientSecret: clientSecret,
|
||||
imageURLPrefix: prefix,
|
||||
fetchURLPrefix: strings.TrimSpace(opts.FetchURLPrefix),
|
||||
tokenURL: tokenURL,
|
||||
uploadURL: uploadURL,
|
||||
client: client,
|
||||
ttl: ttl,
|
||||
waitTimeout: waitTimeout,
|
||||
maxBytes: maxBytes,
|
||||
sem: make(chan struct{}, maxConcurrentCardImageUploads),
|
||||
now: time.Now,
|
||||
entries: make(map[string]cardImageMediaEntry),
|
||||
inflight: make(map[string]*cardImageMediaJob),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// RewriteForCard replaces complete markdown image references that point at our
|
||||
// own published images with a DingTalk media id. With wait=false the rewrite
|
||||
// only uses media ids that are already known and kicks off the upload in the
|
||||
// background, so streaming frames never block on DingTalk; with wait=true it
|
||||
// waits (bounded) for in-flight uploads, which is what the final card update
|
||||
// uses.
|
||||
func (r *cardImageMediaResolver) RewriteForCard(ctx context.Context, content string, wait bool) string {
|
||||
if r == nil || strings.TrimSpace(content) == "" {
|
||||
return content
|
||||
}
|
||||
var waitCtx context.Context
|
||||
if wait {
|
||||
bound := r.waitTimeout
|
||||
if bound <= 0 {
|
||||
bound = defaultCardImageWaitTimeout
|
||||
}
|
||||
var cancel context.CancelFunc
|
||||
waitCtx, cancel = context.WithTimeout(context.WithoutCancel(ctx), bound)
|
||||
defer cancel()
|
||||
}
|
||||
rewritten := cardMarkdownImageRE.ReplaceAllStringFunc(content, func(image string) string {
|
||||
match := cardMarkdownImageRE.FindStringSubmatch(image)
|
||||
if len(match) < 2 {
|
||||
return image
|
||||
}
|
||||
rawURL := strings.TrimSpace(match[1])
|
||||
if rawURL == "" || !r.isOwnImageURL(rawURL) {
|
||||
return image
|
||||
}
|
||||
if mediaID, ok := r.lookup(rawURL); ok {
|
||||
return replaceCardImageURL(image, match[1], mediaID)
|
||||
}
|
||||
job := r.startUpload(rawURL)
|
||||
if !wait {
|
||||
return image
|
||||
}
|
||||
select {
|
||||
case <-job.done:
|
||||
case <-waitCtx.Done():
|
||||
}
|
||||
if job.mediaID != "" {
|
||||
return replaceCardImageURL(image, match[1], job.mediaID)
|
||||
}
|
||||
if job.err != nil {
|
||||
log.Printf("DingTalk card image media upload failed, keeping external reference")
|
||||
}
|
||||
return image
|
||||
})
|
||||
return rewritten
|
||||
}
|
||||
|
||||
func replaceCardImageURL(image, rawURL, mediaID string) string {
|
||||
if mediaID == "" || rawURL == "" {
|
||||
return image
|
||||
}
|
||||
return strings.Replace(image, rawURL, mediaID, 1)
|
||||
}
|
||||
|
||||
func (r *cardImageMediaResolver) isOwnImageURL(rawURL string) bool {
|
||||
if r.imageURLPrefix == "" || !strings.HasPrefix(rawURL, r.imageURLPrefix) {
|
||||
return false
|
||||
}
|
||||
id := strings.TrimPrefix(rawURL, r.imageURLPrefix)
|
||||
if id == "" || strings.ContainsAny(id, "/?#") {
|
||||
return false
|
||||
}
|
||||
return cardImageStableIDRE.MatchString(id)
|
||||
}
|
||||
|
||||
func (r *cardImageMediaResolver) lookup(rawURL string) (string, bool) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
entry, ok := r.entries[rawURL]
|
||||
if !ok || r.now().After(entry.expiresAt) {
|
||||
return "", false
|
||||
}
|
||||
return entry.mediaID, true
|
||||
}
|
||||
|
||||
// startUpload returns a job for the image, reusing an in-flight upload or a
|
||||
// cached media id when available. The upload itself runs on a background
|
||||
// context so a streaming frame's cancellation cannot abort it half way.
|
||||
func (r *cardImageMediaResolver) startUpload(rawURL string) *cardImageMediaJob {
|
||||
r.mu.Lock()
|
||||
if entry, ok := r.entries[rawURL]; ok && r.now().Before(entry.expiresAt) {
|
||||
job := &cardImageMediaJob{done: make(chan struct{}), mediaID: entry.mediaID}
|
||||
close(job.done)
|
||||
r.mu.Unlock()
|
||||
return job
|
||||
}
|
||||
if job, ok := r.inflight[rawURL]; ok {
|
||||
r.mu.Unlock()
|
||||
return job
|
||||
}
|
||||
job := &cardImageMediaJob{done: make(chan struct{})}
|
||||
r.inflight[rawURL] = job
|
||||
r.mu.Unlock()
|
||||
|
||||
go func() {
|
||||
uploadCtx, cancel := context.WithTimeout(context.Background(), defaultCardImageUploadTimeout)
|
||||
defer cancel()
|
||||
mediaID, err := r.upload(uploadCtx, rawURL)
|
||||
job.mediaID = mediaID
|
||||
job.err = err
|
||||
r.mu.Lock()
|
||||
delete(r.inflight, rawURL)
|
||||
if err == nil && mediaID != "" {
|
||||
r.entries[rawURL] = cardImageMediaEntry{mediaID: mediaID, expiresAt: r.now().Add(r.ttl)}
|
||||
}
|
||||
r.mu.Unlock()
|
||||
close(job.done)
|
||||
}()
|
||||
return job
|
||||
}
|
||||
|
||||
func (r *cardImageMediaResolver) upload(ctx context.Context, rawURL string) (string, error) {
|
||||
select {
|
||||
case r.sem <- struct{}{}:
|
||||
case <-ctx.Done():
|
||||
return "", ctx.Err()
|
||||
}
|
||||
defer func() { <-r.sem }()
|
||||
|
||||
data, contentType, err := r.fetch(ctx, rawURL)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
token, err := r.accessToken(ctx)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return r.uploadMedia(ctx, token, data, contentType)
|
||||
}
|
||||
|
||||
func (r *cardImageMediaResolver) fetchURL(rawURL string) string {
|
||||
base := r.imageURLPrefix
|
||||
target := rawURL
|
||||
if r.fetchURLPrefix != "" {
|
||||
base = r.fetchURLPrefix
|
||||
}
|
||||
if !strings.HasSuffix(base, "/") {
|
||||
base += "/"
|
||||
}
|
||||
if strings.HasPrefix(rawURL, r.imageURLPrefix) {
|
||||
target = base + strings.TrimPrefix(rawURL, r.imageURLPrefix)
|
||||
}
|
||||
return target
|
||||
}
|
||||
|
||||
func (r *cardImageMediaResolver) fetch(ctx context.Context, rawURL string) ([]byte, string, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.fetchURL(rawURL), nil)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
resp, err := r.client.Do(req)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, "", fmt.Errorf("card image fetch: unexpected status %d", resp.StatusCode)
|
||||
}
|
||||
contentType := strings.TrimSpace(strings.Split(resp.Header.Get("Content-Type"), ";")[0])
|
||||
if !strings.HasPrefix(strings.ToLower(contentType), "image/") {
|
||||
return nil, "", fmt.Errorf("card image fetch: unexpected content type")
|
||||
}
|
||||
data, err := io.ReadAll(io.LimitReader(resp.Body, r.maxBytes+1))
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
if int64(len(data)) > r.maxBytes {
|
||||
return nil, "", errors.New("card image fetch: image exceeds upload size limit")
|
||||
}
|
||||
if len(data) == 0 {
|
||||
return nil, "", errors.New("card image fetch: empty image")
|
||||
}
|
||||
return data, contentType, nil
|
||||
}
|
||||
|
||||
func (r *cardImageMediaResolver) accessToken(ctx context.Context) (string, error) {
|
||||
r.mu.Lock()
|
||||
if r.token != "" && r.now().Before(r.tokenExpiry) {
|
||||
token := r.token
|
||||
r.mu.Unlock()
|
||||
return token, nil
|
||||
}
|
||||
r.mu.Unlock()
|
||||
|
||||
values := map[string]string{"appkey": r.clientID, "appsecret": r.clientSecret}
|
||||
payload, err := json.Marshal(values)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.tokenURL, bytes.NewReader(payload))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
resp, err := r.client.Do(req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
var parsed struct {
|
||||
ErrCode int `json:"errcode"`
|
||||
ErrMsg string `json:"errmsg"`
|
||||
AccessToken string `json:"access_token"`
|
||||
ExpiresIn int `json:"expires_in"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &parsed); err != nil {
|
||||
return "", fmt.Errorf("card image token: invalid response")
|
||||
}
|
||||
if parsed.ErrCode != 0 || strings.TrimSpace(parsed.AccessToken) == "" {
|
||||
return "", fmt.Errorf("card image token: rejected with errcode %d", parsed.ErrCode)
|
||||
}
|
||||
expiresIn := parsed.ExpiresIn
|
||||
if expiresIn <= 0 {
|
||||
expiresIn = 7200
|
||||
}
|
||||
r.mu.Lock()
|
||||
r.token = parsed.AccessToken
|
||||
r.tokenExpiry = r.now().Add(time.Duration(expiresIn)*time.Second - cardImageTokenRefreshMargin)
|
||||
r.mu.Unlock()
|
||||
return parsed.AccessToken, nil
|
||||
}
|
||||
|
||||
func (r *cardImageMediaResolver) uploadMedia(ctx context.Context, token string, data []byte, contentType string) (string, error) {
|
||||
var buffer bytes.Buffer
|
||||
writer := multipart.NewWriter(&buffer)
|
||||
header := make(textproto.MIMEHeader)
|
||||
header.Set("Content-Disposition", `form-data; name="media"; filename="card-image"`)
|
||||
if contentType == "" {
|
||||
contentType = "image/png"
|
||||
}
|
||||
header.Set("Content-Type", contentType)
|
||||
part, err := writer.CreatePart(header)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if _, err := part.Write(data); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if err := writer.Close(); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
uploadURL := r.uploadURL + "?access_token=" + token + "&type=image"
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, uploadURL, &buffer)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
req.Header.Set("Content-Type", writer.FormDataContentType())
|
||||
resp, err := r.client.Do(req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
var parsed struct {
|
||||
ErrCode int `json:"errcode"`
|
||||
ErrMsg string `json:"errmsg"`
|
||||
MediaID string `json:"media_id"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &parsed); err != nil {
|
||||
return "", errors.New("card image upload: invalid response")
|
||||
}
|
||||
if parsed.ErrCode != 0 || strings.TrimSpace(parsed.MediaID) == "" {
|
||||
return "", fmt.Errorf("card image upload: rejected with errcode %d", parsed.ErrCode)
|
||||
}
|
||||
return parsed.MediaID, nil
|
||||
}
|
||||
@@ -0,0 +1,353 @@
|
||||
package dingtalk
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
|
||||
)
|
||||
|
||||
const cardImageTestID = "5f0a1c2e-3b4d-5e6f-8a9b-0c1d2e3f4a5b"
|
||||
const cardImageTestMediaID = "@lADP-test-media-id"
|
||||
|
||||
type cardImageTestServer struct {
|
||||
server *httptest.Server
|
||||
imageBytes []byte
|
||||
oversizeBytes []byte
|
||||
uploadErrcode int
|
||||
|
||||
tokenCalls int
|
||||
uploadCalls int
|
||||
uploadedBytes int
|
||||
fetched []string
|
||||
|
||||
mu sync.Mutex
|
||||
uploadedC chan struct{}
|
||||
}
|
||||
|
||||
func newCardImageTestServer(t *testing.T) *cardImageTestServer {
|
||||
t.Helper()
|
||||
fixture := &cardImageTestServer{
|
||||
imageBytes: []byte("\x89PNG\r\n\x1a\nfixture-bytes"),
|
||||
oversizeBytes: make([]byte, 512),
|
||||
uploadedC: make(chan struct{}, 4),
|
||||
}
|
||||
mux := http.NewServeMux()
|
||||
serveImage := func(w http.ResponseWriter, r *http.Request) {
|
||||
fixture.mu.Lock()
|
||||
fixture.fetched = append(fixture.fetched, r.URL.Path)
|
||||
fixture.mu.Unlock()
|
||||
body := fixture.imageBytes
|
||||
if strings.HasSuffix(r.URL.Path, "big") {
|
||||
body = fixture.oversizeBytes
|
||||
}
|
||||
w.Header().Set("Content-Type", "image/png")
|
||||
_, _ = w.Write(body)
|
||||
}
|
||||
mux.HandleFunc("/dingtalk/i/", serveImage)
|
||||
mux.HandleFunc("/images/", serveImage)
|
||||
mux.HandleFunc("/gettoken", func(w http.ResponseWriter, r *http.Request) {
|
||||
fixture.mu.Lock()
|
||||
fixture.tokenCalls++
|
||||
fixture.mu.Unlock()
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
"errcode": 0, "errmsg": "ok", "access_token": "test-token", "expires_in": 7200,
|
||||
})
|
||||
})
|
||||
mux.HandleFunc("/media/upload", func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Query().Get("access_token") != "test-token" || r.URL.Query().Get("type") != "image" {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{"errcode": 40001, "errmsg": "bad token"})
|
||||
return
|
||||
}
|
||||
if err := r.ParseMultipartForm(1 << 20); err != nil {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
file, _, err := r.FormFile("media")
|
||||
if err == nil {
|
||||
data, _ := io.ReadAll(file)
|
||||
_ = file.Close()
|
||||
fixture.mu.Lock()
|
||||
fixture.uploadedBytes += len(data)
|
||||
fixture.mu.Unlock()
|
||||
}
|
||||
fixture.mu.Lock()
|
||||
fixture.uploadCalls++
|
||||
code := fixture.uploadErrcode
|
||||
fixture.mu.Unlock()
|
||||
select {
|
||||
case fixture.uploadedC <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
"errcode": code, "errmsg": "ok", "media_id": cardImageTestMediaID,
|
||||
})
|
||||
})
|
||||
fixture.server = httptest.NewServer(mux)
|
||||
t.Cleanup(fixture.server.Close)
|
||||
return fixture
|
||||
}
|
||||
|
||||
func (f *cardImageTestServer) resolver(t *testing.T, mutate func(*CardImageMediaOptions)) CardImageMediaRewriter {
|
||||
t.Helper()
|
||||
opts := CardImageMediaOptions{
|
||||
ClientID: "robot-client-id",
|
||||
ClientSecret: "robot-client-secret",
|
||||
ImageURLPrefix: f.server.URL + "/dingtalk/i/",
|
||||
TokenURL: f.server.URL + "/gettoken",
|
||||
UploadURL: f.server.URL + "/media/upload",
|
||||
HTTPClient: f.server.Client(),
|
||||
}
|
||||
if mutate != nil {
|
||||
mutate(&opts)
|
||||
}
|
||||
rewriter, err := NewCardImageMediaRewriter(opts)
|
||||
if err != nil {
|
||||
t.Fatalf("NewCardImageMediaRewriter: %v", err)
|
||||
}
|
||||
if rewriter == nil {
|
||||
t.Fatal("expected a configured rewriter")
|
||||
}
|
||||
return rewriter
|
||||
}
|
||||
|
||||
func (f *cardImageTestServer) counts() (token, upload int) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.tokenCalls, f.uploadCalls
|
||||
}
|
||||
|
||||
func (f *cardImageTestServer) uploadedByteCount() int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.uploadedBytes
|
||||
}
|
||||
|
||||
func (f *cardImageTestServer) imageURL(id string) string {
|
||||
return f.server.URL + "/dingtalk/i/" + id
|
||||
}
|
||||
|
||||
func (f *cardImageTestServer) waitForUpload(t *testing.T) {
|
||||
t.Helper()
|
||||
select {
|
||||
case <-f.uploadedC:
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("timed out waiting for the media upload")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardImageMediaRewritesOwnImageAndCachesMediaID(t *testing.T) {
|
||||
fixture := newCardImageTestServer(t)
|
||||
rewriter := fixture.resolver(t, nil)
|
||||
content := fmt.Sprintf("看图：", fixture.imageURL(cardImageTestID))
|
||||
|
||||
got := rewriter.RewriteForCard(context.Background(), content, true)
|
||||
want := fmt.Sprintf("看图：", cardImageTestMediaID)
|
||||
if got != want {
|
||||
t.Fatalf("got=%q want=%q", got, want)
|
||||
}
|
||||
if _, uploads := fixture.counts(); uploads != 1 {
|
||||
t.Fatalf("uploads=%d want 1", uploads)
|
||||
}
|
||||
if got := fixture.uploadedByteCount(); got != len(fixture.imageBytes) {
|
||||
t.Fatalf("uploaded bytes=%d want %d", got, len(fixture.imageBytes))
|
||||
}
|
||||
|
||||
// A second rewrite must reuse the cached media id without another upload,
|
||||
// and the token must only be fetched once.
|
||||
again := rewriter.RewriteForCard(context.Background(), content, true)
|
||||
if again != want {
|
||||
t.Fatalf("cached rewrite got=%q want=%q", again, want)
|
||||
}
|
||||
tokens, uploads := fixture.counts()
|
||||
if uploads != 1 || tokens != 1 {
|
||||
t.Fatalf("tokens=%d uploads=%d want 1/1", tokens, uploads)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardImageMediaPartialFramesNeverBlockAndPickUpLater(t *testing.T) {
|
||||
fixture := newCardImageTestServer(t)
|
||||
rewriter := fixture.resolver(t, nil)
|
||||
content := fmt.Sprintf("", fixture.imageURL(cardImageTestID))
|
||||
|
||||
// A streaming frame keeps the external URL but starts the upload.
|
||||
if got := rewriter.RewriteForCard(context.Background(), content, false); got != content {
|
||||
t.Fatalf("streaming frame must stay unchanged, got=%q", got)
|
||||
}
|
||||
fixture.waitForUpload(t)
|
||||
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for {
|
||||
got := rewriter.RewriteForCard(context.Background(), content, false)
|
||||
if strings.Contains(got, cardImageTestMediaID) {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("media id never became available, got=%q", got)
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
if _, uploads := fixture.counts(); uploads != 1 {
|
||||
t.Fatalf("uploads=%d want 1", uploads)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardImageMediaIgnoresForeignAndIncompleteReferences(t *testing.T) {
|
||||
fixture := newCardImageTestServer(t)
|
||||
rewriter := fixture.resolver(t, nil)
|
||||
content := strings.Join([]string{
|
||||
fmt.Sprintf("", cardImageTestID),
|
||||
fmt.Sprintf("", fixture.imageURL("not-a-stable-id")),
|
||||
fmt.Sprintf("![未闭合](%s", fixture.imageURL(cardImageTestID)),
|
||||
fmt.Sprintf("纯链接 %s", fixture.imageURL(cardImageTestID)),
|
||||
}, "\n")
|
||||
|
||||
if got := rewriter.RewriteForCard(context.Background(), content, true); got != content {
|
||||
t.Fatalf("content must be untouched, got=%q", got)
|
||||
}
|
||||
tokens, uploads := fixture.counts()
|
||||
if tokens != 0 || uploads != 0 {
|
||||
t.Fatalf("tokens=%d uploads=%d want 0/0", tokens, uploads)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardImageMediaFallsBackWhenUploadFails(t *testing.T) {
|
||||
fixture := newCardImageTestServer(t)
|
||||
fixture.uploadErrcode = 40035
|
||||
rewriter := fixture.resolver(t, nil)
|
||||
content := fmt.Sprintf("", fixture.imageURL(cardImageTestID))
|
||||
|
||||
if got := rewriter.RewriteForCard(context.Background(), content, true); got != content {
|
||||
t.Fatalf("failed upload must keep the external reference, got=%q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardImageMediaFallsBackForOversizeImages(t *testing.T) {
|
||||
fixture := newCardImageTestServer(t)
|
||||
rewriter := fixture.resolver(t, func(opts *CardImageMediaOptions) {
|
||||
opts.MaxBytes = 128
|
||||
})
|
||||
content := fmt.Sprintf("", fixture.imageURL(cardImageTestID))
|
||||
|
||||
if got := rewriter.RewriteForCard(context.Background(), content, true); got != content {
|
||||
t.Fatalf("oversize image must keep the external reference, got=%q", got)
|
||||
}
|
||||
if _, uploads := fixture.counts(); uploads != 0 {
|
||||
t.Fatalf("uploads=%d want 0", uploads)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardImageMediaFetchUsesLoopbackPrefix(t *testing.T) {
|
||||
fixture := newCardImageTestServer(t)
|
||||
rewriter := fixture.resolver(t, func(opts *CardImageMediaOptions) {
|
||||
opts.ImageURLPrefix = "https://weknora-api.invalid/dingtalk/i/"
|
||||
opts.FetchURLPrefix = fixture.server.URL + "/images/"
|
||||
})
|
||||
content := fmt.Sprintf("", cardImageTestID)
|
||||
|
||||
got := rewriter.RewriteForCard(context.Background(), content, true)
|
||||
if !strings.Contains(got, cardImageTestMediaID) {
|
||||
t.Fatalf("loopback fetch must still produce a media id, got=%q", got)
|
||||
}
|
||||
fixture.mu.Lock()
|
||||
defer fixture.mu.Unlock()
|
||||
if len(fixture.fetched) != 1 || !strings.HasPrefix(fixture.fetched[0], "/images/") {
|
||||
t.Fatalf("fetched=%v want one loopback path", fixture.fetched)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardImageMediaDeduplicatesConcurrentUploads(t *testing.T) {
|
||||
fixture := newCardImageTestServer(t)
|
||||
rewriter := fixture.resolver(t, nil)
|
||||
content := fmt.Sprintf("", fixture.imageURL(cardImageTestID))
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 4; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
rewriter.RewriteForCard(context.Background(), content, true)
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
if _, uploads := fixture.counts(); uploads != 1 {
|
||||
t.Fatalf("uploads=%d want 1 for concurrent callers", uploads)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardImageMediaRewriterRequiresFullConfiguration(t *testing.T) {
|
||||
for name, opts := range map[string]CardImageMediaOptions{
|
||||
"missing client id": {ClientSecret: "s", ImageURLPrefix: "https://img.example/dingtalk/i/"},
|
||||
"missing secret": {ClientID: "id", ImageURLPrefix: "https://img.example/dingtalk/i/"},
|
||||
"missing prefix": {ClientID: "id", ClientSecret: "s"},
|
||||
} {
|
||||
rewriter, err := NewCardImageMediaRewriter(opts)
|
||||
if err != nil || rewriter != nil {
|
||||
t.Fatalf("%s: rewriter=%v err=%v want nil/nil", name, rewriter, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestStreamingCardDeliverySendsMediaIDAndKeepsMarkdownExternal(t *testing.T) {
|
||||
fixture := newCardImageTestServer(t)
|
||||
rewriter := fixture.resolver(t, nil)
|
||||
answer := fmt.Sprintf("CR20V 接线如下：\n\n", fixture.imageURL(cardImageTestID))
|
||||
|
||||
card := &fakeStreamingCard{startID: "track-image-card"}
|
||||
qa := &fakeStreamingQA{chunks: []string{answer}, result: core.QAResult{Answer: answer}}
|
||||
reply := &fakeReply{}
|
||||
h := newTestHandlerWithStreaming(t, qa, reply, card)
|
||||
h.cardImageMedia = rewriter
|
||||
|
||||
if err := h.Handle(context.Background(), directTextCallback()); err != nil {
|
||||
t.Fatalf("Handle: %v", err)
|
||||
}
|
||||
if len(card.updates) == 0 {
|
||||
t.Fatal("no card update was recorded")
|
||||
}
|
||||
final := card.updates[len(card.updates)-1]
|
||||
if !final.isFinalize {
|
||||
t.Fatalf("last update must finalize the card, got %+v", final)
|
||||
}
|
||||
if !strings.Contains(final.content, cardImageTestMediaID) {
|
||||
t.Fatalf("final card body must carry the media id, got=%q", final.content)
|
||||
}
|
||||
if strings.Contains(final.content, fixture.server.URL) {
|
||||
t.Fatalf("final card body must not keep the external URL, got=%q", final.content)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFinalizeFallbackKeepsExternalImageURL(t *testing.T) {
|
||||
fixture := newCardImageTestServer(t)
|
||||
rewriter := fixture.resolver(t, nil)
|
||||
answer := fmt.Sprintf("", fixture.imageURL(cardImageTestID))
|
||||
|
||||
card := &fakeStreamingCard{startID: "track-image-fallback", updateErr: errors.New("card unavailable")}
|
||||
qa := &fakeStreamingQA{chunks: []string{answer}, result: core.QAResult{Answer: answer}}
|
||||
reply := &fakeReply{}
|
||||
h := newTestHandlerWithStreaming(t, qa, reply, card)
|
||||
h.cardImageMedia = rewriter
|
||||
|
||||
if err := h.Handle(context.Background(), directTextCallback()); err != nil {
|
||||
t.Fatalf("Handle: %v", err)
|
||||
}
|
||||
if !reply.called {
|
||||
t.Fatal("markdown fallback was not sent")
|
||||
}
|
||||
if !strings.Contains(reply.lastText, fixture.server.URL) {
|
||||
t.Fatalf("markdown fallback must keep the external URL, got=%q", reply.lastText)
|
||||
}
|
||||
if strings.Contains(reply.lastText, cardImageTestMediaID) {
|
||||
t.Fatalf("markdown fallback must not carry a media id, got=%q", reply.lastText)
|
||||
}
|
||||
}
|
||||
@@ -157,6 +157,9 @@ type HandlerOptions struct {
|
||||
CardSender FinalCardSender
|
||||
Directory UserDirectory
|
||||
OutboundRewriter OutboundContentRewriter
|
||||
// CardImageMedia rewrites our own published image URLs in card bodies into
|
||||
// DingTalk media ids; nil keeps external URLs as-is.
|
||||
CardImageMedia CardImageMediaRewriter
|
||||
SettingsRouter SettingsRouter
|
||||
ReasoningRouter ReasoningRouter
|
||||
LanguageResolver LanguagePreferenceResolver
|
||||
@@ -171,6 +174,7 @@ type Handler struct {
|
||||
card FinalCardSender
|
||||
directory UserDirectory
|
||||
outboundRewriter OutboundContentRewriter
|
||||
cardImageMedia CardImageMediaRewriter
|
||||
settingsRouter SettingsRouter
|
||||
reasoningRouter ReasoningRouter
|
||||
languageResolver LanguagePreferenceResolver
|
||||
@@ -217,6 +221,7 @@ func NewHandlerWithOptions(cfg TenantConfig, qa QAAsker, reply MarkdownSender, o
|
||||
card: opts.CardSender,
|
||||
directory: opts.Directory,
|
||||
outboundRewriter: opts.OutboundRewriter,
|
||||
cardImageMedia: opts.CardImageMedia,
|
||||
settingsRouter: opts.SettingsRouter,
|
||||
reasoningRouter: opts.ReasoningRouter,
|
||||
languageResolver: opts.LanguageResolver,
|
||||
@@ -440,7 +445,7 @@ func (h *Handler) handleNonStreaming(ctx context.Context, webhookURL string, req
|
||||
ConversationType: req.ConversationType,
|
||||
ExternalUserID: req.ExternalUserID,
|
||||
ExternalChatID: req.ExternalChatID,
|
||||
Content: content,
|
||||
Content: h.rewriteCardImages(ctx, content, true),
|
||||
}
|
||||
if outTrackID, cardErr := h.card.SendFinalCard(ctx, cardReq); cardErr == nil {
|
||||
h.recordDelivery(ctx, result, core.DeliverySent, core.DeliveryMethodCard, outTrackID, "")
|
||||
@@ -499,7 +504,7 @@ func (h *Handler) handleStreaming(ctx context.Context, webhookURL string, req co
|
||||
fullContent = decorateReasoningPartial(visible, req.ReasoningExecution)
|
||||
now := h.now()
|
||||
if lastUpdate.IsZero() || now.Sub(lastUpdate) >= h.cardUpdateInterval {
|
||||
partialContent := h.sanitizeOutboundPartial(fullContent)
|
||||
partialContent := h.sanitizeOutboundPartial(ctx, fullContent)
|
||||
if strings.TrimSpace(partialContent) == "" || partialContent == lastPartialContent {
|
||||
return
|
||||
}
|
||||
@@ -561,7 +566,7 @@ func (h *Handler) handleEventStreaming(ctx context.Context, webhookURL string, r
|
||||
return
|
||||
}
|
||||
visibleAnswer := decorateReasoningPartial(visible, req.ReasoningExecution)
|
||||
if partialContent := h.sanitizeOutboundPartial(visibleAnswer); strings.TrimSpace(partialContent) != "" {
|
||||
if partialContent := h.sanitizeOutboundPartial(ctx, visibleAnswer); strings.TrimSpace(partialContent) != "" {
|
||||
pump.Offer(partialContent)
|
||||
}
|
||||
return
|
||||
@@ -613,7 +618,13 @@ func (h *Handler) finalizeCardOrMarkdown(ctx context.Context, webhookURL string,
|
||||
// Final card convergence must still be attempted when the QA context was
|
||||
// canceled or timed out; otherwise the card can remain "processing".
|
||||
finalCtx, cancelFinal := context.WithTimeout(context.WithoutCancel(ctx), timeout)
|
||||
updateErr := streamingCard.UpdateCard(finalCtx, outTrackID, content, true, isError)
|
||||
cardContent := content
|
||||
if !isError {
|
||||
// Card bodies render DingTalk media ids only, so the images are
|
||||
// converted here while the markdown fallback keeps the external URLs.
|
||||
cardContent = h.rewriteCardImages(finalCtx, content, true)
|
||||
}
|
||||
updateErr := streamingCard.UpdateCard(finalCtx, outTrackID, cardContent, true, isError)
|
||||
cancelFinal()
|
||||
if updateErr == nil {
|
||||
if !isError {
|
||||
@@ -651,12 +662,26 @@ func (h *Handler) rewriteOutboundContent(ctx context.Context, content string) st
|
||||
return normalizeWideMarkdownTables(sanitizeDisplayAnswer(h.outboundRewriter.Rewrite(rewriteCtx, content)))
|
||||
}
|
||||
|
||||
func (h *Handler) sanitizeOutboundPartial(content string) string {
|
||||
func (h *Handler) sanitizeOutboundPartial(ctx context.Context, content string) string {
|
||||
sanitizer, ok := h.outboundRewriter.(OutboundPartialSanitizer)
|
||||
if ok {
|
||||
content = sanitizer.SanitizePartial(content)
|
||||
}
|
||||
return stableMarkdownPrefix(sanitizeDisplayAnswer(content))
|
||||
return h.rewriteCardImages(ctx, stableMarkdownPrefix(sanitizeDisplayAnswer(content)), false)
|
||||
}
|
||||
|
||||
// rewriteCardImages converts image references in card content into DingTalk
|
||||
// media ids. It is a no-op when the feature is not configured and never fails
|
||||
// the caller: unuploadable images keep their original external reference.
|
||||
func (h *Handler) rewriteCardImages(ctx context.Context, content string, wait bool) string {
|
||||
if h.cardImageMedia == nil || strings.TrimSpace(content) == "" {
|
||||
return content
|
||||
}
|
||||
rewritten := h.cardImageMedia.RewriteForCard(ctx, content, wait)
|
||||
if strings.TrimSpace(rewritten) == "" {
|
||||
return content
|
||||
}
|
||||
return rewritten
|
||||
}
|
||||
|
||||
func (h *Handler) recordDelivery(ctx context.Context, result core.QAResult, status, method, providerMessageID, failureCode string) {
|
||||
|
||||
Reference in New Issue
Block a user