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。