opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(weixin): add media sync and typing support

Browse source
This commit is contained in:
Hoshina 2026-03-22 13:45:04 +08:00
parent 8fec5a75e9
commit 3e595f6398
7 changed files with 1394 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
pkg/channels/weixin/api.go
View file

@ -128,10 +128,13 @@ func (c *ApiClient) GetUpdates(ctx context.Context, req GetUpdatesReq) (*GetUpda
return &resp, nil
}
func (c *ApiClient) SendMessage(ctx context.Context, req SendMessageReq) error {
func (c *ApiClient) SendMessage(ctx context.Context, req SendMessageReq) (*SendMessageResp, error) {
req.BaseInfo = BaseInfo{ChannelVersion: "1.0.2"}
var resp SendMessageResp
return c.post(ctx, "ilink/bot/sendmessage", req, &resp)
if err := c.post(ctx, "ilink/bot/sendmessage", req, &resp); err != nil {
return nil, err
}
return &resp, nil
}
func (c *ApiClient) GetUploadUrl(ctx context.Context, req GetUploadUrlReq) (*GetUploadUrlResp, error) {
@ -144,9 +147,22 @@ func (c *ApiClient) GetUploadUrl(ctx context.Context, req GetUploadUrlReq) (*Get
return &resp, nil
}
func (c *ApiClient) SendTyping(ctx context.Context, req SendTypingReq) error {
func (c *ApiClient) GetConfig(ctx context.Context, req GetConfigReq) (*GetConfigResp, error) {
req.BaseInfo = BaseInfo{ChannelVersion: "1.0.2"}
return c.post(ctx, "ilink/bot/sendtyping", req, nil)
var resp GetConfigResp
if err := c.post(ctx, "ilink/bot/getconfig", req, &resp); err != nil {
return nil, err
}
return &resp, nil
}
func (c *ApiClient) SendTyping(ctx context.Context, req SendTypingReq) (*SendTypingResp, error) {
req.BaseInfo = BaseInfo{ChannelVersion: "1.0.2"}
var resp SendTypingResp
if err := c.post(ctx, "ilink/bot/sendtyping", req, &resp); err != nil {
return nil, err
}
return &resp, nil
}
func (c *ApiClient) GetQRCode(ctx context.Context, botType string) (*QRCodeResponse, error) {

1037
pkg/channels/weixin/media.go Normal file
View file

File diff suppressed because it is too large Load diff

226
pkg/channels/weixin/state.go Normal file
View file

@ -0,0 +1,226 @@
package weixin
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"os"
"path/filepath"
"strings"
"time"
basechannels "github.com/sipeed/picoclaw/pkg/channels"
"github.com/sipeed/picoclaw/pkg/config"
"github.com/sipeed/picoclaw/pkg/fileutil"
"github.com/sipeed/picoclaw/pkg/logger"
)
const (
weixinDefaultCDNBaseURL = "https://novac2c.cdn.weixin.qq.com/c2c"
weixinConfigCacheTTL = 24 * time.Hour
weixinConfigRetryInitial = 2 * time.Second
weixinConfigRetryMax = time.Hour
weixinSessionPauseDuration = time.Hour
weixinSessionExpiredCode = -14
)
type typingTicketCacheEntry struct {
ticket string
nextFetchAt time.Time
retryDelay time.Duration
}
type syncCursorFile struct {
GetUpdatesBuf string `json:"get_updates_buf"`
}
func picoclawHomeDir() string {
if home := os.Getenv(config.EnvHome); home != "" {
return home
}
userHome, _ := os.UserHomeDir()
return filepath.Join(userHome, ".picoclaw")
}
func buildWeixinSyncBufPath(cfg config.WeixinConfig) string {
key := "default"
token := strings.TrimSpace(cfg.Token)
if token != "" {
sum := sha256.Sum256([]byte(strings.TrimSpace(cfg.BaseURL) + "|" + token))
key = hex.EncodeToString(sum[:8])
}
return filepath.Join(picoclawHomeDir(), "channels", "weixin", "sync", key+".json")
}
func loadGetUpdatesBuf(path string) (string, error) {
data, err := os.ReadFile(path)
if err != nil {
if os.IsNotExist(err) {
return "", nil
}
return "", err
}
var decoded syncCursorFile
if err := json.Unmarshal(data, &decoded); err != nil {
return "", err
}
return decoded.GetUpdatesBuf, nil
}
func saveGetUpdatesBuf(path, cursor string) error {
data, err := json.Marshal(syncCursorFile{GetUpdatesBuf: cursor})
if err != nil {
return err
}
return fileutil.WriteFileAtomic(path, data, 0o600)
}
func (c *WeixinChannel) cdnBaseURL() string {
if base := strings.TrimSpace(c.config.CDNBaseURL); base != "" {
return strings.TrimRight(base, "/")
}
return weixinDefaultCDNBaseURL
}
func isSessionExpiredStatus(ret, errcode int) bool {
return ret == weixinSessionExpiredCode || errcode == weixinSessionExpiredCode
}
func (c *WeixinChannel) pauseSession(operation string, ret, errcode int, errmsg string) time.Duration {
c.pauseMu.Lock()
defer c.pauseMu.Unlock()
until := time.Now().Add(weixinSessionPauseDuration)
if until.After(c.pauseUntil) {
c.pauseUntil = until
}
remaining := time.Until(c.pauseUntil)
logger.ErrorCF("weixin", "Session expired; pausing Weixin channel", map[string]any{
"operation": operation,
"ret": ret,
"errcode": errcode,
"errmsg": errmsg,
"until": c.pauseUntil.Format(time.RFC3339),
"minutes": int((remaining + time.Minute - 1) / time.Minute),
})
return remaining
}
func (c *WeixinChannel) remainingPause() time.Duration {
c.pauseMu.Lock()
defer c.pauseMu.Unlock()
if c.pauseUntil.IsZero() {
return 0
}
remaining := time.Until(c.pauseUntil)
if remaining <= 0 {
c.pauseUntil = time.Time{}
return 0
}
return remaining
}
func (c *WeixinChannel) waitWhileSessionPaused(ctx context.Context) error {
remaining := c.remainingPause()
if remaining <= 0 {
return nil
}
timer := time.NewTimer(remaining)
defer timer.Stop()
select {
case <-ctx.Done():
return ctx.Err()
case <-timer.C:
return nil
}
}
func (c *WeixinChannel) ensureSessionActive() error {
remaining := c.remainingPause()
if remaining <= 0 {
return nil
}
return fmt.Errorf(
"weixin session paused (%d min remaining): %w",
int((remaining+time.Minute-1)/time.Minute),
basechannels.ErrSendFailed,
)
}
func (c *WeixinChannel) getTypingTicket(ctx context.Context, userID string) (string, error) {
now := time.Now()
c.typingMu.Lock()
entry, ok := c.typingCache[userID]
if ok && now.Before(entry.nextFetchAt) {
ticket := entry.ticket
c.typingMu.Unlock()
return ticket, nil
}
cachedTicket := entry.ticket
retryDelay := entry.retryDelay
c.typingMu.Unlock()
contextToken := ""
if v, ok := c.contextTokens.Load(userID); ok {
contextToken, _ = v.(string)
}
resp, err := c.api.GetConfig(ctx, GetConfigReq{
IlinkUserID: userID,
ContextToken: contextToken,
})
if err == nil && resp != nil && resp.Ret == 0 && resp.Errcode == 0 {
ticket := strings.TrimSpace(resp.TypingTicket)
c.typingMu.Lock()
c.typingCache[userID] = typingTicketCacheEntry{
ticket: ticket,
nextFetchAt: now.Add(weixinConfigCacheTTL),
retryDelay: weixinConfigRetryInitial,
}
c.typingMu.Unlock()
return ticket, nil
}
if resp != nil && isSessionExpiredStatus(resp.Ret, resp.Errcode) {
c.pauseSession("getconfig", resp.Ret, resp.Errcode, resp.Errmsg)
}
if retryDelay <= 0 {
retryDelay = weixinConfigRetryInitial
} else {
retryDelay *= 2
if retryDelay > weixinConfigRetryMax {
retryDelay = weixinConfigRetryMax
}
}
c.typingMu.Lock()
c.typingCache[userID] = typingTicketCacheEntry{
ticket: cachedTicket,
nextFetchAt: now.Add(retryDelay),
retryDelay: retryDelay,
}
c.typingMu.Unlock()
if err != nil {
return cachedTicket, err
}
if resp == nil {
return cachedTicket, fmt.Errorf("getconfig returned nil response")
}
return cachedTicket, fmt.Errorf(
"getconfig failed: ret=%d errcode=%d errmsg=%s",
resp.Ret,
resp.Errcode,
resp.Errmsg,
)
}

33
pkg/channels/weixin/types.go
View file

@ -5,6 +5,12 @@ type BaseInfo struct {
ChannelVersion string `json:"channel_version,omitempty"`
}
type APIStatus struct {
Ret int `json:"ret,omitempty"`
Errcode int `json:"errcode,omitempty"`
Errmsg string `json:"errmsg,omitempty"`
}
// UploadMediaType constants
const (
UploadMediaTypeImage = 1
@ -24,11 +30,12 @@ type GetUploadUrlReq struct {
ThumbRawfileMD5 string `json:"thumb_rawfilemd5,omitempty"`
ThumbFilesize int64 `json:"thumb_filesize,omitempty"`
NoNeedThumb bool `json:"no_need_thumb,omitempty"`
Aeskey string `json:"aeskey,omitempty"` // base64
Aeskey string `json:"aeskey,omitempty"` // hex-encoded 16-byte AES key
BaseInfo BaseInfo `json:"base_info,omitempty"`
}
type GetUploadUrlResp struct {
APIStatus
UploadParam string `json:"upload_param,omitempty"`
ThumbUploadParam string `json:"thumb_upload_param,omitempty"`
}
@ -146,9 +153,7 @@ type GetUpdatesReq struct {
}
type GetUpdatesResp struct {
Ret int `json:"ret,omitempty"`
Errcode int `json:"errcode,omitempty"`
Errmsg string `json:"errmsg,omitempty"`
APIStatus
Msgs []WeixinMessage `json:"msgs,omitempty"`
SyncBuf string `json:"sync_buf,omitempty"`
GetUpdatesBuf string `json:"get_updates_buf,omitempty"`
@ -161,15 +166,25 @@ type SendMessageReq struct {
}
type SendMessageResp struct {
// Usually empty
APIStatus
}
type GetConfigReq struct {
IlinkUserID string `json:"ilink_user_id,omitempty"`
ContextToken string `json:"context_token,omitempty"`
BaseInfo BaseInfo `json:"base_info,omitempty"`
}
type GetConfigResp struct {
Ret int `json:"ret,omitempty"`
Errmsg string `json:"errmsg,omitempty"`
APIStatus
TypingTicket string `json:"typing_ticket,omitempty"`
}
const (
TypingStatusTyping = 1
TypingStatusCancel = 2
)
type SendTypingReq struct {
IlinkUserID string `json:"ilink_user_id,omitempty"`
TypingTicket string `json:"typing_ticket,omitempty"`
@ -177,6 +192,10 @@ type SendTypingReq struct {
BaseInfo BaseInfo `json:"base_info,omitempty"`
}
type SendTypingResp struct {
APIStatus
}
type QRCodeResponse struct {
Qrcode string `json:"qrcode"`
QrcodeImgContent string `json:"qrcode_img_content"`

120
pkg/channels/weixin/weixin.go
View file

@ -27,6 +27,11 @@ type WeixinChannel struct {
// contextTokens stores the last context_token per user (from_user_id → context_token).
// This is required by the iLink API to associate replies with the right chat session.
contextTokens sync.Map
typingMu sync.Mutex
typingCache map[string]typingTicketCacheEntry
pauseMu sync.Mutex
pauseUntil time.Time
syncBufPath string
}
func init() {
@ -56,6 +61,8 @@ func NewWeixinChannel(cfg config.WeixinConfig, messageBus *bus.MessageBus) (*Wei
api: api,
config: cfg,
bus: messageBus,
typingCache: make(map[string]typingTicketCacheEntry),
syncBufPath: buildWeixinSyncBufPath(cfg),
}, nil
}
@ -87,7 +94,20 @@ func (c *WeixinChannel) pollLoop(ctx context.Context) {
)
consecutiveFails := 0
getUpdatesBuf := ""
getUpdatesBuf, err := loadGetUpdatesBuf(c.syncBufPath)
if err != nil {
logger.WarnCF("weixin", "Failed to load persisted get_updates_buf", map[string]any{
"path": c.syncBufPath,
"error": err.Error(),
})
getUpdatesBuf = ""
} else if getUpdatesBuf != "" {
logger.InfoCF("weixin", "Resuming persisted get_updates_buf", map[string]any{
"path": c.syncBufPath,
"bytes": len(getUpdatesBuf),
"source": "disk",
})
}
nextTimeoutMs := defaultPollTimeoutMs
for {
@ -98,6 +118,13 @@ func (c *WeixinChannel) pollLoop(ctx context.Context) {
default:
}
if err := c.waitWhileSessionPaused(ctx); err != nil {
if ctx.Err() != nil {
return
}
continue
}
// Build a context with timeout slightly longer than the long-poll
pollCtx, pollCancel := context.WithTimeout(ctx, time.Duration(nextTimeoutMs+5000)*time.Millisecond)
@ -138,9 +165,17 @@ func (c *WeixinChannel) pollLoop(ctx context.Context) {
continue
}
// Check for API-level error codes (-14 = session expired)
const sessionExpiredErrcode = -14
if resp.Errcode != 0 || (resp.Ret != 0 && resp.Ret != sessionExpiredErrcode) {
if isSessionExpiredStatus(resp.Ret, resp.Errcode) {
remaining := c.pauseSession("getupdates", resp.Ret, resp.Errcode, resp.Errmsg)
select {
case <-ctx.Done():
return
case <-time.After(remaining):
}
continue
}
if resp.Errcode != 0 || resp.Ret != 0 {
consecutiveFails++
logger.ErrorCF("weixin", "getUpdates API error", map[string]any{
"ret": resp.Ret,
@ -155,17 +190,6 @@ func (c *WeixinChannel) pollLoop(ctx context.Context) {
continue
}
if resp.Errcode == sessionExpiredErrcode || resp.Ret == sessionExpiredErrcode {
logger.ErrorC("weixin", "Session expired — please re-run login")
// Pause for a long time to avoid hammering with a bad token
select {
case <-ctx.Done():
return
case <-time.After(10 * time.Minute):
}
continue
}
consecutiveFails = 0
// Update the long-poll timeout from server hint
@ -176,6 +200,12 @@ func (c *WeixinChannel) pollLoop(ctx context.Context) {
// Advance cursor
if resp.GetUpdatesBuf != "" {
getUpdatesBuf = resp.GetUpdatesBuf
if err := saveGetUpdatesBuf(c.syncBufPath, getUpdatesBuf); err != nil {
logger.WarnCF("weixin", "Failed to persist get_updates_buf", map[string]any{
"path": c.syncBufPath,
"error": err.Error(),
})
}
}
// Dispatch messages
@ -192,6 +222,11 @@ func (c *WeixinChannel) handleInboundMessage(ctx context.Context, msg WeixinMess
return
}
messageID := msg.ClientID
if messageID == "" {
messageID = uuid.New().String()
}
// Build text content from item_list
var parts []string
for _, item := range msg.ItemList {
@ -205,7 +240,7 @@ func (c *WeixinChannel) handleInboundMessage(ctx context.Context, msg WeixinMess
// Use voice → text transcription from server
parts = append(parts, item.VoiceItem.Text)
} else {
parts = append(parts, "[voice message]")
parts = append(parts, "[audio]")
}
case MessageItemTypeImage:
parts = append(parts, "[image]")
@ -220,8 +255,23 @@ func (c *WeixinChannel) handleInboundMessage(ctx context.Context, msg WeixinMess
}
}
var mediaRefs []string
if mediaItem := selectInboundMediaItem(msg); mediaItem != nil {
ref, err := c.downloadMediaFromItem(ctx, fromUserID, messageID, mediaItem)
if err != nil {
logger.ErrorCF("weixin", "Failed to download inbound media", map[string]any{
"from_user_id": fromUserID,
"message_id": messageID,
"type": mediaItem.Type,
"error": err.Error(),
})
} else if ref != "" {
mediaRefs = append(mediaRefs, ref)
}
}
content := strings.Join(parts, "\n")
if content == "" {
if content == "" && len(mediaRefs) == 0 {
return
}
@ -240,11 +290,6 @@ func (c *WeixinChannel) handleInboundMessage(ctx context.Context, msg WeixinMess
return
}
messageID := msg.ClientID
if messageID == "" {
messageID = uuid.New().String()
}
peer := bus.Peer{Kind: "direct", ID: fromUserID}
metadata := map[string]string{
@ -256,6 +301,7 @@ func (c *WeixinChannel) handleInboundMessage(ctx context.Context, msg WeixinMess
logger.DebugCF("weixin", "Received message", map[string]any{
"from_user_id": fromUserID,
"content_len": len(content),
"media_count": len(mediaRefs),
})
// Store context_token for outbound reply association
@ -263,7 +309,7 @@ func (c *WeixinChannel) handleInboundMessage(ctx context.Context, msg WeixinMess
c.contextTokens.Store(fromUserID, msg.ContextToken)
}
c.HandleMessage(ctx, peer, messageID, fromUserID, fromUserID, content, nil, metadata, sender)
c.HandleMessage(ctx, peer, messageID, fromUserID, fromUserID, content, mediaRefs, metadata, sender)
}
// Send implements channels.Channel by sending a text message to the WeChat user.
@ -271,6 +317,9 @@ func (c *WeixinChannel) Send(ctx context.Context, msg bus.OutboundMessage) error
if !c.IsRunning() {
return channels.ErrNotRunning
}
if err := c.ensureSessionActive(); err != nil {
return err
}
if msg.Content == "" {
return nil
@ -294,32 +343,15 @@ func (c *WeixinChannel) Send(ctx context.Context, msg bus.OutboundMessage) error
})
return fmt.Errorf("weixin send: %w: missing context token for chat %s", channels.ErrSendFailed, toUserID)
}
clientID := "picoclaw-" + uuid.New().String()
req := SendMessageReq{
Msg: WeixinMessage{
FromUserID: "",
ToUserID: toUserID,
ClientID: clientID,
MessageType: MessageTypeBot,
MessageState: MessageStateFinish,
ItemList: []MessageItem{
{
Type: MessageItemTypeText,
TextItem: &TextItem{
Text: msg.Content,
},
},
},
ContextToken: contextToken,
},
}
if err := c.api.SendMessage(ctx, req); err != nil {
if err := c.sendTextMessage(ctx, toUserID, contextToken, msg.Content); err != nil {
logger.ErrorCF("weixin", "Failed to send message", map[string]any{
"to_user_id": toUserID,
"error": err.Error(),
})
if c.remainingPause() > 0 {
return fmt.Errorf("weixin send: %w", channels.ErrSendFailed)
}
return fmt.Errorf("weixin send: %w", channels.ErrTemporary)
}

1
pkg/config/config.go
View file

@ -504,6 +504,7 @@ type WeixinConfig struct {
Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_WEIXIN_ENABLED"`
Token string `json:"token" env:"PICOCLAW_CHANNELS_WEIXIN_TOKEN"`
BaseURL string `json:"base_url" env:"PICOCLAW_CHANNELS_WEIXIN_BASE_URL"`
CDNBaseURL string `json:"cdn_base_url" env:"PICOCLAW_CHANNELS_WEIXIN_CDN_BASE_URL"`
Proxy string `json:"proxy" env:"PICOCLAW_CHANNELS_WEIXIN_PROXY"`
AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_WEIXIN_ALLOW_FROM"`
ReasoningChannelID string `json:"reasoning_channel_id" env:"PICOCLAW_CHANNELS_WEIXIN_REASONING_CHANNEL_ID"`

8
pkg/config/defaults.go
View file

@ -174,6 +174,14 @@ func DefaultConfig() *Config {
WelcomeMessage: "Hello! I'm your AI assistant. How can I help you today?",
ProcessingMessage: DefaultWeComAIBotProcessingMessage,
},
Weixin: WeixinConfig{
Enabled: false,
Token: "",
BaseURL: "https://ilinkai.weixin.qq.com/",
CDNBaseURL: "https://novac2c.cdn.weixin.qq.com/c2c",
AllowFrom: FlexibleStringSlice{},
Proxy: "",
},
Pico: PicoConfig{
Enabled: false,
Token: "",