AskaEth
7505c46b70
state ∈ 有效 / **被拒（需重扫）** / 未知；🔴 -14 是**被拒不是过期**，且**未知原因绝不 写成「过期」**（面板文案与错误码都如实）。三态判定为纯函数 + 三态各有用例。 ⚠️ 未做（如实登记）： 三态（每账号逐条三态）本片未做。
461 lines
15 KiB
Go
461 lines
15 KiB
Go
package weixin_ilink
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"net/http"
|
||
"net/url"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||
)
|
||
|
||
const (
|
||
// msgTypeUser 标记一条入站用户消息（机器人消息的 type 为 2）。
|
||
msgTypeUser = 1
|
||
// 来自 iLink 参考的 item 类型码（design 03 §1）。
|
||
itemText = 1
|
||
itemImage = 2
|
||
itemVoice = 3
|
||
itemFile = 4
|
||
itemVideo = 5
|
||
|
||
// retryFastCount 是使用快速重试的连续失败次数上限。
|
||
retryFastCount = 3
|
||
// retryFastDelay 是头几次失败前的延迟。
|
||
retryFastDelay = 2 * time.Second
|
||
// retrySlowDelay 是连续失败超过 retryFastCount 之后的退避延迟。
|
||
retrySlowDelay = 30 * time.Second
|
||
)
|
||
|
||
// getupdatesResp 是 getupdates（长轮询）的响应外壳。
|
||
type getupdatesResp struct {
|
||
// Ret 是顶层业务结果；非零表示瞬时错误。
|
||
Ret int `json:"ret"`
|
||
// Errcode 是平台错误码；-14 表示 iLink 拒绝了该会话。
|
||
Errcode int `json:"errcode"`
|
||
// Msgs 保存本次轮询投递的原始入站消息。
|
||
Msgs []json.RawMessage `json:"msgs"`
|
||
// GetUpdatesBuf 是下一次要回传的游标（Cursor）。
|
||
GetUpdatesBuf string `json:"get_updates_buf"`
|
||
// LongPollTimeoutMs 是服务端建议的保持时长。
|
||
LongPollTimeoutMs int `json:"longpolling_timeout_ms"`
|
||
}
|
||
|
||
// rawMsg 是一条入站 WeixinMessage 的公共字段解析结果。iLink 的媒体字段名只有一部分
|
||
// 是公开的，因此 item_list 条目在 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 运行长轮询读循环，直到 ctx 被取消。每条转换后的入站消息，在已通过
|
||
// SetMessageHandler 安装了处理器时交给它，否则交给 emit。该循环与 ctx 协作：上下文
|
||
// 被取消即立即返回，且没有任何定时器会活过它。
|
||
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 {
|
||
// 先拍下本次调用将要使用的凭据快照：长轮询成功时用它置位「已证实」
|
||
// （session_status.go），于是「请求发出后配置就被换掉」不会把旧凭据的证据
|
||
// 记到新凭据头上。
|
||
token := a.botToken()
|
||
resp, err := a.fetchUpdates(ctx)
|
||
if err != nil {
|
||
if ctx.Err() != nil {
|
||
return
|
||
}
|
||
// 长轮询超时是正常的「暂无消息」路径。
|
||
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 == errcodeSessionRejected {
|
||
// iLink 拒绝了这份会话凭据：重试无济于事，轮询循环会永久变哑。记录显式的
|
||
// 被拒绝状态（它同时清掉 connected 标志并以 ERROR 记录日志）并停止循环；
|
||
// 主人必须重新扫码，在新 token 到来之前，下游看到的一律是
|
||
// ErrSessionRejected。
|
||
a.markRejected(rescanReasonGetUpdates)
|
||
return
|
||
}
|
||
if resp.Ret != 0 {
|
||
failures++
|
||
a.log.Warn("getupdates business error", "ret", resp.Ret)
|
||
if !a.wait(ctx, a.retryDelay(failures)) {
|
||
return
|
||
}
|
||
continue
|
||
}
|
||
// 一次成功的 getupdates 就是「iLink 接受了这份凭据」的证据：三态由此从
|
||
// 「未知」变成「有效」（session_status.go）。
|
||
a.markVerified(token)
|
||
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 在已安装处理器时把归一化消息交给它，否则交给 emit 回调（design/08 §1 的
|
||
// `ReadMessages` 的 `emit` 与 `SetMessageHandler` 两条投递路径）。文件类附件的字节
|
||
// 在**投递之前**内联（切片 IM-1）：轮询是这条适配器唯一的入站路径，因此这里是
|
||
// 唯一的决策点。
|
||
func (a *Adapter) deliver(ctx context.Context, m adapter.UnifiedMessage, emit func(adapter.UnifiedMessage)) {
|
||
a.inlineInboundMedia(ctx, &m)
|
||
if h := a.handlerLocked(); h != nil {
|
||
h(ctx, m)
|
||
return
|
||
}
|
||
if emit != nil {
|
||
emit(m)
|
||
}
|
||
}
|
||
|
||
// fetchUpdates 执行一次长轮询。若调用方尚未设置更紧的期限，请求上下文以
|
||
// LongPollTimeout 为界。
|
||
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 休眠 d，或在 ctx 被取消时返回 false。delay 非正数时立即返回。
|
||
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 在连续失败足够多之后，从快速重试切换到慢速退避（见 design/08 §4 的
|
||
// weixin-ilink 条目：长轮询 + 游标）。
|
||
func (a *Adapter) retryDelay(failures int) time.Duration {
|
||
if failures <= retryFastCount {
|
||
return retryFastDelay
|
||
}
|
||
return retrySlowDelay
|
||
}
|
||
|
||
// convertMessage 把一条原始 iLink 消息映射为 UnifiedMessage，并返回原始
|
||
// context_token 以便调用方按对端跟踪它。消息不是入站用户消息、或无法解析时返回 nil。
|
||
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
|
||
}
|
||
// 结构取证（DEBUG，见 inbound_debug.go）：它刻意排在 message_type 判定**之前**，
|
||
// 于是 `message_type != 1` 的非用户消息也留得下一行，「平台没推」与「推了但不是
|
||
// 主人消息」才区分得开。
|
||
a.logInboundMediaShape(&rm)
|
||
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 把 item 列表抽取成文本、一个主内容类型和媒体切片。文本 item 的文本
|
||
// 会拼接起来；语音 item 贡献其 ASR 转写；图片/文件/视频 item 先以 CDN 引用的形式
|
||
// 挂上（转换期不下载）；**文件类**附件的字节随后由 deliver → inlineInboundMedia 在
|
||
// 受支持格式与限额内取回并内联（切片 IM-1），图片/视频仍走各自既有链路。
|
||
//
|
||
// 引用在这里一次性归一成**绝对地址**（`cdnURL`）：它同时是媒体取回、来源白名单与
|
||
// 日志里的 host 字段所看的形态 —— 留在平台给的相对引用上，故障日志只会写「未知来源」，
|
||
// 而绝对地址在取回时仍要过同一道白名单，不会因此放宽。
|
||
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, adapter.ContentText, nil
|
||
}
|
||
for i := range media {
|
||
if media[i].URL != "" {
|
||
media[i].URL = a.cdnURL(media[i].URL)
|
||
}
|
||
}
|
||
return content, firstMedia, media
|
||
}
|
||
|
||
// resolveGroup 在消息携带群标识时返回它，否则返回 ""（私聊）。按 design/08 §4，
|
||
// 微信群信号是 ⚠️，因此对多个候选字段做防御式检查。
|
||
func resolveGroup(rm *rawMsg) string {
|
||
if rm.GroupID != "" {
|
||
return rm.GroupID
|
||
}
|
||
return rm.RoomID
|
||
}
|
||
|
||
// parsedItem 是单个原生 item_list 条目的归一化视图。
|
||
type parsedItem struct {
|
||
kind int
|
||
text string
|
||
media adapter.Media
|
||
}
|
||
|
||
// parseItem 防御式地解码一个 item_list 条目。iLink 的媒体字段名并未完全公开，因此
|
||
// 用通用 map 配合多个兜底键来扫描 CDN 引用与元数据。
|
||
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 从指定的嵌套子对象读取文本载荷，兜底回退到平铺的 "text" 键。
|
||
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 从媒体 item 构造 adapter.Media。iLink 的媒体引用**正确**位置是
|
||
// `<kind>_item.media`（见 cdnMediaOf / mediaRefOf）；平铺的 url/cdn_ref 只是早期逆向
|
||
// 稿的猜测形态，真机从不出现，但仍保留为兜底。
|
||
func mediaOf(m map[string]any, kind adapter.MediaKind) adapter.Media {
|
||
res := adapter.Media{
|
||
Kind: kind,
|
||
URL: mediaRefOf(m),
|
||
Name: firstString(m, "file_name", "name"),
|
||
Size: firstInt64(m, "size", "len", "file_size", "video_size"),
|
||
MIMEType: firstString(m, "mime", "mime_type", "content_type"),
|
||
}
|
||
// 对常见的嵌套子对象做有界扫描，补齐平铺层缺失的元数据（引用已由 mediaRefOf
|
||
// 覆盖，这里只补名称/体积/类型）。
|
||
for _, sub := range mediaSubObjects {
|
||
inner, ok := m[sub].(map[string]any)
|
||
if !ok {
|
||
continue
|
||
}
|
||
if res.Name == "" {
|
||
res.Name = firstString(inner, "file_name", "name")
|
||
}
|
||
if res.Size == 0 {
|
||
res.Size = firstInt64(inner, "size", "len", "file_size", "video_size")
|
||
}
|
||
if res.MIMEType == "" {
|
||
res.MIMEType = firstString(inner, "mime", "mime_type", "content_type")
|
||
}
|
||
}
|
||
return res
|
||
}
|
||
|
||
// mediaSubObjects 是一个 item 里可能承载媒体载荷的嵌套子对象名。`media` 虽也在此列，
|
||
// 但它是 **CDNMedia**（引用与密钥），由 cdnMediaOf/mediaRefOf 专门处理。
|
||
var mediaSubObjects = []string{
|
||
"image_item", "file_item", "voice_item", "video_item", "media_item", "cdn_item",
|
||
}
|
||
|
||
// cdnMediaOf 返回 item 内嵌的 CDNMedia 子对象（`<kind>_item.media`）。官方协议把真正的
|
||
// 下载引用与密钥一并放在这里：`full_url` / `encrypt_query_param` / `aes_key`。
|
||
func cdnMediaOf(item map[string]any) map[string]any {
|
||
for _, sub := range mediaSubObjects {
|
||
inner, ok := item[sub].(map[string]any)
|
||
if !ok {
|
||
continue
|
||
}
|
||
if media, ok := inner["media"].(map[string]any); ok {
|
||
return media
|
||
}
|
||
}
|
||
if media, ok := item["media"].(map[string]any); ok {
|
||
return media
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// mediaRefOf 解析一个媒体 item 的 CDN 引用，返回**待 cdnURL 归一**的引用串：优先
|
||
// `media.full_url`（服务端直接给完整地址），其次把 `media.encrypt_query_param` 折成
|
||
// CDN 根下的相对下载引用，最后才是平铺/嵌套的猜测键。
|
||
func mediaRefOf(item map[string]any) string {
|
||
if media := cdnMediaOf(item); media != nil {
|
||
if full := firstString(media, "full_url"); full != "" {
|
||
return full
|
||
}
|
||
if enc := firstString(media, "encrypt_query_param"); enc != "" {
|
||
return downloadRef(enc)
|
||
}
|
||
}
|
||
if u := firstString(item, "url", "cdn_url", "download_url", "file_url", "cdn_ref"); u != "" {
|
||
return u
|
||
}
|
||
for _, sub := range mediaSubObjects {
|
||
inner, ok := item[sub].(map[string]any)
|
||
if !ok {
|
||
continue
|
||
}
|
||
if u := firstString(inner, "url", "cdn_url", "download_url", "file_url", "cdn_ref"); u != "" {
|
||
return u
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// downloadRef 把 CDN 的加密查询参数折成一个相对下载引用（官方 src/cdn/cdn-url.ts 的
|
||
// buildCdnDownloadUrl，只差 CDN 根 —— 根由 cdnURL 拼接）。参数**必须**转义：
|
||
// encrypt_query_param 是 base64，其中未转义的 `+` 会被服务端当成空格。
|
||
func downloadRef(encryptQueryParam string) string {
|
||
return "download?encrypted_query_param=" + url.QueryEscape(encryptQueryParam)
|
||
}
|
||
|
||
// intOf 把 JSON 数字强制转换为 int，缺失或非数字时返回 0。
|
||
func intOf(v any) int {
|
||
f, ok := v.(float64)
|
||
if !ok {
|
||
return 0
|
||
}
|
||
return int(f)
|
||
}
|
||
|
||
// firstString 返回 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 返回 keys 中第一个非零整数：既接受 JSON 数字，也接受平台偶以**字符串**
|
||
// 下发的数值字段（`file_item.len` 就是字符串 `"1234"`）—— 少了后一支，真机上的文件
|
||
// 体积会一直是 0，超限预筛就形同虚设。
|
||
func firstInt64(m map[string]any, keys ...string) int64 {
|
||
for _, k := range keys {
|
||
switch v := m[k].(type) {
|
||
case float64:
|
||
if v != 0 {
|
||
return int64(v)
|
||
}
|
||
case string:
|
||
if n, err := strconv.ParseInt(strings.TrimSpace(v), 10, 64); err == nil && n != 0 {
|
||
return n
|
||
}
|
||
}
|
||
}
|
||
return 0
|
||
}
|