WarrenZhou
273fc90a95
新增完整的异步任务管理体系： 1. 新增Redis缓存层实现任务状态实时存储与降级机制 2. 实现并发拉取+批量写入的高性能入库流程 3. 添加任务管理HTTP接口与路由 4. 重构行情控制器使用共享SQLite存储 5. 配套添加Docker Compose Redis启动配置 6. 前端新增任务管理面板支持操作与进度查看
83 lines
2.2 KiB
Go
83 lines
2.2 KiB
Go
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
|
||
}
|