Spaces:

a3216
/

gcli2api

Running

chore: 同步到上游 1.12.0-panel + 凭证同步/独立启动器/保活

6d60378 verified 20 days ago

9.53 kB

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 - <dest>` 从 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/<bucket>/<path>`。
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)
	}
}

	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 - <dest>` 从 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/<bucket>/<path>`。
	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)
	}
	}