kevin/sese-engine-go
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加反链计算按钮

Browse Source
This commit is contained in:
kevin
2026-04-09 13:43:58 +08:00
parent f307e30496
commit 351b9d36bb
7 changed files with 127 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backlink/backlink.go
+73 -8
View File
@@ -13,6 +13,7 @@ import (
"os" // 文件写入 "os" // 文件写入
"path/filepath" // 路径拼接 "path/filepath" // 路径拼接
"strings" // 字符串操作 "strings" // 字符串操作
"sync" // 互斥锁（保护手动触发的并发安全）
"time" // 时间计算（下次运行时间、睡眠） "time" // 时间计算（下次运行时间、睡眠）
"sese-engine/storage" // 持久化存储 "sese-engine/storage" // 持久化存储
@@ -22,6 +23,11 @@ import (
type Runner struct { type Runner struct {
db *storage.DB db *storage.DB
storagePath string // 存储根目录（用于写入 prosper.json） storagePath string // 存储根目录（用于写入 prosper.json）
mu sync.Mutex
running bool // 是否正在计算中
nextRun time.Time // 下次计划执行时间
lastRunAt *time.Time // 上次完成时间（nil 表示尚未运行）
lastError string // 上次错误信息（空字符串表示无错误）
} }
// New 创建一个 Runner 实例。 // New 创建一个 Runner 实例。
@@ -29,6 +35,23 @@ func New(db *storage.DB, storagePath string) *Runner {
return &Runner{db: db, storagePath: storagePath} return &Runner{db: db, storagePath: storagePath}
} }
// Status 返回反链计算器的当前状态快照。
func (r *Runner) Status() map[string]interface{} {
r.mu.Lock()
defer r.mu.Unlock()
m := map[string]interface{}{
"running": r.running,
"next_run": r.nextRun.Format(time.RFC3339),
}
if r.lastRunAt != nil {
m["last_run"] = r.lastRunAt.Format(time.RFC3339)
}
if r.lastError != "" {
m["last_error"] = r.lastError
}
return m
}
// Run 无限循环，每 48 小时执行一次反向链接计算。 // Run 无限循环，每 48 小时执行一次反向链接计算。
// 每次运行对齐到凌晨 2:00（便于在低峰期执行重计算）。 // 每次运行对齐到凌晨 2:00（便于在低峰期执行重计算）。
func (r *Runner) Run() { func (r *Runner) Run() {
@@ -39,22 +62,64 @@ func (r *Runner) Run() {
if !target.After(now) { if !target.After(now) {
target = target.Add(48 * time.Hour) // 已过凌晨 2 点，则等明天的 2 点 target = target.Add(48 * time.Hour) // 已过凌晨 2 点，则等明天的 2 点
} }
r.mu.Lock()
r.nextRun = target
r.mu.Unlock()
sleep := target.Sub(now) sleep := target.Sub(now)
log.Printf("[backlink] next run at %v (in %v)", target.Format(time.RFC3339), sleep.Round(time.Minute)) log.Printf("[backlink] next run at %v (in %v)", target.Format(time.RFC3339), sleep.Round(time.Minute))
time.Sleep(sleep) time.Sleep(sleep)
log.Printf("[backlink] starting computation at %v", time.Now().Format(time.RFC3339)) r.doCompute()
if err := r.compute(); err != nil {
log.Printf("[backlink] error: %v", err)
} else {
log.Printf("[backlink] done")
}
} }
} }
// RunNow 立即执行一次计算（用于手动触发或测试）。 // RunNow 立即执行一次计算（用于手动触发）。
// 如果已有计算正在运行则返回 nil（幂等）。
// 执行后重新计算下次自动运行的定时。
func (r *Runner) RunNow() error { func (r *Runner) RunNow() error {
return r.compute() r.mu.Lock()
if r.running {
r.mu.Unlock()
log.Printf("[backlink] RunNow: already running, skipping")
return nil
}
r.mu.Unlock()
r.doCompute()
return nil
}
// doCompute 执行一次完整的计算，更新状态字段。
func (r *Runner) doCompute() {
r.mu.Lock()
r.running = true
r.lastError = ""
r.mu.Unlock()
log.Printf("[backlink] starting computation at %v", time.Now().Format(time.RFC3339))
err := r.compute()
now := time.Now()
r.mu.Lock()
r.running = false
r.lastRunAt = &now
if err != nil {
r.lastError = err.Error()
}
// 重新计算下次自动运行时间
target := time.Date(now.Year(), now.Month(), now.Day(), 2, 0, 0, 0, now.Location())
if !target.After(now) {
target = target.Add(48 * time.Hour)
}
r.nextRun = target
r.mu.Unlock()
if err != nil {
log.Printf("[backlink] error: %v", err)
} else {
log.Printf("[backlink] done, next run at %v", target.Format(time.RFC3339))
}
} }
// ---- 计算核心 ---- // ---- 计算核心 ----
dist/assets/index-Bbt6GbEM.css
Vendored
-2
View File
File diff suppressed because one or more lines are too long
dist/assets/index-CLWukEE8.css
Vendored
+2
View File
File diff suppressed because one or more lines are too long
dist/assets/index-DXbMW8tX.js → dist/assets/index-w20XarNx.js
Vendored
+1 -1
View File
File diff suppressed because one or more lines are too long
dist/index.html
Vendored
+2 -2
View File
@@ -5,8 +5,8 @@
<link rel="icon" type="image/svg+xml" href="/vite.svg" /> <link rel="icon" type="image/svg+xml" href="/vite.svg" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" /> <meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>SESE 爬取管理</title> <title>SESE 爬取管理</title>
<script type="module" crossorigin src="/assets/index-DXbMW8tX.js"></script> <script type="module" crossorigin src="/assets/index-w20XarNx.js"></script>
<link rel="stylesheet" crossorigin href="/assets/index-Bbt6GbEM.css"> <link rel="stylesheet" crossorigin href="/assets/index-CLWukEE8.css">
</head> </head>
<body> <body>
<div id="app"></div> <div id="app"></div>
main.go
+1
View File
@@ -164,6 +164,7 @@ func main() {
// ---- 6. 反向链接计算器：每 48 小时运行一次 ---- // ---- 6. 反向链接计算器：每 48 小时运行一次 ----
bl := backlink.New(db, *storageDir) bl := backlink.New(db, *storageDir)
searchSrv.SetBacklinkRunner(bl)
go bl.Run() go bl.Run()
// ---- 7. 爬虫：从入口 URL 开始 BFS 爬取 ---- // ---- 7. 爬虫：从入口 URL 开始 BFS 爬取 ----
search/server.go
+48
View File
@@ -46,6 +46,12 @@ type Server struct {
indexCache map[string][]storage.IndexEntry indexCache map[string][]storage.IndexEntry
indexCacheMu sync.RWMutex indexCacheMu sync.RWMutex
indexCacheHits int64 // 缓存命中计数（原子） indexCacheHits int64 // 缓存命中计数（原子）
// backlinkRunner 反向链接计算器（可为 nil，仅用于 admin 手动触发）
backlinkRunner interface {
Status() map[string]interface{}
RunNow() error
}
} }
// New 创建一个 search Server（内嵌收获服务，统一在同一端口）。 // New 创建一个 search Server（内嵌收获服务，统一在同一端口）。
@@ -64,6 +70,14 @@ func New(db *storage.DB, infoSvc *info.Service, a *analyzer.Analyzer) *Server {
return s return s
} }
// SetBacklinkRunner 注入反向链接计算器（用于 admin 手动触发）。
func (s *Server) SetBacklinkRunner(r interface {
Status() map[string]interface{}
RunNow() error
}) {
s.backlinkRunner = r
}
// runPeriodicFlush 每隔 FlushIntervalSeconds 秒触发一次刷盘。 // runPeriodicFlush 每隔 FlushIntervalSeconds 秒触发一次刷盘。
func (s *Server) runPeriodicFlush() { func (s *Server) runPeriodicFlush() {
ticker := time.NewTicker(time.Duration(config.FlushIntervalSeconds()) * time.Second) ticker := time.NewTicker(time.Duration(config.FlushIntervalSeconds()) * time.Second)
@@ -91,6 +105,7 @@ func (s *Server) Handler() http.Handler {
mux.HandleFunc("/admin/flush", s.handleAdminFlush) mux.HandleFunc("/admin/flush", s.handleAdminFlush)
mux.HandleFunc("/admin/pending", s.handleAdminPending) mux.HandleFunc("/admin/pending", s.handleAdminPending)
mux.HandleFunc("/admin/workers", s.handleAdminWorkers) mux.HandleFunc("/admin/workers", s.handleAdminWorkers)
mux.HandleFunc("/admin/backlink", s.handleAdminBacklink)
// 静态文件（SPA fallback） // 静态文件（SPA fallback）
mux.Handle("/", spaHandler{dist: "dist"}) mux.Handle("/", spaHandler{dist: "dist"})
return mux return mux
@@ -458,6 +473,39 @@ func (s *Server) handleAdminWorkers(w http.ResponseWriter, r *http.Request) {
} }
} }
// handleAdminBacklink 查看反链计算状态和手动触发。
// GET 返回 running（是否计算中）、next_run（下次执行时间）、last_run（上次完成时间）
// POST 触发立即执行一次反链计算
func (s *Server) handleAdminBacklink(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Content-Type", "application/json; charset=utf-8")
if s.backlinkRunner == nil {
http.Error(w, `{"error":"backlink runner not available"}`, 503)
return
}
switch r.Method {
case http.MethodOptions:
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Access-Control-Allow-Methods", "GET,POST,OPTIONS")
w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
w.WriteHeader(204)
return
case http.MethodGet:
json.NewEncoder(w).Encode(s.backlinkRunner.Status())
case http.MethodPost:
go s.backlinkRunner.RunNow() // 异步执行，避免阻塞 HTTP 请求
json.NewEncoder(w).Encode(map[string]string{"status": "started"})
default:
w.Header().Set("Allow", "GET,POST")
http.Error(w, `{"error":"method not allowed"}`, 405)
}
}
// ---- 搜索处理器 ---- // ---- 搜索处理器 ----
// searchResponse 是搜索 API 的 JSON 响应结构。 // searchResponse 是搜索 API 的 JSON 响应结构。