Leo
b355165180
- direct.go: Folders() 正規化（去重保序）/manifestPathFor（單根沿用不丟狀態、多根 sha 尾碼）/RunDirectOnce 多根彙總 - 新測試 4 支全綠（單數相容/多根去重/缺欄驗證/多根 dry-run 標 Root）；既有 7 支不動全綠 - tray: config 加 watch_folders 欄（防存檔洗掉）＋addWatchFolder 資料層；勾選 UI=task 7
349 lines
12 KiB
Go
349 lines
12 KiB
Go
// direct.go — daemon「直送萃取、無 R2」同步模式（SDD ingest-hash-trigger task 11）。
|
||
//
|
||
// 既有 sync 走 collector → R2 → rag_ingest（需 R2 bucket＝綁卡）。direct 模式繞開 R2/Gitea：
|
||
// 監看資料夾 → 偵測新增/改動檔（沿用 Scan 的 hash 差異偵測）→ 讀檔內容 inline POST 進實例的
|
||
// rag_ingest_direct workflow（LLM 萃卡 → 機械切塊 → 寫 kbdb，全在 Arcrun workflow 裡完成）。
|
||
// 刪檔 → 把 removed 事件（collector-trigger.v1）POST 進實例的 rag_ingest workflow removed 分支
|
||
// （只按 page_name 讀 kbdb blocks 並標 deprecated，不碰 R2）。
|
||
//
|
||
// dogfooding（D29 daemon 薄殼豁免）：本檔只做「監看／讀檔／算 hash／HTTP POST」——原生 Go。
|
||
// 萃取／切塊／RAG 一律在實例 workflow，daemon 內零 LLM/切塊邏輯。
|
||
//
|
||
// 用法：
|
||
//
|
||
// collector direct --config <config.json> [--once] [--dry-run]
|
||
//
|
||
// --once：掃一輪就退出（測試/cron 用）；預設常駐輪詢（poll_interval_sec）。
|
||
// --dry-run：只列出會送出的動作，不 POST、不寫 manifest。
|
||
//
|
||
// 跨平台：純 stdlib、輪詢式偵測（不依賴 fsnotify）＝零 CGo，darwin/arm64、windows/amd64 直接交叉編譯。
|
||
package main
|
||
|
||
import (
|
||
"bytes"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
// DirectConfig 是 direct 模式的設定檔（JSON）。設定只走檔案/環境，不落 code。
|
||
type DirectConfig struct {
|
||
WatchFolder string `json:"watch_folder,omitempty"` // 監看的知識資料夾（單數舊制；與 watch_folders 至少填一）
|
||
WatchFolders []string `json:"watch_folders,omitempty"` // 監看的知識資料夾清單（daemon-beta task 1 多資料夾）
|
||
Manifest string `json:"manifest"` // manifest JSON 路徑（必填；多資料夾時為基底名，每根一份帶尾碼）
|
||
CypherURL string `json:"cypher_url"` // 實例 cypher base（必填），如 https://arcrun-cypher-executor.<acct>.workers.dev
|
||
Namespace string `json:"namespace"` // 租戶 namespace（必填），如 demo
|
||
APIKey string `json:"api_key"` // X-Arcrun-API-Key（空＝沿用 namespace，demo 慣例）
|
||
Library string `json:"library"` // 藏書地圖歸庫鍵（空＝kb）
|
||
IngestWF string `json:"ingest_workflow"` // 直送萃取 workflow 名（空＝rag_ingest_direct）
|
||
RemovedWF string `json:"removed_workflow"` // 下架 workflow 名（空＝rag_takedown_direct；吃 {page_name,path}）
|
||
PollSec int `json:"poll_interval_sec"` // 輪詢間隔秒（空/0＝5）
|
||
MaxRemoved float64 `json:"max_removed_ratio"` // 大量刪除防呆門檻（空/0＝0.4）
|
||
}
|
||
|
||
// LoadDirectConfig 讀設定檔並補預設值 + 基本驗證。
|
||
func LoadDirectConfig(path string) (*DirectConfig, error) {
|
||
data, err := os.ReadFile(path)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("讀 config 失敗：%w", err)
|
||
}
|
||
var c DirectConfig
|
||
if err := json.Unmarshal(data, &c); err != nil {
|
||
return nil, fmt.Errorf("config JSON 解析失敗：%w", err)
|
||
}
|
||
var missing []string
|
||
if c.WatchFolder == "" && len(c.WatchFolders) == 0 {
|
||
missing = append(missing, "watch_folder（或 watch_folders）")
|
||
}
|
||
if c.Manifest == "" {
|
||
missing = append(missing, "manifest")
|
||
}
|
||
if c.CypherURL == "" {
|
||
missing = append(missing, "cypher_url")
|
||
}
|
||
if c.Namespace == "" {
|
||
missing = append(missing, "namespace")
|
||
}
|
||
if len(missing) > 0 {
|
||
return nil, fmt.Errorf("config 缺必填欄位：%s", strings.Join(missing, ", "))
|
||
}
|
||
if c.APIKey == "" {
|
||
c.APIKey = c.Namespace
|
||
}
|
||
if c.Library == "" {
|
||
c.Library = "kb"
|
||
}
|
||
if c.IngestWF == "" {
|
||
c.IngestWF = "rag_ingest_direct"
|
||
}
|
||
if c.RemovedWF == "" {
|
||
c.RemovedWF = "rag_takedown_direct"
|
||
}
|
||
if c.PollSec <= 0 {
|
||
c.PollSec = 5
|
||
}
|
||
if c.MaxRemoved <= 0 {
|
||
c.MaxRemoved = DefaultMaxRemovedRatio
|
||
}
|
||
c.CypherURL = strings.TrimSuffix(c.CypherURL, "/")
|
||
return &c, nil
|
||
}
|
||
|
||
// Folders 回傳監看根清單（正規化：單數舊制併入、去重、保序）。
|
||
func (c *DirectConfig) Folders() []string {
|
||
seen := map[string]bool{}
|
||
var out []string
|
||
add := func(p string) {
|
||
if p == "" || seen[p] {
|
||
return
|
||
}
|
||
seen[p] = true
|
||
out = append(out, p)
|
||
}
|
||
add(c.WatchFolder)
|
||
for _, p := range c.WatchFolders {
|
||
add(p)
|
||
}
|
||
return out
|
||
}
|
||
|
||
// manifestPathFor 回傳某根的 manifest 路徑。單根＝沿用 cfg.Manifest（升級不丟既有狀態）；
|
||
// 多根＝每根一份，基底名加 root 絕對路徑的 sha256 前 8 碼尾碼（路徑穩定＝尾碼穩定）。
|
||
func (c *DirectConfig) manifestPathFor(absRoot string) string {
|
||
folders := c.Folders()
|
||
if len(folders) <= 1 {
|
||
return c.Manifest
|
||
}
|
||
sum := sha256.Sum256([]byte(absRoot))
|
||
ext := filepath.Ext(c.Manifest)
|
||
return strings.TrimSuffix(c.Manifest, ext) + "-" + hex.EncodeToString(sum[:4]) + ext
|
||
}
|
||
|
||
// directHTTP 是 direct 模式共用的 HTTP client（萃取 workflow 可能同步跑 LLM，放寬 timeout）。
|
||
var directHTTP = &http.Client{Timeout: 300 * time.Second}
|
||
|
||
// triggerURL 組出 named-webhook 觸發完整 URL。
|
||
func (c *DirectConfig) triggerURL(workflow string) string {
|
||
return fmt.Sprintf("%s/webhooks/named/%s/%s/trigger", c.CypherURL, c.Namespace, workflow)
|
||
}
|
||
|
||
// postJSON POST 一個 JSON body 到 url，回傳 HTTP 狀態碼與回應片段。非 2xx 視為錯誤。
|
||
func (c *DirectConfig) postJSON(url string, body any) (int, string, error) {
|
||
data, err := json.Marshal(body)
|
||
if err != nil {
|
||
return 0, "", err
|
||
}
|
||
req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(data))
|
||
if err != nil {
|
||
return 0, "", err
|
||
}
|
||
req.Header.Set("Content-Type", "application/json")
|
||
req.Header.Set("X-Arcrun-API-Key", c.APIKey)
|
||
resp, err := directHTTP.Do(req)
|
||
if err != nil {
|
||
return 0, "", err
|
||
}
|
||
defer resp.Body.Close()
|
||
snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
|
||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||
return resp.StatusCode, string(snippet), fmt.Errorf("HTTP %d：%s", resp.StatusCode, strings.TrimSpace(string(snippet)))
|
||
}
|
||
return resp.StatusCode, string(snippet), nil
|
||
}
|
||
|
||
// pageNameOf 從相對路徑導出頁名（basename 去副檔名），與 rag_ingest collect_changed 的 pageOf 同語意。
|
||
func pageNameOf(relPath string) string {
|
||
base := relPath
|
||
if i := strings.LastIndex(base, "/"); i >= 0 {
|
||
base = base[i+1:]
|
||
}
|
||
return strings.TrimSuffix(base, filepath.Ext(base))
|
||
}
|
||
|
||
// DirectResult 是單一事件的直送結果（隨每輪 log 輸出）。
|
||
type DirectResult struct {
|
||
Root string `json:"root,omitempty"` // 多資料夾時標明事件屬於哪個根
|
||
Type string `json:"type"`
|
||
Path string `json:"path"`
|
||
Status string `json:"status"` // ingested | removed | planned | failed | skipped
|
||
HTTPStatus int `json:"http_status,omitempty"`
|
||
Error string `json:"error,omitempty"`
|
||
}
|
||
|
||
// RunDirectOnce 對每個監看根掃一輪並彙總結果（daemon-beta task 1 多資料夾）。
|
||
// 單根行為與舊制完全相同（含 manifest 路徑）。回傳彙總結果與退出碼建議（任一根失敗＝1）。
|
||
func RunDirectOnce(cfg *DirectConfig, dryRun bool) ([]DirectResult, int, *TriggerPayload) {
|
||
results := []DirectResult{}
|
||
exit := 0
|
||
var lastPayload *TriggerPayload
|
||
multi := len(cfg.Folders()) > 1
|
||
for _, root := range cfg.Folders() {
|
||
r, e, p := runDirectOnceRoot(cfg, root, dryRun)
|
||
if multi {
|
||
for i := range r {
|
||
r[i].Root = root
|
||
}
|
||
}
|
||
results = append(results, r...)
|
||
if e != 0 {
|
||
exit = e
|
||
}
|
||
if p != nil {
|
||
lastPayload = p
|
||
}
|
||
}
|
||
return results, exit, lastPayload
|
||
}
|
||
|
||
// runDirectOnceRoot 對單一根掃一輪、直送 added/modified/renamed、下架 removed，2xx 後回寫該根 manifest。
|
||
func runDirectOnceRoot(cfg *DirectConfig, root string, dryRun bool) ([]DirectResult, int, *TriggerPayload) {
|
||
results := []DirectResult{}
|
||
exit := 0
|
||
|
||
absRoot, err := filepath.Abs(root)
|
||
if err != nil {
|
||
return append(results, DirectResult{Status: "failed", Error: err.Error()}), 1, nil
|
||
}
|
||
absManifest, err := filepath.Abs(cfg.manifestPathFor(absRoot))
|
||
if err != nil {
|
||
return append(results, DirectResult{Status: "failed", Error: err.Error()}), 1, nil
|
||
}
|
||
m, err := LoadManifest(absManifest, absRoot)
|
||
if err != nil {
|
||
return append(results, DirectResult{Status: "failed", Error: err.Error()}), 1, nil
|
||
}
|
||
payload, err := Scan(absRoot, m, ScanOptions{
|
||
MaxRemovedRatio: cfg.MaxRemoved,
|
||
SkipPaths: map[string]bool{absManifest: true},
|
||
})
|
||
if err != nil {
|
||
return append(results, DirectResult{Status: "failed", Error: err.Error()}), 1, nil
|
||
}
|
||
|
||
now := time.Now().Unix()
|
||
for _, ev := range payload.Events {
|
||
switch ev.Type {
|
||
case "added", "modified", "renamed":
|
||
// renamed 在 direct 模式視同 added：內容未變但為求 kbdb 有這頁名的卡，重送一次萃取
|
||
//（頁名可能改變＝要新頁名的卡）。冪等由 kbdb 端承擔（同頁名覆蓋語意）。
|
||
res := DirectResult{Type: ev.Type, Path: ev.Path}
|
||
full := filepath.Join(absRoot, filepath.FromSlash(ev.Path))
|
||
content, rerr := os.ReadFile(full)
|
||
if rerr != nil {
|
||
res.Status, res.Error = "failed", "讀檔失敗："+rerr.Error()
|
||
results = append(results, res)
|
||
exit = 1
|
||
continue
|
||
}
|
||
if dryRun {
|
||
res.Status = "planned"
|
||
results = append(results, res)
|
||
continue
|
||
}
|
||
status, _, perr := cfg.postJSON(cfg.triggerURL(cfg.IngestWF), map[string]any{
|
||
"page_name": pageNameOf(ev.Path),
|
||
"path": ev.Path,
|
||
"content": string(content),
|
||
"library": cfg.Library,
|
||
})
|
||
res.HTTPStatus = status
|
||
if perr != nil {
|
||
res.Status, res.Error = "failed", perr.Error()
|
||
exit = 1
|
||
} else {
|
||
res.Status = "ingested"
|
||
m.MarkIngested(ev.Path, ev.SourceHash, now) // 2xx 才回寫（下輪不重送）
|
||
}
|
||
results = append(results, res)
|
||
|
||
case "removed":
|
||
res := DirectResult{Type: ev.Type, Path: ev.Path}
|
||
if dryRun {
|
||
res.Status = "planned"
|
||
results = append(results, res)
|
||
continue
|
||
}
|
||
// 下架＝POST {page_name, path} 進 rag_takedown_direct（按 page_name 讀 kbdb blocks
|
||
// 標 deprecated，不碰 R2；獨立於 rag_ingest 的 __CARDS_PREFIX__ 閘——direct 模式檔在
|
||
// 資料夾根，會被 rag_ingest 的前綴閘擋掉，故自帶不含前綴閘的下架 workflow）。
|
||
status, _, perr := cfg.postJSON(cfg.triggerURL(cfg.RemovedWF), map[string]any{
|
||
"page_name": pageNameOf(ev.Path),
|
||
"path": ev.Path,
|
||
})
|
||
res.HTTPStatus = status
|
||
if perr != nil {
|
||
res.Status, res.Error = "failed", perr.Error()
|
||
exit = 1
|
||
} else {
|
||
res.Status = "removed"
|
||
}
|
||
results = append(results, res)
|
||
}
|
||
}
|
||
|
||
// 防呆警告輪：Scan 已壓下 removed 事件，這裡只回報警告不下架。
|
||
for _, w := range payload.Warnings {
|
||
results = append(results, DirectResult{Type: "warning", Status: "skipped", Error: w.Code + ": " + w.Message})
|
||
}
|
||
|
||
if !dryRun {
|
||
if err := m.Save(absManifest); err != nil {
|
||
results = append(results, DirectResult{Status: "failed", Error: "manifest 存檔失敗：" + err.Error()})
|
||
exit = 1
|
||
}
|
||
}
|
||
return results, exit, payload
|
||
}
|
||
|
||
// runDirect 是 `collector direct` 子命令主體。
|
||
func runDirect(args []string) int {
|
||
fs := newFlagSet()
|
||
configPath := fs.String("config", "", "direct 模式設定檔（JSON）路徑（必填）")
|
||
once := fs.Bool("once", false, "掃一輪即退出（測試/cron；預設常駐輪詢）")
|
||
dryRun := fs.Bool("dry-run", false, "只列出會送出的動作，不 POST、不寫 manifest")
|
||
if err := fs.Parse(args); err != nil {
|
||
return 2
|
||
}
|
||
if *configPath == "" {
|
||
fmt.Fprintln(os.Stderr, "錯誤：--config 為必填")
|
||
return 2
|
||
}
|
||
cfg, err := LoadDirectConfig(*configPath)
|
||
if err != nil {
|
||
fmt.Fprintln(os.Stderr, "collector direct:", err)
|
||
return 2
|
||
}
|
||
|
||
runOne := func() int {
|
||
results, exit, _ := RunDirectOnce(cfg, *dryRun)
|
||
out, _ := json.MarshalIndent(struct {
|
||
At string `json:"at"`
|
||
Folders []string `json:"folders"`
|
||
Results []DirectResult `json:"results"`
|
||
}{time.Now().Format(time.RFC3339), cfg.Folders(), results}, "", " ")
|
||
fmt.Println(string(out))
|
||
return exit
|
||
}
|
||
|
||
if *once {
|
||
return runOne()
|
||
}
|
||
// 常駐輪詢：純 stdlib ticker，跨平台。首輪立即跑。
|
||
fmt.Fprintf(os.Stderr, "collector direct daemon 啟動：監看 %s → %s（每 %ds 掃一輪）\n",
|
||
strings.Join(cfg.Folders(), "、"), cfg.triggerURL(cfg.IngestWF), cfg.PollSec)
|
||
runOne()
|
||
ticker := time.NewTicker(time.Duration(cfg.PollSec) * time.Second)
|
||
defer ticker.Stop()
|
||
for range ticker.C {
|
||
runOne()
|
||
}
|
||
return 0
|
||
}
|