mirror of
https://gitcode.com/JianFeeeee/homeagent-sdk.git
synced 2026-09-20 08:58:03 +00:00
4cb3a0bda4
## 背景：内核侧发现的真问题
在真实二进制压力测试里发现：插件只调 `RegisterOutputChannel("cli", ...)`，
却用同一个通道名 `InjectTextSync("cli", ...)` 注入输入 ⇒ 内核 inputch 登记表里
**没有**这个通道，"把 inputch 划给驻留子"直接失败（`划入 inputch cli: inputch 未注册`）。
根因是**契约没有落到插件与 SDK 面上**：inputch 是内核最基本的**输入路由单位**，
"谁会往这个通道注入输入"必须显式声明，而 SDK 文档没说清它与 RegisterOutputChannel
的分工，示例与模板工程也没有示范。
## SDK 面
- `RegisterInputChannel` / `RegisterOutputChannel` 的文档补齐**方向契约**：
入站（谁会注入）与出站（output_send__<name> 的回复发给谁）是分开登记的两件事；
凡是用 `InjectText*/InjectInput*/InjectInterrupt*(source, "<name>", ...)` 注入的
通道名都要 RegisterInputChannel。README 同步补了一段契约说明。
## 示例插件（全部补齐，之前只有 qq/weather 是对的）
`a2a`、`acp`、`browser`、`memo`：注入用 `p.name` ⇒ 登记 `p.name`；
`calendar`、`rss`：注入用字面量通道名 ⇒ 登记同名通道。
（这些插件此前是"能注入、但通道不在登记表里"，与 cli 同类问题。）
## 模板工程（生成器 templates.go）
- `tmplPluginGo`：示范入站+出站两个方向（含 ChannelDef/NoMemory 说明与 `inputch 未注册` 的成因）。
- `tmplMainLua`：同样两个方向（`register_input_channel` / `register_output_channel`）。
- `tmplReadme`：新增 "Channels" 一节（方向对照表 + 兜底告警说明）。
- 实测：`hmapdev init` 生成的 Go/Lua 工程都含通道代码，Go 工程可构建打包出 `.hmap`；
`--lua` 工程同样生成通道代码。
## 生成器两处修正（都是实测踩出来的）
1. `sdk install --from <dir>`：install 原本只能从 Release 归档下载，而 SDK 开发期的新能力
（如 proc 桥要透传的 `InjectOptions.Priority`）还没发版 ⇒ 生成的工程必然编译失败
（`z_proc_gen.go: opts.Priority undefined`）。现在可用本地源码装一个版本并激活。
实测：`hmapdev sdk install --from <local sdk>` → 装成 v1.3.0 并激活 → 工程构建通过。
2. 构建前置校验 `sdkHasInjectPriority`：proc 桥模板需要 `InjectOptions.Priority`，
旧 SDK 没有时应给出**可执行**的报错（升级 SDK 或用 `--from`），
而不是把两条 `opts.Priority undefined` 编译错误甩给用户（那些错误指向生成物，
完全看不出是 SDK 版本问题）。实测：声明 sdk=1.2.0 的工程构建时正确命中该提示。
## 未决（发布期事项）
`InjectOptions.Priority` 属本特性线新增能力，**已发布的 SDK v1.2.0 不含它**；
发版时 SDK 版本需随之内含该能力（当前源码 meta 已是 1.3.0），否则外部开发者
按文档生成的工程会撞上上面那条守卫。
502 lines
12 KiB
Go
502 lines
12 KiB
Go
package main
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"net/http"
|
||
"os"
|
||
"path/filepath"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
sdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk"
|
||
"github.com/mmcdole/gofeed"
|
||
)
|
||
|
||
const injectDedupWindow = 5 * time.Minute
|
||
|
||
type FeedSub struct {
|
||
URL string `json:"url"`
|
||
Title string `json:"title"`
|
||
AddedAt string `json:"added_at"`
|
||
Interval int `json:"interval"`
|
||
}
|
||
|
||
type Plugin struct {
|
||
name string
|
||
sdk *sdk.PluginSDK
|
||
client *http.Client
|
||
fp *gofeed.Parser
|
||
dataDir string
|
||
mu sync.RWMutex
|
||
feeds []FeedSub
|
||
seenGUIDs map[string]bool
|
||
injected map[string]time.Time
|
||
stopCh chan struct{}
|
||
stopOnce sync.Once
|
||
wg sync.WaitGroup
|
||
pollTicker *time.Ticker
|
||
}
|
||
|
||
func NewPluginFactory(name string, config map[string]interface{}) (sdk.Plugin, error) {
|
||
return &Plugin{name: name}, nil
|
||
}
|
||
|
||
func (p *Plugin) Name() string { return p.name }
|
||
|
||
func getSetting[T string | int64 | float64](s sdk.SettingsAPI, key string, fallback T) T {
|
||
v, err := s.Get(key)
|
||
if err != nil || v == nil {
|
||
return fallback
|
||
}
|
||
switch any(fallback).(type) {
|
||
case string:
|
||
if sv, ok := v.(string); ok {
|
||
return any(sv).(T)
|
||
}
|
||
case int64:
|
||
switch val := v.(type) {
|
||
case float64:
|
||
return any(int64(val)).(T)
|
||
case string:
|
||
if n, err := strconv.ParseInt(val, 10, 64); err == nil {
|
||
return any(n).(T)
|
||
}
|
||
}
|
||
case float64:
|
||
switch val := v.(type) {
|
||
case float64:
|
||
return any(val).(T)
|
||
case string:
|
||
if n, err := strconv.ParseFloat(val, 64); err == nil {
|
||
return any(n).(T)
|
||
}
|
||
}
|
||
}
|
||
return fallback
|
||
}
|
||
|
||
func readArg(args map[string]interface{}, key string) string {
|
||
if v, ok := args[key]; ok && v != nil {
|
||
if s, ok := v.(string); ok {
|
||
return s
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func readArgInt(args map[string]interface{}, key string, fallback int) int {
|
||
if v, ok := args[key]; ok && v != nil {
|
||
switch n := v.(type) {
|
||
case float64:
|
||
return int(n)
|
||
case int64:
|
||
return int(n)
|
||
}
|
||
}
|
||
return fallback
|
||
}
|
||
|
||
func (p *Plugin) Start(s *sdk.PluginSDK) error {
|
||
s.SetAutoRestart(true)
|
||
p.sdk = s
|
||
p.client = &http.Client{Timeout: 30 * time.Second}
|
||
// 入站通道：本插件用 "rss" 通道注入输入（见 Inject* 调用），
|
||
// 输入侧必须显式登记 —— 否则"把该 inputch 划给驻留子"会报 `inputch 未注册`。
|
||
_ = s.RegisterInputChannel("rss", sdk.ChannelDef{NoMemory: true})
|
||
p.fp = gofeed.NewParser()
|
||
p.stopCh = make(chan struct{})
|
||
p.seenGUIDs = make(map[string]bool)
|
||
p.injected = make(map[string]time.Time)
|
||
p.feeds = []FeedSub{}
|
||
|
||
dataDirVal, err := s.Settings().GetCore("core.daemon.data_dir")
|
||
if err != nil || dataDirVal == "" {
|
||
dataDirVal = "."
|
||
}
|
||
p.dataDir = filepath.Join(fmt.Sprint(dataDirVal), "rss")
|
||
if err := os.MkdirAll(p.dataDir, 0755); err != nil {
|
||
fmt.Printf("[%s] mkdir %s: %v\n", p.name, p.dataDir, err)
|
||
}
|
||
p.loadData()
|
||
|
||
// 卸载（删除）时清理订阅数据目录；重载不触发
|
||
s.RegisterOnRemoveHandler(p.cleanupData)
|
||
|
||
s.Settings().RegisterDef(sdk.ConfigDef{
|
||
Key: "poll_interval", Default: "30", Type: "string",
|
||
DisplayName: "Poll Interval", Description: "Default polling interval in minutes (default: 30)",
|
||
Category: "rss",
|
||
})
|
||
|
||
tp := p.name + "_"
|
||
s.RegisterTool(tp+"subscribe", sdk.ToolDef{
|
||
Name: tp + "subscribe", Description: "Subscribe to an RSS/Atom feed URL",
|
||
NoMemory: true,
|
||
Parameters: map[string]interface{}{
|
||
"type": "object",
|
||
"properties": map[string]interface{}{
|
||
"url": map[string]interface{}{"type": "string", "description": "Feed URL"},
|
||
"interval": map[string]interface{}{"type": "integer", "description": "Poll interval in minutes (default: 30, minimum: 5)"},
|
||
},
|
||
"required": []string{"url"},
|
||
},
|
||
}, p.handleSubscribe)
|
||
|
||
s.RegisterTool(tp+"unsubscribe", sdk.ToolDef{
|
||
Name: tp + "unsubscribe", Description: "Unsubscribe from a feed",
|
||
NoMemory: true,
|
||
Parameters: map[string]interface{}{
|
||
"type": "object",
|
||
"properties": map[string]interface{}{
|
||
"url": map[string]interface{}{"type": "string", "description": "Feed URL to unsubscribe"},
|
||
},
|
||
"required": []string{"url"},
|
||
},
|
||
}, p.handleUnsubscribe)
|
||
|
||
s.RegisterTool(tp+"list", sdk.ToolDef{
|
||
Name: tp + "list", Description: "List all subscribed feeds",
|
||
Parameters: map[string]interface{}{
|
||
"type": "object",
|
||
"properties": map[string]interface{}{},
|
||
},
|
||
}, p.handleList)
|
||
|
||
s.RegisterTool(tp+"check_now", sdk.ToolDef{
|
||
Name: tp + "check_now", Description: "Manually check all feeds for new articles now",
|
||
NoMemory: true,
|
||
Parameters: map[string]interface{}{
|
||
"type": "object",
|
||
"properties": map[string]interface{}{},
|
||
},
|
||
}, p.handleCheckNow)
|
||
|
||
pollMin := int(getSetting(s.Settings(), "poll_interval", int64(30)))
|
||
if pollMin < 5 {
|
||
pollMin = 5
|
||
}
|
||
p.pollTicker = time.NewTicker(time.Duration(pollMin) * time.Minute)
|
||
|
||
p.wg.Add(1)
|
||
go p.pollLoop()
|
||
|
||
fmt.Printf("[%s] started (%d feeds, poll every %dm)\n", p.name, len(p.feeds), pollMin)
|
||
return nil
|
||
}
|
||
|
||
func (p *Plugin) Stop() error {
|
||
p.stopOnce.Do(func() { close(p.stopCh) })
|
||
p.pollTicker.Stop()
|
||
p.wg.Wait()
|
||
p.saveData()
|
||
fmt.Printf("[%s] stopped\n", p.name)
|
||
return nil
|
||
}
|
||
|
||
func (p *Plugin) pollLoop() {
|
||
defer p.wg.Done()
|
||
|
||
p.checkAllFeeds()
|
||
|
||
for {
|
||
select {
|
||
case <-p.pollTicker.C:
|
||
p.checkAllFeeds()
|
||
case <-p.stopCh:
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
func (p *Plugin) checkAllFeeds() {
|
||
p.mu.RLock()
|
||
feeds := make([]FeedSub, len(p.feeds))
|
||
copy(feeds, p.feeds)
|
||
p.mu.RUnlock()
|
||
|
||
for _, feed := range feeds {
|
||
select {
|
||
case <-p.stopCh:
|
||
return
|
||
default:
|
||
}
|
||
p.checkFeed(feed)
|
||
}
|
||
}
|
||
|
||
func (p *Plugin) checkFeed(sub FeedSub) {
|
||
parsed, err := p.fp.ParseURL(sub.URL)
|
||
if err != nil {
|
||
return
|
||
}
|
||
|
||
title := parsed.Title
|
||
if title == "" {
|
||
title = sub.URL
|
||
}
|
||
|
||
var newArticles []*gofeed.Item
|
||
for _, item := range parsed.Items {
|
||
guid := item.GUID
|
||
if guid == "" {
|
||
guid = item.Link
|
||
}
|
||
if guid == "" {
|
||
continue
|
||
}
|
||
guid = sub.URL + "|" + guid
|
||
p.mu.RLock()
|
||
seen := p.seenGUIDs[guid]
|
||
p.mu.RUnlock()
|
||
if !seen {
|
||
newArticles = append(newArticles, item)
|
||
}
|
||
}
|
||
|
||
if len(newArticles) == 0 {
|
||
return
|
||
}
|
||
|
||
now := time.Now()
|
||
toInject := make([]*gofeed.Item, 0, len(newArticles))
|
||
p.mu.Lock()
|
||
for _, item := range newArticles {
|
||
guid := item.GUID
|
||
if guid == "" {
|
||
guid = item.Link
|
||
}
|
||
if guid == "" {
|
||
continue
|
||
}
|
||
key := sub.URL + "|" + guid
|
||
if t, ok := p.injected[key]; ok && now.Sub(t) < injectDedupWindow {
|
||
continue
|
||
}
|
||
p.injected[key] = now
|
||
p.seenGUIDs[key] = true
|
||
toInject = append(toInject, item)
|
||
}
|
||
p.mu.Unlock()
|
||
|
||
if len(toInject) == 0 {
|
||
return
|
||
}
|
||
|
||
var lines []string
|
||
lines = append(lines, fmt.Sprintf("📡 %s (%s) — %d 篇新文章:", title, sub.URL, len(toInject)))
|
||
for _, item := range toInject {
|
||
pubDate := ""
|
||
if item.PublishedParsed != nil {
|
||
pubDate = item.PublishedParsed.Format("01-02 15:04")
|
||
}
|
||
line := fmt.Sprintf(" • %s", item.Title)
|
||
if pubDate != "" {
|
||
line += fmt.Sprintf(" [%s]", pubDate)
|
||
}
|
||
if item.Link != "" {
|
||
line += "\n " + item.Link
|
||
}
|
||
lines = append(lines, line)
|
||
}
|
||
|
||
// 中断注入是「系统通知」，NoMemory 写明意图：这类提醒不参与记忆计算，
|
||
// 原文仍进上下文（模型当轮看得到）。
|
||
p.sdk.InjectInterruptTextOpts("rss", "rss", strings.Join(lines, "\n"),
|
||
sdk.InjectOptions{NoMemory: true})
|
||
p.saveData()
|
||
}
|
||
|
||
func (p *Plugin) handleSubscribe(args map[string]interface{}) (interface{}, error) {
|
||
url := readArg(args, "url")
|
||
if url == "" {
|
||
return map[string]interface{}{"isError": true, "content": "URL is required"}, nil
|
||
}
|
||
|
||
p.mu.RLock()
|
||
for _, f := range p.feeds {
|
||
if f.URL == url {
|
||
p.mu.RUnlock()
|
||
return map[string]interface{}{"isError": true, "content": "Already subscribed to: " + url}, nil
|
||
}
|
||
}
|
||
p.mu.RUnlock()
|
||
|
||
interval := readArgInt(args, "interval", 30)
|
||
if interval < 5 {
|
||
interval = 5
|
||
}
|
||
|
||
parsed, err := p.fp.ParseURL(url)
|
||
if err != nil {
|
||
return map[string]interface{}{"isError": true, "content": "Failed to parse feed: " + err.Error()}, nil
|
||
}
|
||
|
||
feedTitle := parsed.Title
|
||
if feedTitle == "" {
|
||
feedTitle = url
|
||
}
|
||
|
||
sub := FeedSub{
|
||
URL: url,
|
||
Title: feedTitle,
|
||
AddedAt: time.Now().Format("2006-01-02 15:04"),
|
||
Interval: interval,
|
||
}
|
||
|
||
guidCount := 0
|
||
p.mu.Lock()
|
||
for _, item := range parsed.Items {
|
||
guid := item.GUID
|
||
if guid == "" {
|
||
guid = item.Link
|
||
}
|
||
if guid == "" {
|
||
continue
|
||
}
|
||
p.seenGUIDs[url+"|"+guid] = true
|
||
guidCount++
|
||
}
|
||
p.mu.Unlock()
|
||
|
||
p.mu.Lock()
|
||
p.feeds = append(p.feeds, sub)
|
||
p.mu.Unlock()
|
||
p.saveData()
|
||
|
||
return map[string]interface{}{
|
||
"content": fmt.Sprintf("Subscribed to: %s\nTitle: %s\nArticles found: %d\nPoll interval: %d min", url, feedTitle, guidCount, interval),
|
||
}, nil
|
||
}
|
||
|
||
func (p *Plugin) handleUnsubscribe(args map[string]interface{}) (interface{}, error) {
|
||
url := readArg(args, "url")
|
||
if url == "" {
|
||
return map[string]interface{}{"isError": true, "content": "URL is required"}, nil
|
||
}
|
||
|
||
p.mu.Lock()
|
||
found := false
|
||
for i, f := range p.feeds {
|
||
if f.URL == url {
|
||
p.feeds = append(p.feeds[:i], p.feeds[i+1:]...)
|
||
found = true
|
||
break
|
||
}
|
||
}
|
||
if !found {
|
||
p.mu.Unlock()
|
||
return map[string]interface{}{"isError": true, "content": "Not subscribed to: " + url}, nil
|
||
}
|
||
|
||
for guid := range p.seenGUIDs {
|
||
if strings.HasPrefix(guid, url+"|") {
|
||
delete(p.seenGUIDs, guid)
|
||
}
|
||
}
|
||
p.mu.Unlock()
|
||
p.saveData()
|
||
|
||
return map[string]interface{}{"content": "Unsubscribed: " + url}, nil
|
||
}
|
||
|
||
func (p *Plugin) handleList(args map[string]interface{}) (interface{}, error) {
|
||
p.mu.RLock()
|
||
defer p.mu.RUnlock()
|
||
|
||
if len(p.feeds) == 0 {
|
||
return map[string]interface{}{"content": "No subscriptions. Use rss_subscribe to add one."}, nil
|
||
}
|
||
|
||
sort.Slice(p.feeds, func(i, j int) bool {
|
||
return p.feeds[i].Title < p.feeds[j].Title
|
||
})
|
||
|
||
var lines []string
|
||
lines = append(lines, fmt.Sprintf("📡 Subscriptions (%d):", len(p.feeds)))
|
||
for _, f := range p.feeds {
|
||
lines = append(lines, fmt.Sprintf(" • %s\n %s (every %dm, added %s)", f.Title, f.URL, f.Interval, f.AddedAt))
|
||
}
|
||
|
||
return map[string]interface{}{"content": strings.Join(lines, "\n")}, nil
|
||
}
|
||
|
||
func (p *Plugin) handleCheckNow(args map[string]interface{}) (interface{}, error) {
|
||
select {
|
||
case <-p.stopCh:
|
||
return map[string]interface{}{"isError": true, "content": "plugin is stopping"}, nil
|
||
default:
|
||
}
|
||
p.wg.Add(1)
|
||
go func() {
|
||
defer p.wg.Done()
|
||
p.checkAllFeeds()
|
||
}()
|
||
return map[string]interface{}{"content": "Checking all feeds for updates..."}, nil
|
||
}
|
||
|
||
func (p *Plugin) dataFile() string {
|
||
return filepath.Join(p.dataDir, "feeds.json")
|
||
}
|
||
|
||
func (p *Plugin) loadData() {
|
||
b, err := os.ReadFile(p.dataFile())
|
||
if err != nil {
|
||
return
|
||
}
|
||
var data struct {
|
||
Feeds []FeedSub `json:"feeds"`
|
||
SeenGUIDs map[string]bool `json:"seen"`
|
||
}
|
||
if json.Unmarshal(b, &data) != nil {
|
||
return
|
||
}
|
||
if data.Feeds != nil {
|
||
p.feeds = data.Feeds
|
||
}
|
||
if data.SeenGUIDs != nil {
|
||
p.seenGUIDs = data.SeenGUIDs
|
||
}
|
||
}
|
||
|
||
func (p *Plugin) saveData() {
|
||
p.mu.RLock()
|
||
defer p.mu.RUnlock()
|
||
data := struct {
|
||
Feeds []FeedSub `json:"feeds"`
|
||
SeenGUIDs map[string]bool `json:"seen"`
|
||
}{
|
||
Feeds: p.feeds,
|
||
SeenGUIDs: p.seenGUIDs,
|
||
}
|
||
b, _ := json.MarshalIndent(data, "", " ")
|
||
atomicWriteJSON(p.dataFile(), b)
|
||
}
|
||
|
||
// cleanupData 卸载时清理订阅数据目录（feeds.json 等）
|
||
func (p *Plugin) cleanupData() {
|
||
p.mu.Lock()
|
||
defer p.mu.Unlock()
|
||
if p.dataDir == "" {
|
||
return
|
||
}
|
||
for _, f := range []string{"feeds.json"} {
|
||
path := filepath.Join(p.dataDir, f)
|
||
if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
|
||
fmt.Printf("[%s] onRemove cleanup %s: %v\n", p.name, path, err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// atomicWriteJSON 原子写 JSON：先写临时文件再 rename，避免进程崩溃截断数据文件。
|
||
func atomicWriteJSON(path string, data []byte) error {
|
||
tmp := path + ".tmp"
|
||
if err := os.WriteFile(tmp, data, 0644); err != nil {
|
||
return err
|
||
}
|
||
return os.Rename(tmp, path)
|
||
}
|