feat(channels/weixin_ilink): 微信 iLink 渠道插件（扫码登录+长轮询+游标+context_token+AES-128-ECB 媒体+24h 主动推送窗口）
This commit is contained in:
Aska committed
1 parent
7831b0ea1b
commit
88dc12466d
13 files changed
+2627
No files matched your search
@@ -0,0 +1,378 @@
|
||||
package weixin_ilink
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||||
)
|
||||
|
||||
const (
|
||||
// msgTypeUser marks an inbound user message (bot messages are type 2).
|
||||
msgTypeUser = 1
|
||||
// item type codes from the iLink reference (design 03 §1).
|
||||
itemText = 1
|
||||
itemImage = 2
|
||||
itemVoice = 3
|
||||
itemFile = 4
|
||||
itemVideo = 5
|
||||
|
||||
// retryFastCount is how many consecutive failures use the fast retry.
|
||||
retryFastCount = 3
|
||||
// retryFastDelay is the delay before the first failures.
|
||||
retryFastDelay = 2 * time.Second
|
||||
// retrySlowDelay is the backoff after retryFastCount consecutive failures.
|
||||
retrySlowDelay = 30 * time.Second
|
||||
)
|
||||
|
||||
// getupdatesResp is the getupdates (long-poll) response envelope.
|
||||
type getupdatesResp struct {
|
||||
// Ret is the top-level business result; non-zero is a transient error.
|
||||
Ret int `json:"ret"`
|
||||
// Errcode is the platform error code; -14 means the session expired.
|
||||
Errcode int `json:"errcode"`
|
||||
// Msgs holds the raw inbound messages delivered in this poll.
|
||||
Msgs []json.RawMessage `json:"msgs"`
|
||||
// GetUpdatesBuf is the next Cursor to echo back.
|
||||
GetUpdatesBuf string `json:"get_updates_buf"`
|
||||
// LongPollTimeoutMs is the server's suggested hold time.
|
||||
LongPollTimeoutMs int `json:"longpolling_timeout_ms"`
|
||||
}
|
||||
|
||||
// rawMsg is the parsed common fields of an inbound WeixinMessage. iLink media
|
||||
// field names are only partially public, so item_list entries are decoded
|
||||
// defensively in parseItem.
|
||||
type rawMsg struct {
|
||||
ClientID string `json:"client_id"`
|
||||
FromUserID string `json:"from_user_id"`
|
||||
ToUserID string `json:"to_user_id"`
|
||||
MessageType int `json:"message_type"`
|
||||
MessageState int `json:"message_state"`
|
||||
ContextToken string `json:"context_token"`
|
||||
GroupID string `json:"group_id"`
|
||||
RoomID string `json:"room_id"`
|
||||
ItemList []json.RawMessage `json:"item_list"`
|
||||
}
|
||||
|
||||
// ReadMessages runs the long-poll read loop until ctx is canceled. Each
|
||||
// converted inbound message is delivered to the handler installed via
|
||||
// SetMessageHandler when one is present, otherwise to emit. The loop is
|
||||
// cooperative with ctx: a canceled context returns immediately and no timers
|
||||
// outlive it.
|
||||
func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMessage)) {
|
||||
a.mu.RLock()
|
||||
connected := a.connected
|
||||
a.mu.RUnlock()
|
||||
if !connected {
|
||||
a.log.Warn("read loop started while disconnected")
|
||||
return
|
||||
}
|
||||
|
||||
failures := 0
|
||||
for {
|
||||
resp, err := a.fetchUpdates(ctx)
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
// A long-poll timeout is the normal "no messages yet" path.
|
||||
if errors.Is(err, context.DeadlineExceeded) {
|
||||
continue
|
||||
}
|
||||
failures++
|
||||
a.log.Warn("getupdates failed", "err", err, "failures", failures)
|
||||
if !a.wait(ctx, a.retryDelay(failures)) {
|
||||
return
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
failures = 0
|
||||
if resp.Errcode == errSessionExpired {
|
||||
a.log.Warn("session expired (errcode=-14); stop polling, re-scan required")
|
||||
a.mu.Lock()
|
||||
a.connected = false
|
||||
a.mu.Unlock()
|
||||
return
|
||||
}
|
||||
if resp.Ret != 0 {
|
||||
failures++
|
||||
a.log.Warn("getupdates business error", "ret", resp.Ret)
|
||||
if !a.wait(ctx, a.retryDelay(failures)) {
|
||||
return
|
||||
}
|
||||
continue
|
||||
}
|
||||
a.setCursor(resp.GetUpdatesBuf)
|
||||
|
||||
for _, raw := range resp.Msgs {
|
||||
token, m := a.convertMessage(raw)
|
||||
if m == nil {
|
||||
continue
|
||||
}
|
||||
a.recordInbound(token, *m)
|
||||
a.deliver(ctx, *m, emit)
|
||||
}
|
||||
if !a.wait(ctx, a.cfg.PollInterval) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// deliver routes a normalized message to the handler when installed, else to
|
||||
// the emit callback (design/08 §1: push-style vs pull-style delivery).
|
||||
func (a *Adapter) deliver(ctx context.Context, m adapter.UnifiedMessage, emit func(adapter.UnifiedMessage)) {
|
||||
if h := a.handlerLocked(); h != nil {
|
||||
h(ctx, m)
|
||||
return
|
||||
}
|
||||
if emit != nil {
|
||||
emit(m)
|
||||
}
|
||||
}
|
||||
|
||||
// fetchUpdates performs one long poll. The request context is bounded by
|
||||
// LongPollTimeout if the caller has not already set a tighter deadline.
|
||||
func (a *Adapter) fetchUpdates(ctx context.Context) (*getupdatesResp, error) {
|
||||
reqCtx, cancel := context.WithTimeout(ctx, a.cfg.LongPollTimeout)
|
||||
defer cancel()
|
||||
|
||||
body, err := json.Marshal(map[string]any{
|
||||
"get_updates_buf": a.cursor(),
|
||||
"base_info": map[string]string{"channel_version": channelVersion},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("weixin_ilink: encode getupdates: %w", err)
|
||||
}
|
||||
var out getupdatesResp
|
||||
if err := a.doJSON(reqCtx, http.MethodPost, a.apiURL("/ilink/bot/getupdates"),
|
||||
bytes.NewReader(body), &out); err != nil {
|
||||
return nil, fmt.Errorf("weixin_ilink: getupdates: %w", err)
|
||||
}
|
||||
return &out, nil
|
||||
}
|
||||
|
||||
// wait sleeps for d or returns false when ctx is canceled. A non-positive
|
||||
// delay returns immediately.
|
||||
func (a *Adapter) wait(ctx context.Context, d time.Duration) bool {
|
||||
if d <= 0 {
|
||||
return true
|
||||
}
|
||||
t := time.NewTimer(d)
|
||||
defer t.Stop()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
case <-t.C:
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// retryDelay grows from the fast retry to the slow backoff after enough
|
||||
// consecutive failures (design/08 §5.4).
|
||||
func (a *Adapter) retryDelay(failures int) time.Duration {
|
||||
if failures <= retryFastCount {
|
||||
return retryFastDelay
|
||||
}
|
||||
return retrySlowDelay
|
||||
}
|
||||
|
||||
// convertMessage maps a raw iLink message to a UnifiedMessage and returns the
|
||||
// raw context_token so the caller can track it per peer. It returns nil when
|
||||
// the message is not an inbound user message or cannot be parsed.
|
||||
func (a *Adapter) convertMessage(raw json.RawMessage) (string, *adapter.UnifiedMessage) {
|
||||
var rm rawMsg
|
||||
if err := json.Unmarshal(raw, &rm); err != nil {
|
||||
a.log.Warn("decode inbound message", "err", err)
|
||||
return "", nil
|
||||
}
|
||||
if rm.MessageType != msgTypeUser {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
group := resolveGroup(&rm)
|
||||
m := &adapter.UnifiedMessage{
|
||||
Platform: platformName,
|
||||
SenderID: rm.FromUserID,
|
||||
SenderName: rm.FromUserID,
|
||||
MessageID: rm.ClientID,
|
||||
Timestamp: a.clock(),
|
||||
RawData: raw,
|
||||
}
|
||||
if group != "" {
|
||||
m.ChannelType = adapter.ChannelGroup
|
||||
m.GroupID = group
|
||||
m.ChannelID = group
|
||||
} else {
|
||||
m.ChannelType = adapter.ChannelDirect
|
||||
m.ChannelID = rm.FromUserID
|
||||
}
|
||||
m.Sender = adapter.Actor{
|
||||
Role: adapter.RoleUnknown,
|
||||
UserID: rm.FromUserID,
|
||||
DisplayName: rm.FromUserID,
|
||||
IsGroup: group != "",
|
||||
GroupID: group,
|
||||
}
|
||||
m.Content, m.ContentType, m.Media = a.convertItems(rm.ItemList)
|
||||
return rm.ContextToken, m
|
||||
}
|
||||
|
||||
// convertItems drains the item list into text, a primary content type and a
|
||||
// media slice. Text from text items is concatenated; voice items contribute
|
||||
// their ASR transcript; image/file/video items are attached lazily as CDN
|
||||
// references (never downloaded during the poll).
|
||||
func (a *Adapter) convertItems(rawItems []json.RawMessage) (string, adapter.ContentType, []adapter.Media) {
|
||||
var text strings.Builder
|
||||
var media []adapter.Media
|
||||
firstMedia := adapter.ContentType("")
|
||||
|
||||
for _, raw := range rawItems {
|
||||
item := parseItem(raw)
|
||||
switch item.kind {
|
||||
case itemText:
|
||||
text.WriteString(item.text)
|
||||
case itemVoice:
|
||||
text.WriteString(item.text)
|
||||
media = append(media, item.media)
|
||||
firstMedia = adapter.ContentVoice
|
||||
case itemImage:
|
||||
media = append(media, item.media)
|
||||
firstMedia = adapter.ContentImage
|
||||
case itemFile, itemVideo:
|
||||
media = append(media, item.media)
|
||||
firstMedia = adapter.ContentFile
|
||||
}
|
||||
}
|
||||
|
||||
content := strings.TrimRight(text.String(), "\n")
|
||||
if firstMedia != "" {
|
||||
return content, firstMedia, media
|
||||
}
|
||||
return content, adapter.ContentText, nil
|
||||
}
|
||||
|
||||
// resolveGroup returns the group identifier when the message carries one,
|
||||
// otherwise "" (direct chat). WeChat group signalling is ⚠️ per design/08 §4,
|
||||
// so several candidate fields are checked defensively.
|
||||
func resolveGroup(rm *rawMsg) string {
|
||||
if rm.GroupID != "" {
|
||||
return rm.GroupID
|
||||
}
|
||||
return rm.RoomID
|
||||
}
|
||||
|
||||
// parsedItem is a normalized view of a single native item_list entry.
|
||||
type parsedItem struct {
|
||||
kind int
|
||||
text string
|
||||
media adapter.Media
|
||||
}
|
||||
|
||||
// parseItem decodes one item_list entry defensively. iLink media field names
|
||||
// are not fully public, so a generic map is scanned for the CDN reference and
|
||||
// metadata with several fallback keys.
|
||||
func parseItem(raw json.RawMessage) parsedItem {
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
return parsedItem{}
|
||||
}
|
||||
kind := intOf(m["type"])
|
||||
switch kind {
|
||||
case itemText:
|
||||
return parsedItem{kind: itemText, text: textOf(m, "text_item")}
|
||||
case itemVoice:
|
||||
return parsedItem{kind: itemVoice, text: textOf(m, "voice_item"), media: mediaOf(m, adapter.MediaVoice)}
|
||||
case itemImage:
|
||||
return parsedItem{kind: itemImage, media: mediaOf(m, adapter.MediaImage)}
|
||||
case itemFile:
|
||||
return parsedItem{kind: itemFile, media: mediaOf(m, adapter.MediaFile)}
|
||||
case itemVideo:
|
||||
return parsedItem{kind: itemVideo, media: mediaOf(m, adapter.MediaVideo)}
|
||||
default:
|
||||
return parsedItem{kind: kind}
|
||||
}
|
||||
}
|
||||
|
||||
// textOf reads the text payload from the named nested sub-object, falling back
|
||||
// to a flat "text" key.
|
||||
func textOf(m map[string]any, sub string) string {
|
||||
if inner, ok := m[sub].(map[string]any); ok {
|
||||
if s, ok := inner["text"].(string); ok {
|
||||
return s
|
||||
}
|
||||
}
|
||||
if s, ok := m["text"].(string); ok {
|
||||
return s
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// mediaOf builds an adapter.Media from a media item, resolving the CDN URL and
|
||||
// metadata from the flat item or a nested sub-object.
|
||||
func mediaOf(m map[string]any, kind adapter.MediaKind) adapter.Media {
|
||||
res := adapter.Media{Kind: kind}
|
||||
res.URL = firstString(m, "url", "cdn_url", "download_url", "file_url", "cdn_ref")
|
||||
res.Name = firstString(m, "file_name", "name")
|
||||
res.Size = firstInt64(m, "size", "file_size")
|
||||
res.MIMEType = firstString(m, "mime", "mime_type", "content_type")
|
||||
if res.URL != "" {
|
||||
return res
|
||||
}
|
||||
// Bounded scan of the common nested sub-objects for the CDN reference.
|
||||
for _, sub := range []string{"image_item", "file_item", "video_item", "media_item", "cdn_item"} {
|
||||
inner, ok := m[sub].(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
res.URL = firstString(inner, "url", "cdn_url", "download_url", "file_url", "cdn_ref")
|
||||
if res.Name == "" {
|
||||
res.Name = firstString(inner, "file_name", "name")
|
||||
}
|
||||
if res.Size == 0 {
|
||||
res.Size = firstInt64(inner, "size", "file_size")
|
||||
}
|
||||
if res.MIMEType == "" {
|
||||
res.MIMEType = firstString(inner, "mime", "mime_type", "content_type")
|
||||
}
|
||||
if res.URL != "" {
|
||||
return res
|
||||
}
|
||||
}
|
||||
return res
|
||||
}
|
||||
|
||||
// intOf coerces a JSON number to int, returning 0 for missing or non-numeric.
|
||||
func intOf(v any) int {
|
||||
f, ok := v.(float64)
|
||||
if !ok {
|
||||
return 0
|
||||
}
|
||||
return int(f)
|
||||
}
|
||||
|
||||
// firstString returns the first non-empty string among keys.
|
||||
func firstString(m map[string]any, keys ...string) string {
|
||||
for _, k := range keys {
|
||||
if s, ok := m[k].(string); ok && s != "" {
|
||||
return s
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// firstInt64 returns the first non-zero integer among keys.
|
||||
func firstInt64(m map[string]any, keys ...string) int64 {
|
||||
for _, k := range keys {
|
||||
if f, ok := m[k].(float64); ok && f != 0 {
|
||||
return int64(f)
|
||||
}
|
||||
}
|
||||
return 0
|
||||
}
|
||||
Reference in new issue
Block a user