From a784db591a7c6816cec20486323e4fb841ad811a Mon Sep 17 00:00:00 2001 From: Color2333 <1552429809@qq.com> Date: Sat, 10 Oct 2026 17:14:23 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=A4=96=E9=83=A8=E5=AE=A1=E6=9F=A5=20P?= =?UTF-8?q?1/P2=20=E4=BF=AE=E5=A4=8D=E2=80=94=E2=80=94R11/R08/R19/R15/R27/?= =?UTF-8?q?R01/R22/R14/R17/R26/R16/R05/R09/R13/R24/R12?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit R11 pdftotext -l 参数改单整数（全文提取实际失效→deep_read 一直降级的根治） R08 暂停旗标 'true'→'1' + upsert（跨进程 claim 判断失效） R19 sensemaking act 字面路由 wrapper 显式传参（PathValue 恒空 404） R15 gatewayChat 统一走 GW() 延迟初始化（冷启动 nil panic） R27 心跳周期固定 200s（续约 600s 的 1/3——长任务 lease 过期被 reclaim） R01 API token 方法级 scope 强制（写方法需 write scope） R22 邮件发送失败返回 error（原 nil error 任务假成功） R14 日报调度尊重 enabled 开关 R17 job 详情补 attempts 数组（监控页 selected.attempts crash） R26 topics/stats 重写为前端契约 topics[{topic_id,topic_name,...,status_dist}] R16 search-multi 双契约（URL query/无 body + papers/channel_stats 响应） R05 批量标签事务化 + paper_tags 补 id/created_at 必填列 + UPDATE 占位符修正 R09 cancel 复用 CoreStore.CancelJob 权威状态机 R13 task_effects 去 task_id 外键（36 位 core UUID 写 32 位旧表必 FK 失败） R24 loadVectors 参数编号/limit 修正 R12 claims 契约对齐 apply（quote 必填/certainty 枚举归一） --- core/brief.go | 5 ++-- core/effects.go | 9 ++++--- core/handlers_paper.go | 29 ++++++++++++++++------ core/runner.go | 8 +++--- core/scheduler.go | 11 ++++++--- server/content.go | 4 +-- server/graph.go | 11 +++------ server/handlers.go | 31 ++++++++++++++++++++++- server/jobs.go | 44 ++++++++++++++++++++++++--------- server/main.go | 12 ++++----- server/phase7.go | 54 ++++++++++++++++++++++++++++++++++++++-- server/sensemaking.go | 26 ++++++++++++++++++-- server/tags.go | 25 +++++++++++++------ server/topics.go | 56 +++++++++++++++++++++++++++++++----------- 14 files changed, 251 insertions(+), 74 deletions(-) diff --git a/core/brief.go b/core/brief.go index b68e0f8..7773dd2 100644 --- a/core/brief.go +++ b/core/brief.go @@ -532,9 +532,8 @@ func HandleSendBriefEmailEffect(ctx context.Context, e *HandlerEnv, task *Task) ).Scan(&htmlContent) } if !SendEmailHTML(e.SMTP, recipient, subject, htmlContent) { - return map[string]any{ - "sent": false, "error": "SMTP 未配置或发送失败（账本未登记，可重试）", "effect_key": effectKey, - }, nil + // R22：发送失败必须以 error 结束任务——nil error 会让任务假成功终结 + return nil, fmt.Errorf("SMTP 未配置或发送失败（账本未登记，可重试）") } RegisterEffect(e.Store, effectKey, "mail_send", task.TaskID) return map[string]any{"sent": true, "effect_key": effectKey}, nil diff --git a/core/effects.go b/core/effects.go index d7080b1..97881a8 100644 --- a/core/effects.go +++ b/core/effects.go @@ -113,14 +113,17 @@ func HasEffect(store *CoreStore, effectKey string) bool { } // RegisterEffect 登记副作用（唯一约束兜底并发；已存在返回 false）。 +// R13：task_effects.task_id 外键指向旧 tasks.id(32)——core_tasks 的 36 位 UUID +// 写入必然 FK 失败且被忽略，导致邮件去重键永不登记。改为不关联旧执行体系， +// 幂等键本身已足够去重。 func RegisterEffect(store *CoreStore, effectKey, kind string, taskID string) bool { if HasEffect(store, effectKey) { return false } _, err := store.DB.Exec( - `INSERT INTO task_effects (id, task_id, effect_key, kind, payload, committed_at) - VALUES ($1, $2, $3, $4, '{}', $5)`, - newCoreID(), nullIfEmpty(taskID), effectKey, kind, nowParam()) + `INSERT INTO task_effects (id, effect_key, kind, payload, committed_at) + VALUES ($1, $2, $3, '{}', $4)`, + newCoreID(), effectKey, kind, nowParam()) return err == nil } diff --git a/core/handlers_paper.go b/core/handlers_paper.go index d712bba..634b6e9 100644 --- a/core/handlers_paper.go +++ b/core/handlers_paper.go @@ -12,6 +12,7 @@ import ( "math" "os" "os/exec" + "strconv" "strings" "time" ) @@ -123,8 +124,7 @@ func extractPDFTextRaw(pdfPath string, maxPages int) string { } ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) defer cancel() - pages := fmt.Sprintf("1-%d", maxPages) - out, err := exec.CommandContext(ctx, pdftotext, "-f", "1", "-l", pages, "-layout", pdfPath, "-").Output() + out, err := exec.CommandContext(ctx, pdftotext, "-f", "1", "-l", strconv.Itoa(maxPages), "-layout", pdfPath, "-").Output() if err != nil { return "" } @@ -483,9 +483,23 @@ func HandleExtractClaims(ctx context.Context, env *HandlerEnv, task *Task) (map[ if !ok { continue } - if strings.TrimSpace(stringOf(m["statement"])) != "" { - items = append(items, m) + if strings.TrimSpace(stringOf(m["statement"])) == "" { + continue + } + // R12：对齐 apply 契约——quote 必填（原文逐字），kind 固定 text_passage， + // certainty 归一到真实枚举（tentative/supported） + norm := map[string]any{ + "statement": stringOf(m["statement"]), + "statement_zh": stringOf(m["statement_zh"]), + "quote": stringOf(m["quote"]), + "kind": "text_passage", } + if cert := stringOf(m["certainty"]); cert == "tentative" || cert == "supported" { + norm["certainty"] = cert + } else { + norm["certainty"] = "tentative" + } + items = append(items, norm) } } trace := map[string]any{ @@ -510,13 +524,12 @@ func HandleExtractClaims(ctx context.Context, env *HandlerEnv, task *Task) (map[ func buildClaimsPrompt(title, text string) string { return "你是研究助理。请从以下论文文本中抽取最多 6 条可验证的核心判断（claims）。" + "输出严格 JSON：\n" + - `{"claims":[{"statement":"英文判断句","certainty":"low|medium|high","evidence_quote":"原文引句","evidence_kind":"finding|method|dataset|benchmark"}]}` + "\n" + - "要求：statement 必须是论文明确支持的可检验判断；evidence_quote 必须是原文逐字摘录。\n" + + `{"claims":[{"statement":"英文判断句","statement_zh":"中文表述","certainty":"tentative|supported","quote":"原文逐字摘录","kind":"text_passage"}]}` + "\n" + + "要求：statement 必须是论文明确支持的可检验判断；quote 必须是文本中逐字出现的原句（用于定位验证）；" + + "certainty 只能是 tentative 或 supported。\n" + fmt.Sprintf("论文标题: %s\n文本: %s\n", title, truncateStr(text, 6000)) } -// ---------- download_source / upsert_paper / fetch_topic_papers ---------- - func HandleDownloadSource(ctx context.Context, env *HandlerEnv, task *Task) (map[string]any, error) { arxivID, _ := task.Input["arxiv_id"].(string) if arxivID == "" { diff --git a/core/runner.go b/core/runner.go index 49d4531..ab5ef70 100644 --- a/core/runner.go +++ b/core/runner.go @@ -158,11 +158,9 @@ func (r *Runner) runTask(ctx context.Context, task *Task, pool *classPool) { hbWG.Add(1) go func() { defer hbWG.Done() - // 续约节奏：lease 的 1/3（Python 同语义），至少 30s - interval := time.Duration(task.TimeoutS) * time.Second / 3 - if interval < 30*time.Second { - interval = 30 * time.Second - } + // 续约节奏：HeartbeatTask 固定续约 +600s——周期必须明显小于续约窗口， + // 否则长任务第二次心跳前 lease 已被 reclaim（R27）。取 200s（600/3）。 + interval := 200 * time.Second ticker := time.NewTicker(interval) defer ticker.Stop() for { diff --git a/core/scheduler.go b/core/scheduler.go index ec6b8fa..00afa62 100644 --- a/core/scheduler.go +++ b/core/scheduler.go @@ -114,12 +114,17 @@ func (s *Scheduler) dailyBriefLoop(ctx context.Context) { // cron 从配置读（默认 04:00 UTC = 北京 12:00） hour := 4 var cronExpr string - var sendEmail bool + var sendEmail, enabled bool if err := s.store.DB.QueryRow( - `SELECT COALESCE(cron_expression,''), COALESCE(send_email_report,false) FROM daily_report_configs LIMIT 1`, - ).Scan(&cronExpr, &sendEmail); err == nil && cronExpr != "" { + `SELECT COALESCE(cron_expression,''), COALESCE(send_email_report,false), COALESCE(enabled,false) + FROM daily_report_configs LIMIT 1`, + ).Scan(&cronExpr, &sendEmail, &enabled); err == nil && cronExpr != "" { hour = parseDailyCronHour(cronExpr) } + // R14：日报开关关闭则完全不生成 + if !enabled { + continue + } if now.Hour() != hour { continue } diff --git a/server/content.go b/server/content.go index 7f4115f..dc20ef5 100644 --- a/server/content.go +++ b/server/content.go @@ -264,7 +264,7 @@ func (s *Server) ragAsk(ctx context.Context, question string, topK int) (string, prompt := "请基于上下文回答问题，输出严格 JSON：" + `{"answer":"...", "confidence":0.0}` + "\n" + fmt.Sprintf("问题: %s\n上下文:\n%s", question, strings.Join(joined, "\n\n")) - parsed, res, err := s.gateway.CompleteJSON(ctx, "skim", prompt) + parsed, res, err := s.GW().CompleteJSON(ctx, "skim", prompt) if err != nil { return "", nil, nil, err } @@ -629,7 +629,7 @@ func (s *Server) handleWritingProcessMultimodal(w http.ResponseWriter, r *http.R // gatewayChat 网关单轮补全（writing 等简单场景）。 func (s *Server) gatewayChat(ctx context.Context, prompt string) (string, error) { - res, err := s.gateway.Chat(ctx, "deep", prompt, 2048) + res, err := s.GW().Chat(ctx, "deep", prompt, 2048) if err != nil { return "", err } diff --git a/server/graph.go b/server/graph.go index 937d78e..4a96012 100644 --- a/server/graph.go +++ b/server/graph.go @@ -835,16 +835,13 @@ func toAny(v any) any { return v } // loadVectors 加载论文 embedding（limit 篇，含 topic 标签）。 func (s *Server) loadVectors(limit int, topicID string) ([]string, [][]float64, map[string][]string) { query := `SELECT p.id, p.embedding_vec FROM papers p` - if topicID != "" { - query += ` JOIN paper_topics pt ON pt.paper_id = p.id AND pt.topic_id=$2` - } - query += fmt.Sprintf(` WHERE p.embedding_vec IS NOT NULL ORDER BY p.created_at DESC LIMIT $%d`, map[bool]int{true: 1, false: 2}[topicID != ""]) args := []any{} if topicID != "" { - args = append(args, 50000, topicID) - } else { - args = append(args, limit) + query += ` JOIN paper_topics pt ON pt.paper_id = p.id AND pt.topic_id=$1` + args = append(args, topicID) } + args = append(args, limit) + query += fmt.Sprintf(` WHERE p.embedding_vec IS NOT NULL ORDER BY p.created_at DESC LIMIT $%d`, len(args)) rows, err := s.db.Query(query, args...) if err != nil { return nil, nil, nil diff --git a/server/handlers.go b/server/handlers.go index 6a8ae26..8fd9a14 100644 --- a/server/handlers.go +++ b/server/handlers.go @@ -3,6 +3,7 @@ package main // handlers.go：已移植路由的业务处理（folder-stats / 设备信息与授权决策）。 import ( + "encoding/json" "net/http" "time" ) @@ -80,7 +81,7 @@ func (s *Server) requireAuth(next http.HandlerFunc) http.HandlerFunc { return } // API 令牌（SHA-256 查表）优先；Web JWT 兜底 - if s.validAPIToken(token) { + if s.validAPITokenScoped(token, r.Method) { next(w, r) return } @@ -119,6 +120,34 @@ func (s *Server) validAPIToken(raw string) bool { return err == nil } +// validAPITokenScoped：API 令牌 + 方法级 scope 校验（R01——写方法需 write scope）。 +func (s *Server) validAPITokenScoped(raw string, method string) bool { + if method == http.MethodGet || method == http.MethodHead || method == http.MethodOptions { + return s.validAPIToken(raw) + } + var id string + var scopes []byte + err := s.db.QueryRow( + `SELECT id, scopes FROM api_tokens + WHERE token_hash = $1 AND revoked_at IS NULL + AND (expires_at IS NULL OR expires_at > now()::timestamp)`, + hashToken(raw), + ).Scan(&id, &scopes) + if err != nil { + return false + } + var list []string + if json.Unmarshal(scopes, &list) != nil { + return false + } + for _, sc := range list { + if sc == "write" { + return true + } + } + return false +} + func validWebJWT(token, secret string) bool { parsed, err := jwtParseHS256(token, secret) return err == nil && parsed.valid diff --git a/server/jobs.go b/server/jobs.go index 265f072..3cf66f0 100644 --- a/server/jobs.go +++ b/server/jobs.go @@ -4,6 +4,7 @@ package main // Go Core 本身拥有 Job/Task/Attempt 权威数据——大部分端点直接查 PG core_jobs/core_tasks。 import ( + core "github.com/Color2333/PaperMind/core" "encoding/json" "net/http" ) @@ -70,6 +71,30 @@ func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) { return } + attempts := []map[string]any{} + attemptRows, err := s.db.Query( + `SELECT a.id, a.task_id, a.attempt_no, a.executor_id, a.status, + COALESCE(a.error_class,''), COALESCE(a.error_message,''), a.started_at::text, + COALESCE(TO_CHAR(a.finished_at,'YYYY-MM-DD"T"HH24:MI:SS"Z"'),'') + FROM core_attempts a JOIN core_tasks t ON t.id = a.task_id + WHERE t.job_id=$1 ORDER BY a.started_at DESC LIMIT 200`, jobID) + if err == nil { + for attemptRows.Next() { + var aid, tid, executor, aStatus, started, finished, errClass, errMsg string + var attemptNo int + if err := attemptRows.Scan(&aid, &tid, &attemptNo, &executor, &aStatus, + &errClass, &errMsg, &started, &finished); err == nil { + attempts = append(attempts, map[string]any{ + "id": aid, "task_id": tid, "attempt_no": attemptNo, + "executor_id": executor, "status": aStatus, + "error_class": errClass, "error_message": errMsg, + "started_at": started, "finished_at": finished, + }) + } + } + attemptRows.Close() + } + tasks := []map[string]any{} taskRows, err := s.db.Query( `SELECT id, capability, status, attempt_count, max_attempts, @@ -96,7 +121,7 @@ func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) { json.Unmarshal([]byte(payload), &payloadObj) writeJSON(w, http.StatusOK, map[string]any{ "id": jid, "kind": kind, "capability": capability, "status": jobStatus, - "created_at": createdAt, "payload": payloadObj, "tasks": tasks, + "created_at": createdAt, "payload": payloadObj, "tasks": tasks, "attempts": attempts, }) } @@ -145,18 +170,15 @@ func (s *Server) handleSubmitDurable(w http.ResponseWriter, r *http.Request) { // handleCancelJob：POST /jobs/{id}/cancel。 func (s *Server) handleCancelJob(w http.ResponseWriter, r *http.Request) { jobID := r.PathValue("job_id") - res, err := s.db.Exec( - `UPDATE core_jobs SET status='cancelled', finished_at=now()::timestamp - WHERE id=$1 AND status IN ('queued','running')`, jobID) + // R09：复用 CoreStore.CancelJob 权威状态机——leased 子任务进入 cancelling， + // Runner 心跳探测到 cancel_requested 后安全点退出，而不是只取消 queued。 + store := &core.CoreStore{DB: s.db, IsPG: true} + counts, err := store.CancelJob(jobID) if err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"detail": err.Error()}) return } - n, _ := res.RowsAffected() - // 取消未领取的子任务 - _, _ = s.db.Exec( - `UPDATE core_tasks SET status='cancelled' WHERE job_id=$1 AND status='queued'`, jobID) - writeJSON(w, http.StatusOK, map[string]int{"cancelled": int(n)}) + writeJSON(w, http.StatusOK, counts) } // handleRetryJob：POST /jobs/{id}/retry。 @@ -200,14 +222,14 @@ func (s *Server) handleRetryTask(w http.ResponseWriter, r *http.Request) { // handlePauseQueue / handleResumeQueue：队列暂停/恢复（system_flags 持久化）。 func (s *Server) handlePauseQueue(w http.ResponseWriter, r *http.Request) { _, _ = s.db.Exec( - `INSERT INTO system_flags (key, value, updated_at) VALUES ('queue_paused', 'true', now()::timestamp) + `INSERT INTO system_flags (key, value, updated_at) VALUES ('queue_paused', '1', now()::timestamp) ON CONFLICT (key) DO UPDATE SET value='1', updated_at=now()::timestamp ON CONFLICT (key) DO UPDATE SET value='true', updated_at=now()::timestamp`) writeJSON(w, http.StatusOK, map[string]bool{"paused": true}) } func (s *Server) handleResumeQueue(w http.ResponseWriter, r *http.Request) { _, _ = s.db.Exec( - `INSERT INTO system_flags (key, value, updated_at) VALUES ('queue_paused', 'false', now()::timestamp) + `INSERT INTO system_flags (key, value, updated_at) VALUES ('queue_paused', '0', now()::timestamp) ON CONFLICT (key) DO UPDATE SET value='0', updated_at=now()::timestamp ON CONFLICT (key) DO UPDATE SET value='false', updated_at=now()::timestamp`) writeJSON(w, http.StatusOK, map[string]bool{"paused": false}) } diff --git a/server/main.go b/server/main.go index 55b0fe2..b41267e 100644 --- a/server/main.go +++ b/server/main.go @@ -157,12 +157,12 @@ func main() { mux.HandleFunc("POST /sensemaking/sessions", s.requireAuth(s.handleCreateSenseSession)) mux.HandleFunc("GET /sensemaking/sessions", s.requireAuth(s.handleListSenseSessions)) mux.HandleFunc("GET /sensemaking/sessions/{session_id}", s.requireAuth(s.handleGetSenseSession)) - mux.HandleFunc("PATCH /sensemaking/sessions/{session_id}/act1", s.requireAuth(s.handleUpdateAct)) - mux.HandleFunc("PATCH /sensemaking/sessions/{session_id}/act2", s.requireAuth(s.handleUpdateAct)) - mux.HandleFunc("PATCH /sensemaking/sessions/{session_id}/act3", s.requireAuth(s.handleUpdateAct)) - mux.HandleFunc("POST /sensemaking/sessions/{session_id}/act1/generate", s.requireAuth(s.handleGenerateAct)) - mux.HandleFunc("POST /sensemaking/sessions/{session_id}/act2/generate", s.requireAuth(s.handleGenerateAct)) - mux.HandleFunc("POST /sensemaking/sessions/{session_id}/act3/generate", s.requireAuth(s.handleGenerateAct)) + mux.HandleFunc("PATCH /sensemaking/sessions/{session_id}/act1", s.requireAuth(s.makeActHandler("act1"))) + mux.HandleFunc("PATCH /sensemaking/sessions/{session_id}/act2", s.requireAuth(s.makeActHandler("act2"))) + mux.HandleFunc("PATCH /sensemaking/sessions/{session_id}/act3", s.requireAuth(s.makeActHandler("act3"))) + mux.HandleFunc("POST /sensemaking/sessions/{session_id}/act1/generate", s.requireAuth(s.makeActGenerateHandler("act1"))) + mux.HandleFunc("POST /sensemaking/sessions/{session_id}/act2/generate", s.requireAuth(s.makeActGenerateHandler("act2"))) + mux.HandleFunc("POST /sensemaking/sessions/{session_id}/act3/generate", s.requireAuth(s.makeActGenerateHandler("act3"))) mux.HandleFunc("POST /sensemaking/interactions", s.requireAuth(s.handleCreateInteraction)) mux.HandleFunc("POST /citations/sync/incremental", s.requireAuth(s.handleSyncCitationsIncremental)) mux.HandleFunc("POST /citations/sync/topic/{topic_id}", s.requireAuth(s.handleSyncCitationsTopic)) diff --git a/server/phase7.go b/server/phase7.go index 8c14c21..83a8ee2 100644 --- a/server/phase7.go +++ b/server/phase7.go @@ -279,7 +279,20 @@ func (s *Server) handleSearchMulti(w http.ResponseWriter, r *http.Request) { MaxResults int `json:"max_results"` Channels []string `json:"channels"` } - if err := readBody(r, &body); err != nil || strings.TrimSpace(body.Query) == "" { + // 契约（R16）：前端把 query/channels 放 URL query 且无 JSON body——两种都接受 + if err := readBody(r, &body); err != nil { + writeJSON(w, http.StatusUnprocessableEntity, map[string]string{"detail": "invalid body"}) + return + } + if strings.TrimSpace(body.Query) == "" { + body.Query = r.URL.Query().Get("query") + } + if len(body.Channels) == 0 { + if ch := r.URL.Query().Get("channels"); ch != "" { + body.Channels = strings.Split(ch, ",") + } + } + if strings.TrimSpace(body.Query) == "" { writeJSON(w, http.StatusUnprocessableEntity, map[string]string{"detail": "query required"}) return } @@ -337,7 +350,35 @@ func (s *Server) handleSearchMulti(w http.ResponseWriter, r *http.Request) { if channelEnabled(body.Channels, "biorxiv") { channels["biorxiv"] = map[string]any{"error": "bioRxiv 渠道暂不支持关键词检索"} } - writeJSON(w, http.StatusOK, map[string]any{"query": body.Query, "channels": channels}) + // 契约（R16）：MultiSourceSearchResult{papers:[MultiSourcePaper], channel_stats} + papers := []map[string]any{} + channelStats := map[string]any{} + for ch, raw := range channels { + if errObj, isErr := raw.(map[string]any); isErr { + if _, onlyErr := errObj["error"]; onlyErr { + channelStats[ch] = map[string]any{"total": 0, "new": 0, "duplicates": 0, "error": errObj["error"]} + continue + } + } + list, _ := raw.([]any) + channelStats[ch] = map[string]any{"total": len(list), "new": len(list), "duplicates": 0} + for _, it := range list { + m, ok := it.(map[string]any) + if !ok { + continue + } + papers = append(papers, map[string]any{ + "id": firstNonEmptyStr(strOf(m["arxiv_id"]), strOf(m["doi"])), + "title": strOf(m["title"]), + "authors": []any{}, + "abstract": strOf(m["abstract"]), + "year": m["year"], + "venue": m["venue"], + "sources": []any{map[string]any{"channel": ch}}, + }) + } + } + writeJSON(w, http.StatusOK, map[string]any{"papers": papers, "channel_stats": channelStats}) } func channelEnabled(channels []string, name string) bool { @@ -562,3 +603,12 @@ func urlQueryEscape(s string) string { } var httpDefaultClient = &http.Client{Timeout: 30 * time.Second} + +func firstNonEmptyStr(vals ...string) string { + for _, v := range vals { + if v != "" { + return v + } + } + return "" +} diff --git a/server/sensemaking.go b/server/sensemaking.go index a7b0db1..5b8f1f9 100644 --- a/server/sensemaking.go +++ b/server/sensemaking.go @@ -402,10 +402,28 @@ func (s *Server) handleListSenseSessions(w http.ResponseWriter, r *http.Request) writeJSON(w, http.StatusOK, items) } +// makeActHandler 显式绑定 act 字面路由（PathValue 无通配）。 +func (s *Server) makeActHandler(act string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + s.updateAct(w, r, act) + } +} + +// makeActGenerateHandler act 生成路由绑定。 +func (s *Server) makeActGenerateHandler(act string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + s.generateAct(w, r, act) + } +} + // handleUpdateAct PATCH /sensemaking/sessions/{session_id}/act{1|2|3}。 func (s *Server) handleUpdateAct(w http.ResponseWriter, r *http.Request) { - sessionID := r.PathValue("session_id") act := r.PathValue("act") + s.updateAct(w, r, act) +} + +func (s *Server) updateAct(w http.ResponseWriter, r *http.Request, act string) { + sessionID := r.PathValue("session_id") var data map[string]any if err := readBody(r, &data); err != nil { writeJSON(w, http.StatusUnprocessableEntity, map[string]string{"detail": "invalid body"}) @@ -438,8 +456,12 @@ func (s *Server) handleUpdateAct(w http.ResponseWriter, r *http.Request) { // handleGenerateAct POST /sensemaking/sessions/{session_id}/act{1|2|3}/generate。 func (s *Server) handleGenerateAct(w http.ResponseWriter, r *http.Request) { - sessionID := r.PathValue("session_id") act := r.PathValue("act") + s.generateAct(w, r, act) +} + +func (s *Server) generateAct(w http.ResponseWriter, r *http.Request, act string) { + sessionID := r.PathValue("session_id") // 上下文加载：论文 + schema + 前序 act var paperID, schemaID string var act1Raw, act2Raw []byte diff --git a/server/tags.go b/server/tags.go index a1f50d2..ffe4114 100644 --- a/server/tags.go +++ b/server/tags.go @@ -103,7 +103,7 @@ func (s *Server) handleUpdateTag(w http.ResponseWriter, r *http.Request) { return } sets = append(sets, "updated_at=now()::timestamp") - res, err := s.db.Exec("UPDATE tags SET "+strings.Join(sets, ", ")+" WHERE id=$"+itoa(len(sets)+1), + res, err := s.db.Exec("UPDATE tags SET "+strings.Join(sets, ", ")+" WHERE id=$"+itoa(len(args)+1), append(args, id)...) if err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"detail": err.Error()}) @@ -151,8 +151,8 @@ func (s *Server) handleAddPaperTag(w http.ResponseWriter, r *http.Request) { return } if _, err := s.db.Exec( - `INSERT INTO paper_tags (paper_id, tag_id) VALUES ($1, $2) - ON CONFLICT DO NOTHING`, paperID, tagID, + `INSERT INTO paper_tags (id, paper_id, tag_id, created_at) VALUES ($1, $2, $3, now()::timestamp) + ON CONFLICT DO NOTHING`, newUUID(), paperID, tagID, ); err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"detail": err.Error()}) return @@ -181,18 +181,29 @@ func (s *Server) handleBatchPaperTags(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadRequest, map[string]string{"detail": "bad json"}) return } - if _, err := s.db.Exec("DELETE FROM paper_tags WHERE paper_id=$1", paperID); err != nil { + // R05：整批替换必须事务化——删除后插入失败要回滚 + tx, err := s.db.Begin() + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"detail": err.Error()}) + return + } + defer tx.Rollback() + if _, err := tx.Exec("DELETE FROM paper_tags WHERE paper_id=$1", paperID); err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"detail": err.Error()}) return } for _, tagID := range tagIDs { - if _, err := s.db.Exec( - "INSERT INTO paper_tags (paper_id, tag_id) VALUES ($1, $2) ON CONFLICT DO NOTHING", - paperID, tagID, + if _, err := tx.Exec( + "INSERT INTO paper_tags (id, paper_id, tag_id, created_at) VALUES ($1, $2, $3, now()::timestamp) ON CONFLICT DO NOTHING", + newUUID(), paperID, tagID, ); err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"detail": err.Error()}) return } } + if err := tx.Commit(); err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"detail": err.Error()}) + return + } writeJSON(w, http.StatusOK, map[string]any{"items": s.tagsForPaper(paperID)}) } diff --git a/server/topics.go b/server/topics.go index 5567014..fa010d7 100644 --- a/server/topics.go +++ b/server/topics.go @@ -230,33 +230,61 @@ func (s *Server) handleManualFetchTopic(w http.ResponseWriter, r *http.Request) // handleTopicStats：GET /topics/stats。 func (s *Server) handleTopicStats(w http.ResponseWriter, r *http.Request) { + // 契约：TopicStatsResponse{topics:[TopicStats{topic_id,topic_name,paper_count,total_citations,recent_30d,status_dist}]} rows, err := s.db.Query( - `SELECT ts.name, COUNT(pt.paper_id) AS cnt, - COALESCE(ts.last_error,''), ts.enabled + `SELECT ts.id, ts.name, COUNT(DISTINCT pt.paper_id), + COALESCE((SELECT COUNT(*) FROM citations c + JOIN paper_topics pt2 ON pt2.paper_id = c.source_paper_id + WHERE pt2.topic_id = ts.id), 0) AS total_citations, + COALESCE((SELECT COUNT(*) FROM paper_topics pt3 JOIN papers p3 ON p3.id = pt3.paper_id + WHERE pt3.topic_id = ts.id AND p3.created_at >= NOW() - INTERVAL '30 days'), 0) FROM topic_subscriptions ts LEFT JOIN paper_topics pt ON pt.topic_id = ts.id - GROUP BY ts.id, ts.name, ts.last_error, ts.enabled - ORDER BY cnt DESC`) + GROUP BY ts.id, ts.name ORDER BY COUNT(DISTINCT pt.paper_id) DESC`) if err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"detail": err.Error()}) return } defer rows.Close() - items := []map[string]any{} + topics := []map[string]any{} + ids := []string{} for rows.Next() { - var name string - var cnt int - var lastErr *string - var enabled bool - if err := rows.Scan(&name, &cnt, &lastErr, &enabled); err != nil { + var id, name string + var paperCount, citations, recent int + if err := rows.Scan(&id, &name, &paperCount, &citations, &recent); err != nil { continue } - items = append(items, map[string]any{ - "name": name, "paper_count": cnt, "enabled": enabled, - "last_error": jsonStrOrNull(lastErr), + ids = append(ids, id) + topics = append(topics, map[string]any{ + "topic_id": id, "topic_name": name, "paper_count": paperCount, + "total_citations": citations, "recent_30d": recent, + "status_dist": map[string]int{"unread": 0, "skimmed": 0, "deep_read": 0}, }) } - writeJSON(w, http.StatusOK, map[string]any{"items": items}) + rows.Close() + // 每主题的阅读状态分布 + for _, t := range topics { + tid := t["topic_id"].(string) + sd := map[string]int{"unread": 0, "skimmed": 0, "deep_read": 0} + srows, err := s.db.Query( + `SELECT p.read_status, COUNT(*) FROM papers p + JOIN paper_topics pt ON pt.paper_id = p.id + WHERE pt.topic_id=$1 GROUP BY p.read_status`, tid) + if err == nil { + for srows.Next() { + var rs string + var c int + if srows.Scan(&rs, &c) == nil { + if _, ok := sd[rs]; ok { + sd[rs] = c + } + } + } + srows.Close() + t["status_dist"] = sd + } + } + writeJSON(w, http.StatusOK, map[string]any{"topics": topics}) } // handleIngestArxiv：POST /ingest/arxiv —— 提交 ingest_arxiv_query 任务到 Go Core。