package syncer import ( "bytes" "context" "encoding/base64" "encoding/json" "fmt" "io" "net/http" "os" "os/exec" "strings" "time" ) // Writer 是"把文件写到中转"的抽象。 // // 为什么要抽象:HF Bucket 的上传协议要求先做 Xet 内容寻址上传再用 xetHash 登记, // 在 Go 里重造不划算;而除了官方 CLI,中转网关自己就能写(Space 内 /app/storage // 正是桶的 FUSE 挂载点)。两种通道各有取舍,交给配置选。 type Writer interface { // Name 返回通道名(日志与 status 输出用)。 Name() string // Put 写入(覆盖)一个对象,path 为桶内相对路径。 Put(ctx context.Context, path string, content []byte) error // Delete 删除一个对象;对象不存在不应视为错误。 Delete(ctx context.Context, path string) error } // --------------------------------------------------------------------------- // hf CLI:官方实现,自带 Xet 上传 // --------------------------------------------------------------------------- // HFCLIWriter 通过官方 hf CLI 写桶(`hf buckets cp - ` 从 stdin 读内容)。 // // 依赖:本机装了 huggingface_hub(`pip install huggingface_hub`),且 `hf` 在 PATH。 // 这是唯一"零网关依赖"的写入通道——Space 完全不用改、不用唤醒。 type HFCLIWriter struct { Bin string // 可执行文件名或绝对路径,默认 "hf" Bucket string Token string CABundle string // 见 Config.CABundle:把抓包代理根证书喂给 httpx } // Name 实现 Writer。 func (w *HFCLIWriter) Name() string { return "hf-cli" } // Available 报告 hf 可执行文件是否存在(用于启动时选择通道)。 func (w *HFCLIWriter) Available() bool { _, err := w.bin() return err == nil } func (w *HFCLIWriter) bin() (string, error) { bin := w.Bin if strings.TrimSpace(bin) == "" { bin = "hf" } return exec.LookPath(bin) } // Put 实现 Writer:内容经 stdin 交给 `hf buckets cp - hf://buckets//`。 func (w *HFCLIWriter) Put(ctx context.Context, path string, content []byte) error { bin, err := w.bin() if err != nil { return fmt.Errorf("找不到 hf 可执行文件(%v);请 pip install huggingface_hub 或用 writer=relay", err) } dest := fmt.Sprintf("hf://buckets/%s/%s", strings.Trim(w.Bucket, "/"), path) cmd := exec.CommandContext(ctx, bin, "buckets", "cp", "-", dest) cmd.Stdin = bytes.NewReader(content) cmd.Env = w.env() var stderr bytes.Buffer cmd.Stderr = &stderr if err := cmd.Run(); err != nil { return fmt.Errorf("hf buckets cp 失败 (%s): %v: %s", path, err, truncate(stderr.String(), 300)) } return nil } // Delete 实现 Writer。 func (w *HFCLIWriter) Delete(ctx context.Context, path string) error { bin, err := w.bin() if err != nil { return fmt.Errorf("找不到 hf 可执行文件: %w", err) } target := fmt.Sprintf("hf://buckets/%s/%s", strings.Trim(w.Bucket, "/"), path) cmd := exec.CommandContext(ctx, bin, "buckets", "rm", target) cmd.Env = w.env() cmd.Stdout = io.Discard var stderr bytes.Buffer cmd.Stderr = &stderr if err := cmd.Run(); err != nil { // 对象本就不存在 → 删除视为成功(幂等)。 if strings.Contains(strings.ToLower(stderr.String()), "not found") { return nil } return fmt.Errorf("hf buckets rm 失败 (%s): %v: %s", path, err, truncate(stderr.String(), 300)) } return nil } // env 构造子进程环境:注入 HF_TOKEN,并清掉会导致 httpx 解析失败的 [::1] no_proxy。 func (w *HFCLIWriter) env() []string { env := os.Environ() out := env[:0] for _, kv := range env { k := strings.SplitN(kv, "=", 2)[0] // httpx 无法解析 no_proxy 里的 IPv6 方括号写法(实测会抛 // InvalidURL: Invalid port: ':1]'),把它替换成安全值。 if strings.EqualFold(k, "no_proxy") || strings.EqualFold(k, "NO_PROXY") { continue } out = append(out, kv) } out = append(out, "no_proxy=localhost,127.0.0.1", "NO_PROXY=localhost,127.0.0.1") if w.Token != "" { out = append(out, "HF_TOKEN="+w.Token) } // 让 hf 以 UTF-8 输出,避免 Windows 默认 GBK 打死在 ✓ 之类的字符上。 out = append(out, "PYTHONIOENCODING=utf-8") // 本机跑抓包代理时,httpx 只认 certifi(不像 Go 会吃 Windows 信任库), // 会把代理的中间人证书判成不可信 → CERTIFICATE_VERIFY_FAILED。 // 把代理根证书通过这三个环境变量喂给它(不同版本/依赖读的变量名不一样,全给)。 if w.CABundle != "" { out = append(out, "SSL_CERT_FILE="+w.CABundle, "REQUESTS_CA_BUNDLE="+w.CABundle, "CURL_CA_BUNDLE="+w.CABundle, ) } return out } // --------------------------------------------------------------------------- // 中转网关:Space 内写 /app/storage(= Bucket 的 FUSE 挂载点) // --------------------------------------------------------------------------- // RelayWriter 通过中转网关的 /panel/api/sync/files 写文件。 // // 优点:本地不需要 Python/hf CLI,凭证不出网关;写进去的文件立即被池热加载。 // 代价:需要 Space 处于唤醒状态(冷启动 30~60s),所以它适合做"写入通道", // 而读取仍走免唤醒的桶直读。 type RelayWriter struct { BaseURL string // 如 https://cli.3216gemini.dpdns.org APIKey string CABundle string HTTP *http.Client } // Name 实现 Writer。 func (w *RelayWriter) Name() string { return "relay" } // syncPushRequest 是 POST /panel/api/sync/files 的请求体。 type syncPushRequest struct { Files []syncPushFile `json:"files"` HotLoad bool `json:"hot_load"` } type syncPushFile struct { Path string `json:"path"` SHA256 string `json:"sha256,omitempty"` ContentB64 string `json:"content_b64"` Mtime int64 `json:"mtime,omitempty"` // Unix 秒;保留对端 mtime 便于下一轮比较 } // syncPushResponse 是网关的应答。 type syncPushResponse struct { OK bool `json:"ok"` Written int `json:"written"` HotLoaded int `json:"hot_loaded"` Errors []string `json:"errors"` Error string `json:"error"` } // Put 实现 Writer:单文件推送。 func (w *RelayWriter) Put(ctx context.Context, path string, content []byte) error { return w.PutBatch(ctx, []PushFile{{Path: path, Content: content}}, true) } // Delete 实现 Writer:网关侧目前不提供删除(删账号请走面板"移除",避免同步通道 // 变成误删凭证的后门)。 func (w *RelayWriter) Delete(_ context.Context, path string) error { return fmt.Errorf("relay 通道不支持删除(%s):删除账号请在面板操作", path) } // PushFile 是一个待推送文件。 type PushFile struct { Path string Content []byte Mtime time.Time } // PutBatch 批量推送文件(一次 HTTP 调用,减少冷启动后的往返)。 func (w *RelayWriter) PutBatch(ctx context.Context, files []PushFile, hotLoad bool) error { if w.BaseURL == "" { return fmt.Errorf("relay_url 未配置") } reqBody := syncPushRequest{HotLoad: hotLoad} for _, f := range files { reqBody.Files = append(reqBody.Files, syncPushFile{ Path: f.Path, ContentB64: base64.StdEncoding.EncodeToString(f.Content), Mtime: f.Mtime.Unix(), }) } raw, err := json.Marshal(reqBody) if err != nil { return err } url := strings.TrimRight(w.BaseURL, "/") + "/panel/api/sync/files" req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(raw)) if err != nil { return err } req.Header.Set("Content-Type", "application/json") req.Header.Set("User-Agent", userAgent) if w.APIKey != "" { req.Header.Set("Authorization", "Bearer "+w.APIKey) } cli := w.HTTP if cli == nil { cli = newHTTPClient(120*time.Second, false, w.CABundle) } resp, err := cli.Do(req) if err != nil { return fmt.Errorf("推送中转失败: %w", err) } defer func() { _ = resp.Body.Close() }() body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) var out syncPushResponse _ = json.Unmarshal(body, &out) if resp.StatusCode != http.StatusOK { msg := out.Error if msg == "" { msg = truncate(string(body), 300) } return fmt.Errorf("中转返回 %d: %s", resp.StatusCode, msg) } if !out.OK || len(out.Errors) > 0 { return fmt.Errorf("中转部分失败: %s", strings.Join(out.Errors, "; ")) } return nil } // NewWriter 按配置选写入通道。priority: relay 优先还是 hf-cli 优先由 name 决定。 func NewWriter(cfg Config) (Writer, error) { switch strings.ToLower(strings.TrimSpace(cfg.Writer)) { case "", "auto": // auto:有 token 且装了 hf 就用 hf(不依赖网关唤醒);否则退回 relay。 cli := &HFCLIWriter{Bucket: cfg.Bucket, Token: cfg.Token, CABundle: cfg.CABundle} if cfg.Bucket != "" && cli.Available() { return cli, nil } if cfg.RelayURL != "" { return &RelayWriter{BaseURL: cfg.RelayURL, APIKey: cfg.RelayKey}, nil } if cli.Available() { return cli, nil } return nil, fmt.Errorf("无可用写入通道:既没有 hf CLI,也没有配置 relay_url") case "hf", "hf-cli", "cli": return &HFCLIWriter{Bin: cfg.HFBin, Bucket: cfg.Bucket, Token: cfg.Token, CABundle: cfg.CABundle}, nil case "relay", "gateway", "http": if cfg.RelayURL == "" { return nil, fmt.Errorf("writer=relay 但 relay_url 为空") } return &RelayWriter{BaseURL: cfg.RelayURL, APIKey: cfg.RelayKey}, nil default: return nil, fmt.Errorf("未知 writer %q(可选 auto/hf/relay)", cfg.Writer) } }