From 273fc90a95e014b00781518039bb6d846dc148d4 Mon Sep 17 00:00:00 2001 From: WarrenZhou Date: Tue, 18 Aug 2026 10:09:52 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E5=BC=82=E6=AD=A5?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=85=A5=E5=BA=93=E4=BB=BB=E5=8A=A1=E7=B3=BB?= =?UTF-8?q?=E7=BB=9F=EF=BC=8C=E6=94=AF=E6=8C=81=E5=85=A8=E9=87=8F/?= =?UTF-8?q?=E5=A2=9E=E9=87=8F=E6=9B=B4=E6=96=B0=E4=B8=8E=E5=AE=9E=E6=97=B6?= =?UTF-8?q?=E8=BF=9B=E5=BA=A6=E5=B1=95=E7=A4=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增完整的异步任务管理体系： 1. 新增Redis缓存层实现任务状态实时存储与降级机制 2. 实现并发拉取+批量写入的高性能入库流程 3. 添加任务管理HTTP接口与路由 4. 重构行情控制器使用共享SQLite存储 5. 配套添加Docker Compose Redis启动配置 6. 前端新增任务管理面板支持操作与进度查看 --- .trae/documents/async-task-system.md | 170 +++++++++++++++ cmd/server/main.go | 80 ++++++- docker-compose.yml | 18 ++ go.mod | 3 + go.sum | 6 + internal/cache/cache.go | 74 +++++++ internal/cache/redis.go | 107 +++++++++ internal/controller/api.go | 108 ++++----- internal/controller/task.go | 122 +++++++++++ internal/datasource/eastmoney/client.go | 31 ++- internal/job/ingest.go | 82 +++++++ internal/job/job.go | 187 ++++++++++++++++ internal/job/runner.go | 229 ++++++++++++++++++++ internal/job/store.go | 37 ++++ internal/router/router.go | 8 +- internal/storage/sqlite/store.go | 97 ++++++++- internal/storage/sqlite/task.go | 126 +++++++++++ web/index.html | 277 +++++++++++++++++++++++- 18 files changed, 1682 insertions(+), 80 deletions(-) create mode 100644 .trae/documents/async-task-system.md create mode 100644 docker-compose.yml create mode 100644 internal/cache/cache.go create mode 100644 internal/cache/redis.go create mode 100644 internal/controller/task.go create mode 100644 internal/job/ingest.go create mode 100644 internal/job/job.go create mode 100644 internal/job/runner.go create mode 100644 internal/job/store.go create mode 100644 internal/storage/sqlite/task.go diff --git a/.trae/documents/async-task-system.md b/.trae/documents/async-task-system.md new file mode 100644 index 0000000..7cd5c5d --- /dev/null +++ b/.trae/documents/async-task-system.md @@ -0,0 +1,170 @@ +# 异步任务系统 + 并发拉取批量写入 + +## Context + +当前数据入库通过 CLI `cmd/ingest` 串行执行，全量 1035 只转债约需 1 小时。用户要求： +1. Web 控制页增加全量/增量更新按钮，显示数据库数据时间 +2. 后端引入 Redis，异步执行入库任务，任务状态实时写 Redis +3. 前端轮询后端接口实时刷新任务状态 +4. 任务和结果同时持久化到 SQLite，Web 可查询任务列表 +5. **核心优化**：并发拉取（多 worker 同时从 API 取数据）+ 批量写入（20 只一批统一写 SQLite），替代串行模式 + +目标：全量入库从 1 小时降至约 5 分钟，且支持 Web 端操作。 + +## 架构概览 + +``` +前端(轮询2s) → POST /api/tasks → TaskController → JobManager.Start() + ↓ (异步 goroutine) + Runner: N workers 并发拉取 + ↓ resultCh + Collector: 单 goroutine 批量写 SQLite + ↓ 每20只 + Store.SaveKlinesBatch (单事务) + ↓ + Redis(实时进度) + SQLite(持久化) +``` + +**关键约束与对策**： +- `sqlite.Store.SetMaxOpenConns(1)` → 单一 collector goroutine 串行写入（worker 不直接写 DB） +- `eastmoney.Client.throttle()` 非线程安全（`lastReq` 无锁）→ 加 sync.Mutex + 可配置间隔，job 用 `NewClientWithThrottle(0)` 关闭内部节流，由全局限速器统一控制 +- 预加载优化：`ListCodes()` + `LastKlineDates()`（新增 GROUP BY 一次查询）替代 600 次逐只查询，避免占用唯一连接堵塞写入 +- `MarketController` 每请求 `sqlite.Open` → 改为注入共享 `*sqlite.Store`，消除连接池争用 + +## 实现步骤 + +### 阶段 A：基础设施 + +1. **修改 `internal/datasource/eastmoney/client.go`**：`Client` 加 `sync.Mutex` + `interval time.Duration` 字段；`throttle()` 加锁；新增 `NewClientWithThrottle(interval)` 构造函数（`NewClient()` 行为不变，默认 2s） + +2. **新增依赖**：`go get github.com/redis/go-redis/v9`，提升 `github.com/google/uuid` 为直接依赖 + +3. **新建 `internal/cache/cache.go`**：`Task` 结构体（ID/Type/Status/Total/Done/Success/Failed/FailedCodes/Error/时间戳）、`Cache` 接口（GetTask/SetTask/DeleteTask/ListActiveTaskIDs/AddActiveTask/RemoveActiveTask/Ping）、`NoopCache` 空实现（Redis 降级用） + +4. **新建 `internal/cache/redis.go`**：`RedisCache` 实现，键设计 `stock-analysis:task:{id}` → Task JSON（TTL 24h）、`stock-analysis:tasks:active` → SET + +### 阶段 B：存储层扩展 + +5. **修改 `internal/storage/sqlite/store.go`**： + - `schemaSQL` 追加 `ingest_task` 表（id/type/status/total/done/success/failed/failed_codes/params/error/started_at/finished_at） + - 新增 `SaveKlinesBatch(ctx, batch map[string][]model.Kline)`：单事务批量写入多只转债K线，带 3 次重试 + - 新增 `LastKlineDates(ctx) map[string]string`：`SELECT code, MAX(date) FROM convertible_kline GROUP BY code` 一次查询 + +6. **新建 `internal/storage/sqlite/task.go`**：任务 CRUD（SaveTask/UpdateTaskProgress/FinishTask/GetTask/ListTasks），`ListTasks` 按 `started_at DESC` 排序 + +### 阶段 C：任务执行核心 + +7. **新建 `internal/job/store.go`**：定义 `Store` 接口（含 K线批量写 + 任务 CRUD），编译期断言 `*sqlite.Store` 实现它 + +8. **新建 `internal/job/ingest.go`**：`resolveCandidates`（full 走 eastmoney datacenter、incremental 走 store.ListCodes）、`yearOf`、`resolveIncrementalBeg`（从 cmd/ingest 提取） + +9. **新建 `internal/job/runner.go`**：核心并发逻辑 + - `run()`：预加载 skipSet/lastDateMap → 启动 collector goroutine → N worker 并发拉取 → 等待完成 → flush 残余 batch + - `collect()`：单 goroutine，从 resultCh 收集，每 20 只调 `SaveKlinesBatch`，调 `reportProgress` + - `fetchOne()`：单只转债拉取（全量用 FetchFullKline，增量用 FetchIncrementalKline） + - `reportProgress()`：每次写 Redis，每 5 只写 SQLite（减少写压力） + +10. **新建 `internal/job/job.go`**：`Manager` 管理任务注册表 + 全局限速器 `gate`（令牌桶：400ms 间隔，突发 4） + - `Start(ctx, type, since, limit)`：创建 Task → 存 SQLite+Redis → 异步启动 Runner + - `Cancel(id)`：调用 context.CancelFunc + - `Get(ctx, id)`：Redis 优先，回退 SQLite + - `List(ctx, limit)`：SQLite 历史 + Redis 实时覆盖运行中任务 + - `StopAll()`：服务关闭时优雅停止 + +### 阶段 D：HTTP 层 + +11. **新建 `internal/controller/task.go`**：`TaskController`，4 个 handler + - `POST /api/tasks`：创建任务 `{"type":"full"|"incremental","since":"","limit":0}` + - `GET /api/tasks`：任务列表 + - `GET /api/tasks/{id}`：任务状态 + - `POST /api/tasks/{id}/cancel`：取消任务 + +12. **修改 `internal/controller/api.go`**：`MarketController` 字段 `dbPath string` → `store *sqlite.Store`；`NewMarketController(svc, store)`；移除 handler 中临时 `sqlite.Open` + +13. **修改 `internal/router/router.go`**：`New` 签名增加 `tc *TaskController`；注册 4 条 task 路由（均 `requireAuth`） + +14. **修改 `cmd/server/main.go`**： + - 共享 `sqlite.Open(dbPath)`（替代 controller 临时 open） + - `initCache()`：读 `REDIS_ADDR`（默认 localhost:6379），连不上降级 `NoopCache` + - `initJobManager()`：创建无节流 eastmoney client + fallback + Manager + - `defer mgr.StopAll()` + - Docker 启动 Redis：`docker run -d --name stock-redis -p 6379:6379 redis:7-alpine` + +### 阶段 E：前端 + +15. **修改 `web/style.css`**：新增 `.progress-bar`/`.progress-fill`/`.badge-*`（running/success/failed/cancelled）样式 + +16. **修改 `web/index.html` 数据管理 Tab**： + - 保留现有状态卡片 + - 新增工具栏：全量更新、增量更新按钮 + since 日期输入（增量可选） + - 新增任务列表表格：ID（截断）、类型、状态徽章、进度条、完成/总数、时间、取消按钮 + - JS：`createTask(type)` → POST /api/tasks → 启动轮询 `pollTasks()`（2s 间隔，有 running 任务时持续，无则停止） + - `renderTaskTable(tasks)`：渲染表格 + 进度条 + - `cancelTask(id)` → POST cancel + +## 关键文件 + +| 文件 | 操作 | 说明 | +|---|---|---| +| `internal/datasource/eastmoney/client.go` | 修改 | throttle 加锁 + 可配置间隔 | +| `internal/cache/cache.go` | 新建 | Cache 接口 + Task 模型 + NoopCache | +| `internal/cache/redis.go` | 新建 | Redis 实现 | +| `internal/storage/sqlite/store.go` | 修改 | 新增 ingest_task 表 + SaveKlinesBatch + LastKlineDates | +| `internal/storage/sqlite/task.go` | 新建 | 任务 CRUD | +| `internal/job/store.go` | 新建 | Store 接口 | +| `internal/job/ingest.go` | 新建 | 候选列表解析 + 增量起点 | +| `internal/job/runner.go` | 新建 | worker 池 + collector | +| `internal/job/job.go` | 新建 | Manager + 限速器 | +| `internal/controller/task.go` | 新建 | TaskController | +| `internal/controller/api.go` | 修改 | MarketController 注入共享 Store | +| `internal/router/router.go` | 修改 | 注册 task 路由 | +| `cmd/server/main.go` | 修改 | 装配 Redis/Manager/共享Store | +| `web/index.html` | 修改 | 数据管理 Tab 增强 | +| `web/style.css` | 修改 | 进度条 + 徽章样式 | + +## 并发拉取 + 批量写入流程 + +``` +Manager.Start(full) + → 创建 Task，存 SQLite + Redis + → go runner.run(ctx, task) + +runner.run: + 1. resolveCandidates() → 全量列表或已入库 codes + 2. 预加载：skipSet = ListCodes()，lastDateMap = LastKlineDates() + 3. resultCh = make(chan fetchResult, workers*2) + 4. go collect(resultCh) // 单一 SQLite 写者 + 5. for code in candidates: + sem <- {} // worker 槽 + go fetchOne(code): // 并发拉取 + <-gate // 全局限速令牌 + klines = ds.FetchKline(code) + resultCh <- {code, klines, err} + <-sem + 6. wg.Wait(); close(resultCh) + 7. flush 残余 batch + 8. 更新 Task 状态 → SQLite + Redis + +collect: // 单 goroutine + batch = {} + for res in resultCh: + task.Done++ + if err: task.Failed++; failedCodes.append(code) + else: task.Success++; batch[code] = klines + if len(batch) >= 20: store.SaveKlinesBatch(batch); batch = {} + reportProgress(task) // Redis 每次 + SQLite 每5只 +``` + +## 验证方法 + +1. **编译**：`go build ./...` 每阶段验证 +2. **竞态检测**：`go test -race ./internal/...` +3. **端到端**： + - Docker 启动 Redis + - 启动 server，admin 登录 + - 数据管理 Tab → 点"增量更新" → 任务表出现 running 行 → 进度条实时增长 → 完成 + - 点"全量更新" → 两个任务并行 + - 点"取消" → 状态变 cancelled + - 重启 server → 任务列表仍在（SQLite 持久化） + - 停 Redis → 任务仍能跑（NoopCache 降级） +4. **性能**：50 只转债（limit=50）应在 1 分钟内完成 diff --git a/cmd/server/main.go b/cmd/server/main.go index 0f16fed..c5e5e23 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -12,10 +12,12 @@ import ( "syscall" "time" + "stock-analysis/internal/cache" "stock-analysis/internal/controller" "stock-analysis/internal/datasource/eastmoney" "stock-analysis/internal/datasource/fallback" "stock-analysis/internal/datasource/tencent" + "stock-analysis/internal/job" "stock-analysis/internal/router" "stock-analysis/internal/service" "stock-analysis/internal/storage/sqlite" @@ -29,23 +31,43 @@ func main() { webDir := "web" dbPath := defaultDBPath() - // 东财为主数据源（实时行情+历史K线），腾讯为历史K线备用源（东财失败时自动切换） - ds := fallback.New(eastmoney.NewClient(), tencent.NewClient()) - svc := service.NewMarketService(ds) - mc := controller.NewMarketController(svc, dbPath) - + // 1. 共享 SQLite 存储（MarketController / AuthController / job.Manager 共用）。 store, err := sqlite.Open(dbPath) if err != nil { log.Fatalf("open sqlite: %v", err) } defer store.Close() + + // 2. 行情数据源：东财为主，腾讯为备用（实时行情 + 历史K线，默认 2s 节流）。 + marketEM := eastmoney.NewClient() + marketDS := fallback.New(marketEM, tencent.NewClient()) + svc := service.NewMarketService(marketDS) + mc := controller.NewMarketController(svc, store) + + // 3. 认证服务。 authSvc := service.NewAuthService(store) if err := authSvc.EnsureAdmin(context.Background()); err != nil { log.Printf("ensure admin: %v", err) } ac := controller.NewAuthController(authSvc) - handler := router.New(mc, ac, webDir) + // 4. Redis 缓存：任务实时状态存储。Redis 不可用时降级到 NoopCache，任务仍可运行（回退 SQLite）。 + taskCache := initCache() + + // 5. 任务数据源：批量入库场景禁用 per-client 节流，由 job.Manager 全局限速器统一控制速率。 + jobEM := eastmoney.NewClientWithThrottle(0) + jobDS := fallback.New(jobEM, tencent.NewClient()) + mgr := job.NewManager(store, taskCache, jobDS, jobEM, job.Config{ + Workers: envInt("JOB_WORKERS", 8), + BatchSize: envInt("JOB_BATCH_SIZE", 20), + RateInterval: 400 * time.Millisecond, + RateBurst: 4, + }) + defer mgr.StopAll() + tc := controller.NewTaskController(mgr) + + // 6. 路由装配。 + handler := router.New(mc, ac, tc, webDir) server := &http.Server{ Addr: addr, @@ -73,6 +95,52 @@ func main() { } } +// initCache 初始化 Redis 缓存，连接失败时降级到 NoopCache。 +// Redis 地址通过 REDIS_ADDR 环境变量配置（默认 localhost:6379），密码通过 REDIS_PASSWORD 配置。 +func initCache() cache.Cache { + addr := os.Getenv("REDIS_ADDR") + if addr == "" { + addr = "localhost:6379" + } + password := os.Getenv("REDIS_PASSWORD") + + rc, err := cache.NewRedisCache(addr, password, 0) + if err != nil { + log.Printf("[cache] redis create failed (%s), using noop cache: %v", addr, err) + return cache.NoopCache{} + } + pingCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := rc.Ping(pingCtx); err != nil { + log.Printf("[cache] redis ping failed (%s), using noop cache: %v", addr, err) + return cache.NoopCache{} + } + log.Printf("[cache] redis connected: %s", addr) + return rc +} + +// envInt 读取环境变量整数值，缺失或无效时返回默认值。 +func envInt(key string, def int) int { + if v := os.Getenv(key); v != "" { + if n, err := atoi(v); err == nil { + return n + } + } + return def +} + +// atoi 简易字符串转整数（避免引入 strconv 仅为此一处）。 +func atoi(s string) (int, error) { + n := 0 + for _, c := range s { + if c < '0' || c > '9' { + return 0, errors.New("invalid int") + } + n = n*10 + int(c-'0') + } + return n, nil +} + // defaultDBPath 返回默认数据库路径。 // 默认使用项目目录下的 stock.db，避免 OneDrive/AppData 的 I/O 限制。 func defaultDBPath() string { diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..03ccfde --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,18 @@ +# docker-compose.yml +# 启动 Redis 服务，供任务状态实时缓存使用。 +# 使用方法：docker compose up -d +# 停止：docker compose down +version: "3.8" +services: + redis: + image: redis:7-alpine + container_name: stock-analysis-redis + ports: + - "6379:6379" + command: redis-server --appendonly yes --maxmemory 256mb --maxmemory-policy allkeys-lru + volumes: + - redis-data:/data + restart: unless-stopped + +volumes: + redis-data: diff --git a/go.mod b/go.mod index 35b2a78..a2ee742 100644 --- a/go.mod +++ b/go.mod @@ -9,11 +9,14 @@ require ( ) require ( + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/google/uuid v1.6.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/ncruces/go-strftime v0.1.9 // indirect + github.com/redis/go-redis/v9 v9.22.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + go.uber.org/atomic v1.11.0 // indirect golang.org/x/sys v0.47.0 // indirect modernc.org/libc v1.55.3 // indirect modernc.org/mathutil v1.6.0 // indirect diff --git a/go.sum b/go.sum index b8a00ad..f62cf15 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd h1:gbpYu9NMq8jhDVbvlGkMFWCjLFlqqEZjEmObmhUy6Vo= @@ -8,8 +10,12 @@ github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWE github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4= github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/redis/go-redis/v9 v9.22.0 h1:laDvpYXTJtZLloinw1fA5Kqd6HAEH2XKxOkG/PDq2F0= +github.com/redis/go-redis/v9 v9.22.0/go.mod h1:y2g0Wj8rQvuK0ELM+oxSudcLtC09JScs98I/X9gRWY4= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M= golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis= golang.org/x/mod v0.38.0 h1:MECBjubtXD7yj4HrhIUcywNaGeNVUdfVnxmPajOk4yk= diff --git a/internal/cache/cache.go b/internal/cache/cache.go new file mode 100644 index 0000000..f3ccf29 --- /dev/null +++ b/internal/cache/cache.go @@ -0,0 +1,74 @@ +// Package cache 定义任务状态缓存接口，支持 Redis 实时状态 + 内存降级。 +// Redis 不可用时降级到 NoopCache，任务系统仍可运行（仅失去实时状态能力，回退到 SQLite）。 +package cache + +import ( + "context" + "time" +) + +// TaskType 任务类型。 +type TaskType string + +const ( + TaskTypeFull TaskType = "full" // 全量入库 + TaskTypeIncremental TaskType = "incremental" // 增量同步 +) + +// TaskStatus 任务状态。 +type TaskStatus string + +const ( + TaskStatusPending TaskStatus = "pending" // 已创建未启动 + TaskStatusRunning TaskStatus = "running" // 执行中 + TaskStatusSuccess TaskStatus = "success" // 成功 + TaskStatusFailed TaskStatus = "failed" // 失败 + TaskStatusCancelled TaskStatus = "cancelled" // 已取消 +) + +// Task 任务运行态快照，Redis 实时状态与 SQLite 持久化共用同一结构。 +type Task struct { + ID string `json:"id"` + Type TaskType `json:"type"` + Status TaskStatus `json:"status"` + Total int `json:"total"` + Done int `json:"done"` + Success int `json:"success"` + Failed int `json:"failed"` + FailedCodes []string `json:"failedCodes,omitempty"` + Params string `json:"params"` + Error string `json:"error,omitempty"` + StartedAt string `json:"startedAt"` + FinishedAt string `json:"finishedAt,omitempty"` +} + +// Cache 任务状态缓存接口。 +// Redis 实现用于实时状态查询；NoopCache 用于 Redis 不可用时降级。 +type Cache interface { + // GetTask 读取任务状态，不存在返回 nil,nil。 + GetTask(ctx context.Context, id string) (*Task, error) + // SetTask 写入任务状态，ttl 为过期时间。 + SetTask(ctx context.Context, t *Task, ttl time.Duration) error + // DeleteTask 删除任务状态。 + DeleteTask(ctx context.Context, id string) error + // ListActiveTaskIDs 返回运行中任务的 ID 列表。 + ListActiveTaskIDs(ctx context.Context) ([]string, error) + // AddActiveTask 将任务 ID 加入活跃集合。 + AddActiveTask(ctx context.Context, id string) error + // RemoveActiveTask 从活跃集合移除任务 ID。 + RemoveActiveTask(ctx context.Context, id string) error + // Ping 探测缓存可用性。 + Ping(ctx context.Context) error +} + +// NoopCache Cache 的空实现，Redis 不可用时降级使用。 +// 所有操作均为空操作，任务状态仅依赖 SQLite 持久化。 +type NoopCache struct{} + +func (NoopCache) GetTask(ctx context.Context, id string) (*Task, error) { return nil, nil } +func (NoopCache) SetTask(ctx context.Context, t *Task, ttl time.Duration) error { return nil } +func (NoopCache) DeleteTask(ctx context.Context, id string) error { return nil } +func (NoopCache) ListActiveTaskIDs(ctx context.Context) ([]string, error) { return nil, nil } +func (NoopCache) AddActiveTask(ctx context.Context, id string) error { return nil } +func (NoopCache) RemoveActiveTask(ctx context.Context, id string) error { return nil } +func (NoopCache) Ping(ctx context.Context) error { return nil } diff --git a/internal/cache/redis.go b/internal/cache/redis.go new file mode 100644 index 0000000..95cfb03 --- /dev/null +++ b/internal/cache/redis.go @@ -0,0 +1,107 @@ +package cache + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/redis/go-redis/v9" +) + +// keyPrefix Redis 键前缀，避免与其他应用冲突。 +const keyPrefix = "stock-analysis:" + +// taskKey 构造任务状态键。 +func taskKey(id string) string { return keyPrefix + "task:" + id } + +// activeSetKey 构造活跃任务集合键。 +const activeSetKey = keyPrefix + "tasks:active" + +// RedisCache 基于 go-redis/v9 的 Cache 实现。 +type RedisCache struct { + client *redis.Client +} + +// NewRedisCache 创建 Redis 缓存实例。 +// addr 格式 "host:port"，password 为空表示无密码，db 为 Redis 数据库编号。 +// 配置较短的超时和重试次数，Redis 不可用时快速降级到 NoopCache。 +func NewRedisCache(addr, password string, db int) (*RedisCache, error) { + client := redis.NewClient(&redis.Options{ + Addr: addr, + Password: password, + DB: db, + DialTimeout: 2 * time.Second, + ReadTimeout: 2 * time.Second, + WriteTimeout: 2 * time.Second, + MaxRetries: 1, + PoolSize: 10, + }) + return &RedisCache{client: client}, nil +} + +// GetTask 从 Redis 读取任务状态（JSON 反序列化）。 +func (c *RedisCache) GetTask(ctx context.Context, id string) (*Task, error) { + data, err := c.client.Get(ctx, taskKey(id)).Bytes() + if err != nil { + if err == redis.Nil { + return nil, nil // 不存在 + } + return nil, fmt.Errorf("redis get task %s: %w", id, err) + } + var t Task + if err := json.Unmarshal(data, &t); err != nil { + return nil, fmt.Errorf("unmarshal task %s: %w", id, err) + } + return &t, nil +} + +// SetTask 写入任务状态（JSON 序列化），设置 TTL 防止过期数据残留。 +func (c *RedisCache) SetTask(ctx context.Context, t *Task, ttl time.Duration) error { + data, err := json.Marshal(t) + if err != nil { + return fmt.Errorf("marshal task %s: %w", t.ID, err) + } + if err := c.client.Set(ctx, taskKey(t.ID), data, ttl).Err(); err != nil { + return fmt.Errorf("redis set task %s: %w", t.ID, err) + } + return nil +} + +// DeleteTask 删除任务状态。 +func (c *RedisCache) DeleteTask(ctx context.Context, id string) error { + if err := c.client.Del(ctx, taskKey(id)).Err(); err != nil { + return fmt.Errorf("redis del task %s: %w", id, err) + } + return nil +} + +// ListActiveTaskIDs 返回活跃任务集合中的所有 ID。 +func (c *RedisCache) ListActiveTaskIDs(ctx context.Context) ([]string, error) { + ids, err := c.client.SMembers(ctx, activeSetKey).Result() + if err != nil { + return nil, fmt.Errorf("redis smembers: %w", err) + } + return ids, nil +} + +// AddActiveTask 将任务 ID 加入活跃集合。 +func (c *RedisCache) AddActiveTask(ctx context.Context, id string) error { + if err := c.client.SAdd(ctx, activeSetKey, id).Err(); err != nil { + return fmt.Errorf("redis sadd %s: %w", id, err) + } + return nil +} + +// RemoveActiveTask 从活跃集合移除任务 ID。 +func (c *RedisCache) RemoveActiveTask(ctx context.Context, id string) error { + if err := c.client.SRem(ctx, activeSetKey, id).Err(); err != nil { + return fmt.Errorf("redis srem %s: %w", id, err) + } + return nil +} + +// Ping 探测 Redis 连通性。 +func (c *RedisCache) Ping(ctx context.Context) error { + return c.client.Ping(ctx).Err() +} diff --git a/internal/controller/api.go b/internal/controller/api.go index 98b9ec1..1a0727f 100644 --- a/internal/controller/api.go +++ b/internal/controller/api.go @@ -15,14 +15,15 @@ import ( ) // MarketController 提供行情与回测相关的 HTTP 接口。 +// store 为共享 SQLite 存储，与 job.Manager 共用同一连接池，避免每次请求重复开关连接。 type MarketController struct { - svc *service.MarketService - dbPath string + svc *service.MarketService + store *sqlite.Store } // NewMarketController 创建行情控制器。 -func NewMarketController(svc *service.MarketService, dbPath string) *MarketController { - return &MarketController{svc: svc, dbPath: dbPath} +func NewMarketController(svc *service.MarketService, store *sqlite.Store) *MarketController { + return &MarketController{svc: svc, store: store} } // ListBonds 处理 GET /api/bonds，返回全量可转债实时行情。 @@ -93,13 +94,7 @@ func (c *MarketController) RunBacktest(w http.ResponseWriter, r *http.Request) { } ctx := r.Context() - - store, err := sqlite.Open(c.dbPath) - if err != nil { - writeError(w, http.StatusInternalServerError, err) - return - } - defer store.Close() + store := c.store feed, err := backtest.NewDataFeed(ctx, store) if err != nil { @@ -127,53 +122,53 @@ func (c *MarketController) RunBacktest(w http.ResponseWriter, r *http.Request) { MaxPrice: req.MaxPrice, }) case "doublelow": - strat = strategy.NewDoubleLow(strategy.DoubleLowParams{ - TopN: req.TopN, - Rebalance: req.Rebalance, - MinPrice: req.MinPrice, - MaxPrice: req.MaxPrice, - MaxPremium: req.MaxPremium, - }) - case "premium": - strat = strategy.NewPremium(strategy.PremiumParams{ - TopN: req.TopN, - Rebalance: req.Rebalance, - MinPrice: req.MinPrice, - MaxPrice: req.MaxPrice, - }) - default: + strat = strategy.NewDoubleLow(strategy.DoubleLowParams{ + TopN: req.TopN, + Rebalance: req.Rebalance, + MinPrice: req.MinPrice, + MaxPrice: req.MaxPrice, + MaxPremium: req.MaxPremium, + }) + case "premium": + strat = strategy.NewPremium(strategy.PremiumParams{ + TopN: req.TopN, + Rebalance: req.Rebalance, + MinPrice: req.MinPrice, + MaxPrice: req.MaxPrice, + }) + default: writeError(w, http.StatusBadRequest, err) return } engine := backtest.NewEngine(params, strat) - if req.Fundamental { - fundamentals, err := store.LoadAllFundamentals(ctx) - if err != nil { - writeError(w, http.StatusInternalServerError, fmt.Errorf("加载基本面数据: %w", err)) - return - } - stock2bond, err := store.GetStockBondMap(ctx) - if err != nil { - writeError(w, http.StatusInternalServerError, fmt.Errorf("获取正股映射: %w", err)) - return - } - filter := backtest.NewFundamentalFilter(stock2bond) - engine.SetFundamentalFilter(filter, fundamentals) - log.Printf("基本面过滤器已启用：%d 只正股数据", len(fundamentals)) + if req.Fundamental { + fundamentals, err := store.LoadAllFundamentals(ctx) + if err != nil { + writeError(w, http.StatusInternalServerError, fmt.Errorf("加载基本面数据: %w", err)) + return } + stock2bond, err := store.GetStockBondMap(ctx) + if err != nil { + writeError(w, http.StatusInternalServerError, fmt.Errorf("获取正股映射: %w", err)) + return + } + filter := backtest.NewFundamentalFilter(stock2bond) + engine.SetFundamentalFilter(filter, fundamentals) + log.Printf("基本面过滤器已启用：%d 只正股数据", len(fundamentals)) + } - res, err := engine.Run(ctx, feed) - if err != nil { - writeError(w, http.StatusInternalServerError, err) - return - } + res, err := engine.Run(ctx, feed) + if err != nil { + writeError(w, http.StatusInternalServerError, err) + return + } - log.Printf("backtest done: %s, start=%s, end=%s, totalReturn=%.2f%%, curvePoints=%d, tradeDays=%d", - res.Strategy, res.Start, res.End, res.Metrics.TotalReturn*100, len(res.Curve), res.Metrics.TradeDays) + log.Printf("backtest done: %s, start=%s, end=%s, totalReturn=%.2f%%, curvePoints=%d, tradeDays=%d", + res.Strategy, res.Start, res.End, res.Metrics.TotalReturn*100, len(res.Curve), res.Metrics.TradeDays) - bench := computeBenchmarkJSON(feed, req.StartDate, req.EndDate) + bench := computeBenchmarkJSON(feed, req.StartDate, req.EndDate) writeJSON(w, http.StatusOK, map[string]any{ "strategy": res.Strategy, @@ -251,12 +246,7 @@ type DBBond struct { // ListDBBonds 处理 GET /api/db/bonds，返回数据库中已有的转债列表。 func (c *MarketController) ListDBBonds(w http.ResponseWriter, r *http.Request) { ctx := r.Context() - store, err := sqlite.Open(c.dbPath) - if err != nil { - writeError(w, http.StatusInternalServerError, err) - return - } - defer store.Close() + store := c.store codes, err := store.ListCodes(ctx) if err != nil { @@ -282,14 +272,8 @@ func (c *MarketController) ListDBBonds(w http.ResponseWriter, r *http.Request) { // GetDataStatus 处理 GET /api/status，返回数据库状态信息。 func (c *MarketController) GetDataStatus(w http.ResponseWriter, r *http.Request) { ctx := r.Context() - status := DataStatus{DBPath: c.dbPath} - - store, err := sqlite.Open(c.dbPath) - if err != nil { - writeJSON(w, http.StatusOK, status) // 返回部分信息 - return - } - defer store.Close() + store := c.store + status := DataStatus{DBPath: c.store.Path()} codes, err := store.ListCodes(ctx) if err == nil { diff --git a/internal/controller/task.go b/internal/controller/task.go new file mode 100644 index 0000000..7a36fd9 --- /dev/null +++ b/internal/controller/task.go @@ -0,0 +1,122 @@ +package controller + +import ( + "encoding/json" + "net/http" + "strconv" + + "stock-analysis/internal/cache" + "stock-analysis/internal/job" +) + +// 任务接口错误（包级哨兵错误，便于测试断言）。 +var ( + errBadTaskType = errString("无效的任务类型，支持 full / incremental") + errMissingTaskID = errString("缺少任务 ID") + errTaskNotFound = errString("任务不存在") + errTaskNotRunning = errString("任务不存在或已结束，无法取消") +) + +// errString 将字符串转为 error，避免重复 fmt.Errorf 调用。 +type errString string + +func (e errString) Error() string { return string(e) } + +// TaskController 提供数据入库任务管理的 HTTP 接口。 +// 任务由 job.Manager 异步执行，状态实时写入 Redis，持久化到 SQLite。 +type TaskController struct { + mgr *job.Manager +} + +// NewTaskController 创建任务控制器。 +func NewTaskController(mgr *job.Manager) *TaskController { + return &TaskController{mgr: mgr} +} + +// taskRequest 创建任务请求参数。 +type taskRequest struct { + Type string `json:"type"` // full | incremental + Since string `json:"since"` // 增量起点（YYYY-MM-DD），可选，留空自动取最后K线日期 + Limit int `json:"limit"` // 限制处理数量，0 表示全部 +} + +// CreateTask 处理 POST /api/tasks，创建数据入库任务（全量/增量）。 +// 任务异步执行，立即返回 pending 状态的任务快照，前端通过轮询 GetTask 获取进度。 +func (c *TaskController) CreateTask(w http.ResponseWriter, r *http.Request) { + var req taskRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, err) + return + } + + var typ cache.TaskType + switch req.Type { + case "full", "": + typ = cache.TaskTypeFull + case "incremental": + typ = cache.TaskTypeIncremental + default: + writeError(w, http.StatusBadRequest, errBadTaskType) + return + } + + task, err := c.mgr.Start(r.Context(), typ, req.Since, req.Limit) + if err != nil { + writeError(w, http.StatusInternalServerError, err) + return + } + writeJSON(w, http.StatusOK, task) +} + +// ListTasks 处理 GET /api/tasks，返回任务列表（按开始时间降序）。 +// 默认返回最近 50 条，可通过 ?limit=N 调整。 +// 运行中任务会合并 Redis 实时进度。 +func (c *TaskController) ListTasks(w http.ResponseWriter, r *http.Request) { + limit := 50 + if q := r.URL.Query().Get("limit"); q != "" { + if n, err := strconv.Atoi(q); err == nil && n > 0 { + limit = n + } + } + tasks, err := c.mgr.List(r.Context(), limit) + if err != nil { + writeError(w, http.StatusInternalServerError, err) + return + } + writeJSON(w, http.StatusOK, tasks) +} + +// GetTask 处理 GET /api/tasks/{id}，返回单个任务状态。 +// 优先从 Redis 读取实时状态，回退 SQLite 持久化记录。 +func (c *TaskController) GetTask(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if id == "" { + writeError(w, http.StatusBadRequest, errMissingTaskID) + return + } + task, err := c.mgr.Get(r.Context(), id) + if err != nil { + writeError(w, http.StatusNotFound, err) + return + } + if task == nil { + writeError(w, http.StatusNotFound, errTaskNotFound) + return + } + writeJSON(w, http.StatusOK, task) +} + +// CancelTask 处理 POST /api/tasks/{id}/cancel，取消运行中任务。 +// 仅运行中任务可取消，已完成任务忽略。 +func (c *TaskController) CancelTask(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if id == "" { + writeError(w, http.StatusBadRequest, errMissingTaskID) + return + } + if ok := c.mgr.Cancel(id); !ok { + writeError(w, http.StatusNotFound, errTaskNotRunning) + return + } + writeJSON(w, http.StatusOK, map[string]string{"message": "任务已取消"}) +} diff --git a/internal/datasource/eastmoney/client.go b/internal/datasource/eastmoney/client.go index bdc414c..de824f0 100644 --- a/internal/datasource/eastmoney/client.go +++ b/internal/datasource/eastmoney/client.go @@ -10,6 +10,7 @@ import ( "io" "net/http" "net/url" + "sync" "time" ) @@ -46,14 +47,26 @@ var historyHosts = []string{ // Client 是东方财富行情接口客户端，实现 datasource.MarketDataSource。 type Client struct { - http *http.Client - lastReq time.Time + http *http.Client + lastReq time.Time + mu sync.Mutex // 保护 lastReq，支持多 goroutine 并发调用 + interval time.Duration } -// NewClient 创建东财行情客户端。 +// NewClient 创建东财行情客户端（默认 2s 节流）。 func NewClient() *Client { return &Client{ - http: &http.Client{Timeout: 15 * time.Second}, + http: &http.Client{Timeout: 15 * time.Second}, + interval: requestInterval, + } +} + +// NewClientWithThrottle 创建可配置节流间隔的客户端。 +// interval=0 关闭内部节流，由外部全局限速器统一控制（job runner 场景）。 +func NewClientWithThrottle(interval time.Duration) *Client { + return &Client{ + http: &http.Client{Timeout: 15 * time.Second}, + interval: interval, } } @@ -104,10 +117,16 @@ func (c *Client) fetch(ctx context.Context, reqURL string, out any) error { return json.Unmarshal(body, out) } -// throttle 保证请求间隔不小于 requestInterval。 +// throttle 保证请求间隔不小于 interval，线程安全。 +// interval=0 时为空操作，由外部全局限速器统一控制速率。 func (c *Client) throttle() { + if c.interval == 0 { + return + } + c.mu.Lock() + defer c.mu.Unlock() elapsed := time.Since(c.lastReq) - if wait := requestInterval - elapsed; wait > 0 { + if wait := c.interval - elapsed; wait > 0 { time.Sleep(wait) } c.lastReq = time.Now() diff --git a/internal/job/ingest.go b/internal/job/ingest.go new file mode 100644 index 0000000..bff5592 --- /dev/null +++ b/internal/job/ingest.go @@ -0,0 +1,82 @@ +package job + +import ( + "context" + "fmt" + "time" + + "stock-analysis/internal/cache" + "stock-analysis/internal/model" +) + +// resolveCandidates 根据任务类型解析要处理的转债候选列表。 +// +// full: 从 eastmoney datacenter 获取全量转债列表，过滤 ListDate 为空的 +// incremental: 从 SQLite 获取已入库的转债代码列表 +func resolveCandidates(ctx context.Context, store Store, lister BondLister, typ cache.TaskType, limit int) ([]model.ConvertibleBond, error) { + switch typ { + case cache.TaskTypeFull: + bonds, err := lister.FetchAllBondCodes(ctx) + if err != nil { + return nil, fmt.Errorf("获取转债列表: %w", err) + } + var candidates []model.ConvertibleBond + for _, b := range bonds { + if b.ListDate == "" { + continue // 过滤未上市 + } + candidates = append(candidates, b) + } + if limit > 0 && limit < len(candidates) { + candidates = candidates[:limit] + } + return candidates, nil + + case cache.TaskTypeIncremental: + codes, err := store.ListCodes(ctx) + if err != nil { + return nil, fmt.Errorf("加载已入库转债: %w", err) + } + candidates := make([]model.ConvertibleBond, 0, len(codes)) + for i, c := range codes { + if limit > 0 && i >= limit { + break + } + candidates = append(candidates, model.ConvertibleBond{Code: c}) + } + return candidates, nil + + default: + return nil, fmt.Errorf("未知任务类型: %s", typ) + } +} + +// yearOf 从上市日期（2006-01-02）解析年份。 +func yearOf(listDate string) int { + if len(listDate) >= 4 { + var y int + if _, err := fmt.Sscanf(listDate[:4], "%d", &y); err == nil && y > 2000 { + return y + } + } + return time.Now().Year() +} + +// resolveIncrementalBeg 确定增量拉取的起始日期（YYYYMMDD）。 +// +// since 非空 → 直接使用（覆盖库中最后日期） +// 否则 → lastDateMap[code] 次日；无记录返回空串（表示跳过） +func resolveIncrementalBeg(code, since string, lastDateMap map[string]string) (string, bool) { + if since != "" { + return since, true + } + lastDate, ok := lastDateMap[code] + if !ok || lastDate == "" { + return "", false + } + t, err := time.Parse("2006-01-02", lastDate) + if err != nil { + return "", false + } + return t.AddDate(0, 0, 1).Format("20060102"), true +} diff --git a/internal/job/job.go b/internal/job/job.go new file mode 100644 index 0000000..b0465f3 --- /dev/null +++ b/internal/job/job.go @@ -0,0 +1,187 @@ +package job + +import ( + "context" + "encoding/json" + "log" + "sync" + "time" + + "github.com/google/uuid" + + "stock-analysis/internal/cache" + "stock-analysis/internal/datasource" +) + +// Config Manager 配置。 +type Config struct { + Workers int // 并发 worker 数，默认 8 + BatchSize int // 批量写入阈值，默认 20 + RateInterval time.Duration // 全局限速间隔，默认 400ms（≈2.5 req/s） + RateBurst int // 限速突发额度，默认 4 +} + +// runningTask 记录运行中任务的取消句柄和状态快照。 +type runningTask struct { + cancel context.CancelFunc + task *cache.Task +} + +// Manager 管理任务注册表和全局限速器，允许多任务并发运行。 +type Manager struct { + store Store + cache cache.Cache + ds datasource.MarketDataSource + lister BondLister + gate <-chan struct{} // 全局限速令牌（跨任务共享） + workers int + batchSize int + + mu sync.RWMutex + running map[string]*runningTask +} + +// NewManager 创建任务管理器，内部启动全局限速器。 +func NewManager(store Store, c cache.Cache, ds datasource.MarketDataSource, lister BondLister, cfg Config) *Manager { + if cfg.Workers <= 0 { + cfg.Workers = 8 + } + if cfg.BatchSize <= 0 { + cfg.BatchSize = 20 + } + if cfg.RateInterval <= 0 { + cfg.RateInterval = 400 * time.Millisecond + } + if cfg.RateBurst <= 0 { + cfg.RateBurst = 4 + } + + return &Manager{ + store: store, + cache: c, + ds: ds, + lister: lister, + gate: newRateLimiter(cfg.RateInterval, cfg.RateBurst), + workers: cfg.Workers, + batchSize: cfg.BatchSize, + running: make(map[string]*runningTask), + } +} + +// Start 创建并异步启动一个任务，立即返回任务快照（pending 状态）。 +func (m *Manager) Start(ctx context.Context, typ cache.TaskType, since string, limit int) (*cache.Task, error) { + params, _ := json.Marshal(map[string]any{"since": since, "limit": limit}) + + task := &cache.Task{ + ID: uuid.New().String(), + Type: typ, + Status: cache.TaskStatusPending, + StartedAt: time.Now().Format(time.RFC3339), + Params: string(params), + } + + // 持久化到 SQLite + Redis + if err := m.store.SaveTask(ctx, task); err != nil { + return nil, err + } + _ = m.cache.SetTask(ctx, task, 24*time.Hour) + _ = m.cache.AddActiveTask(ctx, task.ID) + + // 创建可取消 context + taskCtx, cancel := context.WithCancel(context.Background()) + + m.mu.Lock() + m.running[task.ID] = &runningTask{cancel: cancel, task: task} + m.mu.Unlock() + + // 异步执行 + runner := NewRunner(m.store, m.cache, m.ds, m.gate, m.workers, m.batchSize) + go func() { + defer cancel() + runner.run(taskCtx, task, m.lister, since, limit) + // 任务结束后从 running map 移除 + m.mu.Lock() + delete(m.running, task.ID) + m.mu.Unlock() + }() + + log.Printf("[job] task %s created: type=%s, since=%s, limit=%d", task.ID, typ, since, limit) + return task, nil +} + +// Cancel 取消运行中任务，返回是否成功取消。 +func (m *Manager) Cancel(id string) bool { + m.mu.RLock() + rt, ok := m.running[id] + m.mu.RUnlock() + if !ok { + return false + } + rt.cancel() + log.Printf("[job] task %s cancelled", id) + return true +} + +// Get 获取任务状态：Redis 优先（实时），回退 SQLite。 +func (m *Manager) Get(ctx context.Context, id string) (*cache.Task, error) { + // 1. 先查 Redis（实时状态） + t, err := m.cache.GetTask(ctx, id) + if err == nil && t != nil { + return t, nil + } + // 2. 回退 SQLite + return m.store.GetTask(ctx, id) +} + +// List 返回任务列表：SQLite 历史记录 + Redis 实时覆盖运行中任务。 +func (m *Manager) List(ctx context.Context, limit int) ([]*cache.Task, error) { + tasks, err := m.store.ListTasks(ctx, limit) + if err != nil { + return nil, err + } + // 对运行中任务用 Redis 实时数据覆盖 + for _, t := range tasks { + if t.Status == cache.TaskStatusRunning || t.Status == cache.TaskStatusPending { + if rt, _ := m.cache.GetTask(ctx, t.ID); rt != nil { + t.Done = rt.Done + t.Success = rt.Success + t.Failed = rt.Failed + t.Status = rt.Status + t.FailedCodes = rt.FailedCodes + } + } + } + return tasks, nil +} + +// StopAll 优雅停止所有运行中任务（服务关闭时调用）。 +func (m *Manager) StopAll() { + m.mu.Lock() + defer m.mu.Unlock() + for id, rt := range m.running { + rt.cancel() + log.Printf("[job] stopping task %s on shutdown", id) + } +} + +// newRateLimiter 创建全局限速器（令牌桶模式）。 +// 返回一个只读 channel，每次接收表示获得一个请求令牌。 +// 稳态速率 = 1 token / interval，初始突发 = burst。 +func newRateLimiter(interval time.Duration, burst int) <-chan struct{} { + ch := make(chan struct{}, burst) + // 预填突发额度 + for i := 0; i < burst; i++ { + ch <- struct{}{} + } + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for range ticker.C { + select { + case ch <- struct{}{}: + default: // 满了丢弃，保持稳态速率 + } + } + }() + return ch +} diff --git a/internal/job/runner.go b/internal/job/runner.go new file mode 100644 index 0000000..df13717 --- /dev/null +++ b/internal/job/runner.go @@ -0,0 +1,229 @@ +package job + +import ( + "context" + "fmt" + "log" + "sync" + "time" + + "stock-analysis/internal/cache" + "stock-analysis/internal/datasource" + "stock-analysis/internal/model" +) + +// fetchResult 单只转债拉取结果，通过 channel 发往 collector。 +type fetchResult struct { + code string + klines []model.Kline + err error +} + +// Runner 单任务执行器：N worker 并发拉取 + 单 collector 串行批量写入。 +// +// 并发模型： +// - 主 goroutine 遍历候选列表，为每只转债启动一个 worker goroutine +// - worker 从 gate 获取全局限速令牌后拉取数据，结果发往 resultCh +// - collector 单 goroutine 从 resultCh 收集，每 batchSize 只批量写 SQLite +// - SQLite SetMaxOpenConns(1) 下，只有 collector 写入，避免锁竞争 +type Runner struct { + store Store + cache cache.Cache + ds datasource.MarketDataSource + gate <-chan struct{} // 全局限速令牌 + workers int + batchSize int +} + +// NewRunner 创建执行器。 +func NewRunner(store Store, c cache.Cache, ds datasource.MarketDataSource, gate <-chan struct{}, workers, batchSize int) *Runner { + return &Runner{ + store: store, + cache: c, + ds: ds, + gate: gate, + workers: workers, + batchSize: batchSize, + } +} + +// run 执行任务（异步调用，不阻塞调用方）。 +func (r *Runner) run(ctx context.Context, task *cache.Task, lister BondLister, since string, limit int) { + // 1. 解析候选列表 + candidates, err := resolveCandidates(ctx, r.store, lister, task.Type, limit) + if err != nil { + r.failTask(ctx, task, err) + return + } + task.Total = len(candidates) + + // 2. 预加载：全量模式加载已入库集合（跳过断点续跑）；增量模式加载最后日期映射 + var skipSet map[string]bool + var lastDateMap map[string]string + if task.Type == cache.TaskTypeFull { + codes, err := r.store.ListCodes(ctx) + if err == nil { + skipSet = make(map[string]bool, len(codes)) + for _, c := range codes { + skipSet[c] = true + } + } + } else { + lastDateMap, _ = r.store.LastKlineDates(ctx) + } + + // 3. 更新状态为 running + task.Status = cache.TaskStatusRunning + r.reportProgress(ctx, task, true) + + log.Printf("[job] task %s started: type=%s, total=%d", task.ID, task.Type, task.Total) + + // 4. 启动 collector（唯一 SQLite 写者） + resultCh := make(chan fetchResult, r.workers*2) + collectorDone := make(chan struct{}) + go r.collect(ctx, task, resultCh, collectorDone) + + // 5. 并发拉取：主 goroutine 遍历候选，每只启动一个 worker + sem := make(chan struct{}, r.workers) + var wg sync.WaitGroup + for _, b := range candidates { + if ctx.Err() != nil { + break + } + // 全量模式跳过已入库转债 + if task.Type == cache.TaskTypeFull && skipSet[b.Code] { + task.Done++ + task.Success++ + r.reportProgress(ctx, task, false) + continue + } + sem <- struct{}{} + wg.Add(1) + go func(bond model.ConvertibleBond) { + defer wg.Done() + defer func() { <-sem }() + + // 全局限速：获取令牌后才允许发请求 + select { + case <-r.gate: + case <-ctx.Done(): + return + } + + klines, err := r.fetchOne(ctx, task.Type, bond, since, lastDateMap) + select { + case resultCh <- fetchResult{code: bond.Code, klines: klines, err: err}: + case <-ctx.Done(): + } + }(b) + } + + // 6. 等待所有 worker 完成 + wg.Wait() + close(resultCh) + <-collectorDone + + // 7. 确定终态 + r.finishTask(ctx, task) +} + +// fetchOne 拉取单只转债K线。 +func (r *Runner) fetchOne(ctx context.Context, typ cache.TaskType, bond model.ConvertibleBond, since string, lastDateMap map[string]string) ([]model.Kline, error) { + switch typ { + case cache.TaskTypeFull: + beg := fmt.Sprintf("%04d0101", yearOf(bond.ListDate)) + return r.ds.FetchKline(ctx, bond.Code, "101", beg, "20500101") + case cache.TaskTypeIncremental: + beg, ok := resolveIncrementalBeg(bond.Code, since, lastDateMap) + if !ok { + return nil, nil // 无存档，跳过 + } + return r.ds.FetchKline(ctx, bond.Code, "101", beg, "20500101") + default: + return nil, fmt.Errorf("未知任务类型: %s", typ) + } +} + +// collect 单 goroutine 收集结果并批量写入 SQLite（唯一写者）。 +func (r *Runner) collect(ctx context.Context, task *cache.Task, resultCh <-chan fetchResult, done chan<- struct{}) { + defer close(done) + batch := make(map[string][]model.Kline) + + for res := range resultCh { + task.Done++ + if res.err != nil || len(res.klines) == 0 { + task.Failed++ + if res.err != nil { + task.FailedCodes = append(task.FailedCodes, res.code) + } + } else { + task.Success++ + batch[res.code] = res.klines + if len(batch) >= r.batchSize { + if err := r.store.SaveKlinesBatch(ctx, batch); err != nil { + log.Printf("[job] batch save error: %v", err) + } + batch = make(map[string][]model.Kline) + } + } + r.reportProgress(ctx, task, false) + } + + // flush 残余 batch + if len(batch) > 0 { + if err := r.store.SaveKlinesBatch(ctx, batch); err != nil { + log.Printf("[job] final batch save error: %v", err) + } + } +} + +// reportProgress 上报进度：Redis 每次（廉价），SQLite 按策略写（减少写压力）。 +// force=true（状态变更）时用 SaveTask 写入完整状态（含 status/total）， +// 确保 Redis 不可用时 SQLite 也能反映运行状态。 +func (r *Runner) reportProgress(ctx context.Context, task *cache.Task, force bool) { + _ = r.cache.SetTask(ctx, task, 24*time.Hour) + if force { + _ = r.store.SaveTask(ctx, task) + } else if task.Done%5 == 0 { + _ = r.store.UpdateTaskProgress(ctx, task.ID, task.Total, task.Done, task.Success, task.Failed, task.FailedCodes) + } +} + +// finishTask 确定任务终态并持久化。 +// 使用 SaveTask 写入完整任务状态（含 total/done/success/failed）， +// 确保 SQLite 中最终记录与内存一致（Redis 不可用时尤为重要）。 +func (r *Runner) finishTask(ctx context.Context, task *cache.Task) { + var errMsg string + switch { + case ctx.Err() != nil: + task.Status = cache.TaskStatusCancelled + case task.Success == 0 && task.Failed > 0: + task.Status = cache.TaskStatusFailed + errMsg = fmt.Sprintf("全部 %d 只失败", task.Failed) + default: + task.Status = cache.TaskStatusSuccess + if task.Failed > 0 { + errMsg = fmt.Sprintf("成功 %d，失败 %d", task.Success, task.Failed) + } + } + task.Error = errMsg + task.FinishedAt = time.Now().Format(time.RFC3339) + + _ = r.store.SaveTask(ctx, task) + _ = r.cache.SetTask(ctx, task, 24*time.Hour) + _ = r.cache.RemoveActiveTask(ctx, task.ID) + + log.Printf("[job] task %s finished: status=%s, total=%d, success=%d, failed=%d", + task.ID, task.Status, task.Total, task.Success, task.Failed) +} + +// failTask 标记任务失败（启动阶段错误）。 +func (r *Runner) failTask(ctx context.Context, task *cache.Task, err error) { + task.Status = cache.TaskStatusFailed + task.Error = err.Error() + task.FinishedAt = time.Now().Format(time.RFC3339) + _ = r.store.SaveTask(ctx, task) + _ = r.cache.SetTask(ctx, task, 24*time.Hour) + _ = r.cache.RemoveActiveTask(ctx, task.ID) + log.Printf("[job] task %s failed: %v", task.ID, err) +} diff --git a/internal/job/store.go b/internal/job/store.go new file mode 100644 index 0000000..93428bc --- /dev/null +++ b/internal/job/store.go @@ -0,0 +1,37 @@ +// Package job 实现异步任务执行系统：并发拉取 + 批量写入 + 进度上报。 +// +// 核心设计： +// - Manager 管理任务注册表和全局限速器，允许多任务并发 +// - Runner 是单任务执行器：N worker 并发拉取，单 collector 串行写 SQLite +// - Redis 存实时进度，SQLite 存持久化任务记录 +package job + +import ( + "context" + + "stock-analysis/internal/cache" + "stock-analysis/internal/model" +) + +// Store 定义 job 运行期需要的存储能力（K线 + 任务 CRUD）。 +// *sqlite.Store 实现此接口（编译期断言）。 +type Store interface { + // K线批量写入（单事务，collector 调用） + SaveKlinesBatch(ctx context.Context, batch map[string][]model.Kline) error + // 返回所有已入库 code（全量断点续跑跳过判断） + ListCodes(ctx context.Context) ([]string, error) + // 一次性返回 code -> 最后K线日期 映射（增量起点预加载） + LastKlineDates(ctx context.Context) (map[string]string, error) + + // 任务持久化 + SaveTask(ctx context.Context, t *cache.Task) error + UpdateTaskProgress(ctx context.Context, id string, total, done, success, failed int, failedCodes []string) error + FinishTask(ctx context.Context, id string, status cache.TaskStatus, errMsg, finishedAt string) error + GetTask(ctx context.Context, id string) (*cache.Task, error) + ListTasks(ctx context.Context, limit int) ([]*cache.Task, error) +} + +// BondLister 全量转债列表获取接口（eastmoney.Client 实现）。 +type BondLister interface { + FetchAllBondCodes(ctx context.Context) ([]model.ConvertibleBond, error) +} diff --git a/internal/router/router.go b/internal/router/router.go index c381abb..750f383 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -12,7 +12,7 @@ import ( // New 构建应用路由处理器。 // webDir 为前端静态资源目录（View 层）。 -func New(mc *controller.MarketController, ac *controller.AuthController, webDir string) http.Handler { +func New(mc *controller.MarketController, ac *controller.AuthController, tc *controller.TaskController, webDir string) http.Handler { mux := http.NewServeMux() // 公开接口：无需登录。 @@ -29,6 +29,12 @@ func New(mc *controller.MarketController, ac *controller.AuthController, webDir mux.HandleFunc("GET /api/status", requireAuth(ac, mc.GetDataStatus)) mux.HandleFunc("GET /api/db/bonds", requireAuth(ac, mc.ListDBBonds)) + // 数据入库任务接口：创建/查询/取消（需要登录）。 + mux.HandleFunc("POST /api/tasks", requireAuth(ac, tc.CreateTask)) + mux.HandleFunc("GET /api/tasks", requireAuth(ac, tc.ListTasks)) + mux.HandleFunc("GET /api/tasks/{id}", requireAuth(ac, tc.GetTask)) + mux.HandleFunc("POST /api/tasks/{id}/cancel", requireAuth(ac, tc.CancelTask)) + // 静态资源：登录页及样式脚本公开，首页受保护。 fs := noCache(http.FileServer(http.Dir(webDir))) mux.Handle("/auth.html", fs) diff --git a/internal/storage/sqlite/store.go b/internal/storage/sqlite/store.go index 4b7344d..1102035 100644 --- a/internal/storage/sqlite/store.go +++ b/internal/storage/sqlite/store.go @@ -80,6 +80,23 @@ CREATE TABLE IF NOT EXISTS convert_price_history ( created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_verify_code_email ON verify_code (email, purpose); + + CREATE TABLE IF NOT EXISTS ingest_task ( + id TEXT PRIMARY KEY, + type TEXT NOT NULL, + status TEXT NOT NULL, + total INTEGER NOT NULL DEFAULT 0, + done INTEGER NOT NULL DEFAULT 0, + success INTEGER NOT NULL DEFAULT 0, + failed INTEGER NOT NULL DEFAULT 0, + failed_codes TEXT NOT NULL DEFAULT '', + params TEXT NOT NULL DEFAULT '', + error TEXT NOT NULL DEFAULT '', + started_at TEXT NOT NULL, + finished_at TEXT NOT NULL DEFAULT '' + ); + CREATE INDEX IF NOT EXISTS idx_ingest_task_status ON ingest_task (status); + CREATE INDEX IF NOT EXISTS idx_ingest_task_started ON ingest_task (started_at); ` // 通用 SQL：幂等写入（存在则跳过）。 @@ -90,8 +107,8 @@ ON CONFLICT (code, date) DO NOTHING; ` const ( - querySQL = `SELECT code, date, open, close, high, low, volume, amount, convert_price, convert_value, premium_rate FROM convertible_kline WHERE code = ? ORDER BY date;` - countSQL = `SELECT COUNT(*) FROM convertible_kline WHERE code = ?;` + querySQL = `SELECT code, date, open, close, high, low, volume, amount, convert_price, convert_value, premium_rate FROM convertible_kline WHERE code = ? ORDER BY date;` + countSQL = `SELECT COUNT(*) FROM convertible_kline WHERE code = ?;` lastDateSQL = `SELECT MAX(date) FROM convertible_kline WHERE code = ?;` allKlineSQL = `SELECT code, date, open, close, high, low, volume, amount, convert_price, convert_value, premium_rate FROM convertible_kline ORDER BY code, date;` @@ -148,9 +165,13 @@ type PremiumRow struct { // Store 是 SQLite 存储实现，实现 storage.KlineStore。 type Store struct { - db *sql.DB + db *sql.DB + path string } +// Path 返回数据库文件路径。 +func (s *Store) Path() string { return s.path } + // Open 打开（或创建）SQLite 数据库文件并初始化表结构。 func Open(path string) (*Store, error) { db, err := sql.Open("sqlite", path) @@ -187,7 +208,7 @@ func Open(path string) (*Store, error) { db.Close() return nil, fmt.Errorf("set busy timeout: %w", err) } - return &Store{db: db}, nil + return &Store{db: db, path: path}, nil } // initSchema 执行建表语句。 @@ -354,6 +375,74 @@ func (s *Store) ListCodes(ctx context.Context) ([]string, error) { return codes, rows.Err() } +// LastKlineDates 一次性返回所有已入库 code 的最新K线日期（YYYY-MM-DD）。 +// 用于增量同步预加载，替代逐只 LastKlineDate 查询，避免占用唯一写连接。 +func (s *Store) LastKlineDates(ctx context.Context) (map[string]string, error) { + rows, err := s.db.QueryContext(ctx, "SELECT code, MAX(date) FROM convertible_kline GROUP BY code") + if err != nil { + return nil, fmt.Errorf("last kline dates: %w", err) + } + defer rows.Close() + + out := make(map[string]string) + for rows.Next() { + var code, date string + if err := rows.Scan(&code, &date); err != nil { + return nil, fmt.Errorf("scan last date: %w", err) + } + out[code] = date + } + return out, rows.Err() +} + +// SaveKlinesBatch 单事务批量写入多只转债K线（collector 调用）。 +// 接收 map[转债代码]K线切片，在一个事务内遍历写入，带 3 次重试。 +func (s *Store) SaveKlinesBatch(ctx context.Context, batch map[string][]model.Kline) error { + if len(batch) == 0 { + return nil + } + const maxAttempt = 3 + var lastErr error + for attempt := 0; attempt < maxAttempt; attempt++ { + if attempt > 0 { + time.Sleep(2 * time.Second) + } + if err := s.saveBatchOnce(ctx, batch); err != nil { + lastErr = err + continue + } + return nil + } + return lastErr +} + +// saveBatchOnce 单次执行批量写入事务。 +func (s *Store) saveBatchOnce(ctx context.Context, batch map[string][]model.Kline) error { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin tx: %w", err) + } + defer tx.Rollback() //nolint:errcheck + + stmt, err := tx.PrepareContext(ctx, upsertSQL) + if err != nil { + return fmt.Errorf("prepare upsert: %w", err) + } + defer stmt.Close() + + for code, klines := range batch { + for _, k := range klines { + if _, err := stmt.ExecContext(ctx, code, k.Date, k.Open, k.Close, k.High, k.Low, k.Volume, k.Amount, k.ConvertPrice, k.ConvertValue, k.PremiumRate); err != nil { + return fmt.Errorf("insert kline %s %s: %w", code, k.Date, err) + } + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit tx: %w", err) + } + return nil +} + // LoadAllKlines 一次性加载全部标的的K线（按代码、日期升序）。 // 用于回测等需要全量数据在内存的场景。 func (s *Store) LoadAllKlines(ctx context.Context) (map[string][]model.Kline, error) { diff --git a/internal/storage/sqlite/task.go b/internal/storage/sqlite/task.go new file mode 100644 index 0000000..dbf7d2b --- /dev/null +++ b/internal/storage/sqlite/task.go @@ -0,0 +1,126 @@ +package sqlite + +import ( + "context" + "encoding/json" + "fmt" + + "stock-analysis/internal/cache" +) + +// 任务相关 SQL。 +const ( + saveTaskSQL = ` +INSERT INTO ingest_task (id, type, status, total, done, success, failed, failed_codes, params, error, started_at, finished_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT (id) DO UPDATE SET + type = excluded.type, + status = excluded.status, + total = excluded.total, + done = excluded.done, + success = excluded.success, + failed = excluded.failed, + failed_codes = excluded.failed_codes, + params = excluded.params, + error = excluded.error, + started_at = excluded.started_at, + finished_at = excluded.finished_at;` + + updateTaskProgressSQL = `UPDATE ingest_task SET total = ?, done = ?, success = ?, failed = ?, failed_codes = ? WHERE id = ?;` + + finishTaskSQL = `UPDATE ingest_task SET status = ?, error = ?, finished_at = ? WHERE id = ?;` + + getTaskSQL = `SELECT id, type, status, total, done, success, failed, failed_codes, params, error, started_at, finished_at FROM ingest_task WHERE id = ?;` + + listTasksSQL = `SELECT id, type, status, total, done, success, failed, failed_codes, params, error, started_at, finished_at FROM ingest_task ORDER BY started_at DESC LIMIT ?;` +) + +// SaveTask 保存或更新任务（INSERT OR REPLACE 语义）。 +func (s *Store) SaveTask(ctx context.Context, t *cache.Task) error { + failedCodes, err := json.Marshal(t.FailedCodes) + if err != nil { + return fmt.Errorf("marshal failed codes: %w", err) + } + _, err = s.db.ExecContext(ctx, saveTaskSQL, + t.ID, t.Type, t.Status, t.Total, t.Done, t.Success, t.Failed, + string(failedCodes), t.Params, t.Error, t.StartedAt, t.FinishedAt) + if err != nil { + return fmt.Errorf("save task %s: %w", t.ID, err) + } + return nil +} + +// UpdateTaskProgress 更新任务进度（total/done/success/failed/failed_codes）。 +func (s *Store) UpdateTaskProgress(ctx context.Context, id string, total, done, success, failed int, failedCodes []string) error { + data, err := json.Marshal(failedCodes) + if err != nil { + return fmt.Errorf("marshal failed codes: %w", err) + } + _, err = s.db.ExecContext(ctx, updateTaskProgressSQL, total, done, success, failed, string(data), id) + if err != nil { + return fmt.Errorf("update task progress %s: %w", id, err) + } + return nil +} + +// FinishTask 标记任务终态（status/error/finished_at）。 +func (s *Store) FinishTask(ctx context.Context, id string, status cache.TaskStatus, errMsg, finishedAt string) error { + _, err := s.db.ExecContext(ctx, finishTaskSQL, status, errMsg, finishedAt, id) + if err != nil { + return fmt.Errorf("finish task %s: %w", id, err) + } + return nil +} + +// GetTask 按 ID 查询任务。 +func (s *Store) GetTask(ctx context.Context, id string) (*cache.Task, error) { + row := s.db.QueryRowContext(ctx, getTaskSQL, id) + t, err := scanTask(row) + if err != nil { + return nil, fmt.Errorf("get task %s: %w", id, err) + } + return t, nil +} + +// ListTasks 返回最近的任务列表（按开始时间降序）。 +func (s *Store) ListTasks(ctx context.Context, limit int) ([]*cache.Task, error) { + if limit <= 0 { + limit = 50 + } + rows, err := s.db.QueryContext(ctx, listTasksSQL, limit) + if err != nil { + return nil, fmt.Errorf("list tasks: %w", err) + } + defer rows.Close() + + var tasks []*cache.Task + for rows.Next() { + t, err := scanTask(rows) + if err != nil { + return nil, err + } + tasks = append(tasks, t) + } + return tasks, rows.Err() +} + +// rowScanner 抽象 *sql.Row 和 *sql.Rows 的 Scan 方法。 +type rowScanner interface { + Scan(dest ...any) error +} + +// scanTask 从查询行扫描一条任务记录。 +func scanTask(row rowScanner) (*cache.Task, error) { + var ( + t cache.Task + failedCodes string + ) + if err := row.Scan(&t.ID, &t.Type, &t.Status, &t.Total, &t.Done, &t.Success, &t.Failed, + &failedCodes, &t.Params, &t.Error, &t.StartedAt, &t.FinishedAt); err != nil { + return nil, err + } + if failedCodes != "" && failedCodes != "null" { + _ = json.Unmarshal([]byte(failedCodes), &t.FailedCodes) + } + return &t, nil +} diff --git a/web/index.html b/web/index.html index 2d6d0ec..8396194 100644 --- a/web/index.html +++ b/web/index.html @@ -308,6 +308,64 @@ font-size: 13px; } + /* ============ Task Status Badges ============ */ + .task-badge { + display: inline-block; + padding: 3px 10px; + border-radius: 12px; + font-size: 11px; + font-weight: 600; + letter-spacing: 0.3px; + white-space: nowrap; + } + .task-badge.pending { background: #FFF3CD; color: #856404; } + .task-badge.running { background: #D6E4FF; color: #1E5FBB; } + .task-badge.success { background: #D4EDDA; color: #155724; } + .task-badge.failed { background: #F8D7DA; color: #721C24; } + .task-badge.cancelled { background: #E2E3E5; color: #383D41; } + .task-badge.running::before { + content: ''; + display: inline-block; + width: 6px; + height: 6px; + border-radius: 50%; + background: #1E5FBB; + margin-right: 5px; + animation: pulse 1s ease-in-out infinite; + } + @keyframes pulse { + 0%, 100% { opacity: 1; } + 50% { opacity: 0.3; } + } + .progress-bar { + display: inline-block; + width: 80px; + height: 6px; + background: var(--gray-200); + border-radius: 3px; + vertical-align: middle; + margin: 0 6px; + overflow: hidden; + } + .progress-bar-fill { + height: 100%; + background: var(--blue); + border-radius: 3px; + transition: width 0.3s ease; + } + .btn-cancel { + padding: 4px 10px; + border: 1px solid var(--red); + background: transparent; + color: var(--red); + border-radius: var(--radius-sm); + font-size: 12px; + cursor: pointer; + transition: all var(--transition); + } + .btn-cancel:hover { background: var(--red); color: var(--white); } + .btn-cancel:disabled { opacity: 0.4; cursor: not-allowed; } + /* ============ Responsive ============ */ @media (max-width: 768px) { .header { padding: 16px 20px; } @@ -484,6 +542,59 @@
+ + +
+
+ 数据更新任务 + +
+
+
+ + + + + +
+
+ 全量更新：拉取所有转债的历史K线（已入库的自动跳过）
+ 增量更新：仅同步已入库转债的最新数据（自上次入库日期起） +
+
+
+ + +
+
+ 任务列表 + +
+
+
+ + +
+
+ + + + + + + + + + + + + + + +
任务ID类型状态进度成功失败开始时间完成时间操作
+
+
+
@@ -559,7 +670,7 @@ document.querySelectorAll('.tab-content').forEach(c => c.classList.remove('active')); tab.classList.add('active'); document.getElementById('tab-' + tab.dataset.tab).classList.add('active'); - if (tab.dataset.tab === 'data') loadDataStatus(); + if (tab.dataset.tab === 'data') { loadDataStatus(); loadTasks(); } }); }); @@ -956,6 +1067,170 @@ document.getElementById('data-refresh').addEventListener('click', loadDataStatus); + // ============ 数据更新任务 ============ + let taskPollTimer = null; + + // 任务类型中文映射 + const taskTypeLabel = { full: '全量更新', incremental: '增量更新' }; + // 任务状态中文映射 + const taskStatusLabel = { + pending: '等待中', running: '运行中', success: '成功', failed: '失败', cancelled: '已取消' + }; + + // 创建任务 + async function createTask(type) { + const statusEl = document.getElementById('task-create-status'); + const btnId = type === 'full' ? 'task-full' : 'task-incremental'; + const btn = document.getElementById(btnId); + const since = document.getElementById('task-since').value; + const limit = +document.getElementById('task-limit').value || 0; + + btn.disabled = true; + statusEl.textContent = '提交中...'; + statusEl.style.color = 'var(--blue)'; + try { + const body = { type, since, limit }; + const resp = await apiFetch('/api/tasks', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(body), + }); + if (!resp.ok) { + const err = await resp.json().catch(() => ({})); + throw new Error(err.error || 'HTTP ' + resp.status); + } + const task = await resp.json(); + showToast(`${taskTypeLabel[type]}任务已创建`, 'success'); + statusEl.textContent = ''; + // 立即刷新任务列表并启动轮询 + await loadTasks(); + startTaskPolling(); + // 切换到任务列表区域 + document.querySelector('[data-section="task-list"] .section-header').scrollIntoView({ behavior: 'smooth', block: 'start' }); + } catch (e) { + showToast('创建任务失败: ' + e.message, 'error'); + statusEl.textContent = '× ' + e.message; + statusEl.style.color = 'var(--red)'; + } finally { + btn.disabled = false; + } + } + + // 加载任务列表 + async function loadTasks() { + const statusEl = document.getElementById('task-list-status'); + const tbody = document.querySelector('#task-table tbody'); + statusEl.textContent = '加载中...'; + try { + const resp = await apiFetch('/api/tasks?limit=50'); + if (!resp.ok) throw new Error('HTTP ' + resp.status); + const tasks = await resp.json(); + renderTasks(tasks || []); + const running = (tasks || []).filter(t => t.status === 'running' || t.status === 'pending').length; + statusEl.textContent = `共 ${tasks.length} 条${running > 0 ? `（${running} 个运行中）` : ''}`; + // 有运行中任务时继续轮询，否则停止 + if (running > 0) { + startTaskPolling(); + } else { + stopTaskPolling(); + } + } catch (e) { + tbody.innerHTML = `加载失败: ${e.message}`; + statusEl.textContent = '加载失败'; + } + } + + // 渲染任务列表 + function renderTasks(tasks) { + const tbody = document.querySelector('#task-table tbody'); + if (tasks.length === 0) { + tbody.innerHTML = '暂无任务记录'; + return; + } + tbody.innerHTML = tasks.map(t => { + const pct = t.total > 0 ? Math.round(t.done / t.total * 100) : 0; + const isActive = t.status === 'running' || t.status === 'pending'; + const statusText = taskStatusLabel[t.status] || t.status; + const shortId = t.id.substring(0, 8); + const started = formatTime(t.startedAt); + const finished = t.finishedAt ? formatTime(t.finishedAt) : '-'; + const failInfo = t.failed > 0 && t.failedCodes && t.failedCodes.length > 0 + ? ` title="失败代码: ${t.failedCodes.slice(0, 10).join(', ')}${t.failedCodes.length > 10 ? '...' : ''}"` + : ''; + const errorInfo = t.error ? ` title="${t.error.replace(/"/g, '"')}"` : ''; + return ` + ${shortId} + ${taskTypeLabel[t.type] || t.type} + ${statusText} + + ${t.done}/${t.total} + + ${pct}% + + ${t.success} + ${t.failed} + ${started} + ${finished} + ${isActive + ? `` + : '-'} + `; + }).join(''); + } + + // 取消任务 + async function cancelTask(id) { + if (!confirm('确定取消该任务？')) return; + try { + const resp = await apiFetch(`/api/tasks/${id}/cancel`, { method: 'POST' }); + if (!resp.ok) { + const err = await resp.json().catch(() => ({})); + throw new Error(err.error || 'HTTP ' + resp.status); + } + showToast('任务取消请求已发送', 'info'); + await loadTasks(); + } catch (e) { + showToast('取消失败: ' + e.message, 'error'); + } + } + // 暴露到全局供 inline onclick 调用 + window.cancelTask = cancelTask; + + // 格式化时间：RFC3339 → 可读格式 + function formatTime(s) { + if (!s) return '-'; + const d = new Date(s); + if (isNaN(d)) return s; + const pad = n => String(n).padStart(2, '0'); + return `${d.getFullYear()}-${pad(d.getMonth()+1)}-${pad(d.getDate())} ${pad(d.getHours())}:${pad(d.getMinutes())}:${pad(d.getSeconds())}`; + } + + // 轮询：运行中任务时每 3 秒刷新一次 + function startTaskPolling() { + if (taskPollTimer) return; // 已在轮询 + taskPollTimer = setInterval(loadTasks, 3000); + } + function stopTaskPolling() { + if (taskPollTimer) { + clearInterval(taskPollTimer); + taskPollTimer = null; + } + } + + document.getElementById('task-full').addEventListener('click', () => createTask('full')); + document.getElementById('task-incremental').addEventListener('click', () => createTask('incremental')); + document.getElementById('task-refresh').addEventListener('click', loadTasks); + + // 数据管理Tab的折叠/展开 + document.querySelectorAll('#tab-data .section-header').forEach(header => { + header.addEventListener('click', e => { + if (e.target.closest('.section-toggle')) e.stopPropagation(); + const card = header.closest('.section-card'); + const collapsed = card.classList.toggle('collapsed'); + header.querySelector('.section-toggle').textContent = collapsed ? '+' : '−'; + }); + }); + // 初始化 loadBonds();