diff options
| -rw-r--r-- | .gitignore | 15 | ||||
| -rw-r--r-- | Containerfile | 31 | ||||
| -rw-r--r-- | README.md | 55 | ||||
| -rw-r--r-- | cmds/export.go | 402 | ||||
| -rw-r--r-- | cmds/export_test.go | 441 | ||||
| -rw-r--r-- | cmds/retrostar.go | 127 | ||||
| -rw-r--r-- | cmds/retrostar_test.go | 64 | ||||
| -rw-r--r-- | cmds/rss.go | 82 | ||||
| -rw-r--r-- | cmds/rss_test.go | 73 | ||||
| -rw-r--r-- | cmds/score.go | 103 | ||||
| -rw-r--r-- | cmds/serve.go | 452 | ||||
| -rw-r--r-- | cmds/serve_test.go | 407 | ||||
| -rw-r--r-- | cmds/train.go | 430 | ||||
| -rw-r--r-- | cmds/train_test.go | 260 | ||||
| -rw-r--r-- | core/constants.go | 23 | ||||
| -rw-r--r-- | core/labels.go | 127 | ||||
| -rw-r--r-- | core/labels_test.go | 192 | ||||
| -rw-r--r-- | core/ml.go | 435 | ||||
| -rw-r--r-- | core/ml_test.go | 76 | ||||
| -rw-r--r-- | core/model.go | 22 | ||||
| -rw-r--r-- | core/runs.go | 167 | ||||
| -rw-r--r-- | core/scoring.go | 16 | ||||
| -rw-r--r-- | core/scoring_text.go | 43 | ||||
| -rw-r--r-- | core/scoring_text_test.go | 99 | ||||
| -rw-r--r-- | core/text.go | 38 | ||||
| -rw-r--r-- | core/types.go | 59 | ||||
| -rw-r--r-- | go.mod | 3 | ||||
| -rw-r--r-- | justfile | 68 | ||||
| -rw-r--r-- | main.go | 76 | ||||
| -rw-r--r-- | miniflux/client.go | 201 | ||||
| -rw-r--r-- | miniflux/client_test.go | 307 |
31 files changed, 4894 insertions, 0 deletions
diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..cd37101 --- /dev/null +++ b/.gitignore @@ -0,0 +1,15 @@ +# Built binary +fluxrec + +# Data artifacts (plumbing, not source) +labels.jsonl +model.json +cursor.json +runs.jsonl +export_state.json +export_run.json +report.json +data/ + +# Dev fixtures: real API dumps, personal titles/URLs — never commit +testdata/ diff --git a/Containerfile b/Containerfile new file mode 100644 index 0000000..3f52aef --- /dev/null +++ b/Containerfile @@ -0,0 +1,31 @@ +# fluxrec serve container (rss + status on :8080, state in /data). +# +# build: docker build -f Containerfile -t fluxrec . +# run: docker run -d --name fluxrec \ +# --env-file /srv/fluxrec/env \ +# -v /srv/fluxrec:/data \ +# -p 127.0.0.1:8080:8080 \ +# fluxrec +# +# The env file holds MINIFLUX_URL and MINIFLUX_TOKEN (plus optional TZ — +# refresh times are computed in it; default UTC, tzdata is installed). The +# env file lives on the server only, never in this repo. /data holds +# model.json (scp'd by `just deploy`), cursor.json and runs.jsonl. +FROM golang:1.25-alpine AS build +WORKDIR /src +COPY . . +RUN CGO_ENABLED=0 go build -trimpath -o /out/fluxrec . + +FROM alpine:3.21 +RUN apk add --no-cache tzdata \ + && adduser -D fluxrec \ + && mkdir -p /data && chown fluxrec:fluxrec /data +USER fluxrec +COPY --from=build /out/fluxrec /usr/local/bin/fluxrec +EXPOSE 8080 +ENTRYPOINT ["fluxrec", "serve", \ + "--model", "/data/model.json", \ + "--cursor", "/data/cursor.json", \ + "--runs", "/data/runs.jsonl"] +# Override refresh times / rec feed ID via extra args after the image name, e.g.: +# docker run ... fluxrec --refresh-at "06:00,12:00,18:00" --rec-feed-id 123 diff --git a/README.md b/README.md new file mode 100644 index 0000000..a8e8ec1 --- /dev/null +++ b/README.md @@ -0,0 +1,55 @@ +# fluxrec + +Personal Miniflux article recommender. Ranks your incoming RSS entries with a +title-only classifier trained on the ones you star, and serves the top picks +as an RSS feed you subscribe to in Miniflux itself. Read-only toward Miniflux. +No deps beyond stdlib + vendored ML core. + +## Requirements + +- Miniflux ≥ 2.0.49 (uses the `changed_after`/`published_after` entry filters) +- Go ≥ 1.25 to build; everything else is stdlib + the vendored core in `core/` +- Training runs wherever you read (`export`/`train`); `serve` runs next to + Miniflux and only needs the model file + +## Commands + +``` +fluxrec export stars + sampled negatives → labels.jsonl (local) +fluxrec train labels.jsonl → model.json + report.json (local) +fluxrec serve poll, score, RSS at /recommendations.xml (server) +fluxrec retrostar model-ranked candidates to star by hand (cold start) +fluxrec score debug: stdin titles → stdout scores +``` + +## Setup + +``` +just auth # prints the two export lines; paste them in your shell +# one-time: retroactively star 50–100 remembered-good entries in Miniflux +# then weekly-ish: +just retrain # export + train; read report.json (precision@15) +just ship myhost # scp model.json + restart serve (myhost = ssh alias) +# finally: subscribe your Miniflux to https://server/recommendations.xml +``` + +Deploy serve only after a real model exists with an honest precision@15. + +## Files + +``` +labels.jsonl canonical label store (survives Miniflux deletion) +export_state.json local incremental cursor +export_run.json per-run summary (overwritten each export) +model.json TF-IDF vocab + LR weights (trained locally, shipped to server) +report.json train's validation report (precision@15 headline metric) +cursor.json server poll cursor +runs.jsonl server run log; doubles as the RSS render source +``` + +## Notes + +- Never writes to Miniflux. Never needs credentials in the repo. +- Training stays on the local PC; the server only scores (low RAM). +- Deploy `serve` only after a `train` run on enough organic stars shows an + honest precision@15 in report.json; tiny validation sets lie. diff --git a/cmds/export.go b/cmds/export.go new file mode 100644 index 0000000..102c94a --- /dev/null +++ b/cmds/export.go @@ -0,0 +1,402 @@ +// Export command: pull stars and sampled negatives from Miniflux into +// labels.jsonl, the canonical label store. +// +// Positives: starred entries, incremental via a persisted changed_after +// cursor, deduped by ID then normalized URL. Negatives: seeded random +// sample of recent read+unstarred entries, topped up to a fixed ratio of +// the positive pool. Also carries --dump-raw fixture capture. +package cmds + +import ( + "encoding/json" + "flag" + "fmt" + "io" + "math/rand" + "os" + "path/filepath" + "time" + + "fluxrec/core" + "fluxrec/miniflux" +) + +type ExportCommand struct { + DumpRaw string + Labels string + State string + RunOut string + RecFeedID int64 + NegDays int + NegRatio int + NegSeed int64 +} + +func (c *ExportCommand) Name() string { return "export" } + +func (c *ExportCommand) Init(args []string) error { + fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError) + fs.Usage = func() { + fmt.Fprint(fs.Output(), `Usage: fluxrec export [--labels FILE] [--state FILE] [--rec-feed-id ID] + fluxrec export --dump-raw DIR + +Export labels from Miniflux into labels.jsonl: starred entries become +positives (incremental via --state cursor), plus a seeded random sample of +recent read+unstarred entries as negatives at --neg-ratio per positive. +Each run also writes a JSON summary to --run-out. + +With --dump-raw, instead fetch ~25 starred and ~100 read unstarred entries +and write the raw API responses to DIR for offline development fixtures. +`) + fs.PrintDefaults() + } + fs.StringVar(&c.DumpRaw, "dump-raw", "", "Write raw API fixtures to DIR and exit") + fs.StringVar(&c.Labels, "labels", "labels.jsonl", "Path to labels.jsonl") + fs.StringVar(&c.State, "state", "export_state.json", "Path to the export cursor file") + fs.StringVar(&c.RunOut, "run-out", "export_run.json", "Path to write the per-run summary JSON") + fs.Int64Var(&c.RecFeedID, "rec-feed-id", 0, "Miniflux feed ID of the recommendations feed (0 = not deployed yet)") + fs.IntVar(&c.NegDays, "neg-days", 45, "Only sample negatives published within this many days") + fs.IntVar(&c.NegRatio, "neg-ratio", 20, "Target negatives per positive") + fs.Int64Var(&c.NegSeed, "neg-seed", 20240301, "Seed for negative sampling (same data + same seed = same sample)") + return fs.Parse(args) +} + +func (c *ExportCommand) Run(stdin io.Reader, stdout io.Writer) error { + if c.DumpRaw != "" { + return c.runDumpRaw(stdout) + } + return c.runExport(stdout) +} + +// exportState is the persisted incremental cursor: the largest changed_at +// (unix seconds) seen so far. Refetching starts one second before it so a +// star landing in the same second as the previous run's last entry can't +// slip through; the overlap is absorbed by dedup. +type exportState struct { + ChangedAfter int64 `json:"changed_after"` +} + +func loadExportState(path string) (exportState, error) { + var st exportState + body, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return st, nil + } + return st, fmt.Errorf("reading state %s: %w", path, err) + } + if err := json.Unmarshal(body, &st); err != nil { + return st, fmt.Errorf("parsing state %s: %w", path, err) + } + return st, nil +} + +func saveExportState(path string, st exportState) error { + body, err := json.Marshal(st) + if err != nil { + return err + } + if err := os.WriteFile(path, body, 0o644); err != nil { + return fmt.Errorf("writing state %s: %w", path, err) + } + return nil +} + +// exportCounts is what the run reports to stdout. +type exportCounts struct { + seen, appended, idDup, urlDup, recHit, recUnresolved int +} + +func (c *ExportCommand) runExport(stdout io.Writer) error { + started := time.Now() + cfg, err := miniflux.ConfigFromEnv() + if err != nil { + return err + } + client := miniflux.NewClient(cfg) + + state, err := loadExportState(c.State) + if err != nil { + return err + } + + existing, err := core.ReadLabels(c.Labels) + if err != nil { + return err + } + seenIDs := make(map[int64]bool, len(existing)) + seenURLs := make(map[string]bool, len(existing)) + for _, row := range existing { + seenIDs[row.EntryID] = true + seenURLs[core.NormalizeURL(row.URL)] = true + } + + starred := true + f := miniflux.Filters{Starred: &starred, Limit: 200} + if state.ChangedAfter > 0 { + f.ChangedAfter = state.ChangedAfter - 1 // boundary overlap, see exportState doc + } + + var counts exportCounts + var newRows []core.LabelRow + maxChanged := state.ChangedAfter + now := time.Now().UTC() + + for { + set, err := client.Entries(f) + if err != nil { + return err + } + if len(set.Entries) == 0 { + break + } + for _, e := range set.Entries { + counts.seen++ + if ts := e.ChangedAt.Unix(); ts > maxChanged { + maxChanged = ts + } + + // Stars on the recommendations feed are not original entries; + // resolve by URL against existing labels. If the original was + // never exported, there is nothing to attach the star to. + if c.RecFeedID != 0 && e.Feed.ID == c.RecFeedID { + if seenURLs[core.NormalizeURL(e.URL)] { + counts.recHit++ + } else { + counts.recUnresolved++ + } + continue + } + + if seenIDs[e.ID] { + counts.idDup++ + continue + } + nu := core.NormalizeURL(e.URL) + if seenURLs[nu] { + counts.urlDup++ + continue + } + newRows = append(newRows, core.LabelRow{ + EntryID: e.ID, URL: e.URL, Title: e.Title, + FeedID: e.Feed.ID, FeedTitle: e.Feed.Title, + Label: core.LabelPos, CapturedAt: now, + }) + seenIDs[e.ID] = true + seenURLs[nu] = true + counts.appended++ + } + f.Offset += len(set.Entries) + if f.Offset >= set.Total { + break + } + } + + if err := core.AppendLabels(c.Labels, newRows); err != nil { + return err + } + if maxChanged > state.ChangedAfter { + if err := saveExportState(c.State, exportState{ChangedAfter: maxChanged}); err != nil { + return err + } + } + + posTotal, negTotal := 0, 0 + for _, row := range existing { + if row.Label == core.LabelPos { + posTotal++ + } else { + negTotal++ + } + } + posTotal += counts.appended + + fmt.Fprintf(stdout, "export: seen=%d appended=%d (id-dup=%d url-dup=%d rec-hit=%d rec-unresolved=%d) pos-total=%d\n", + counts.seen, counts.appended, counts.idDup, counts.urlDup, + counts.recHit, counts.recUnresolved, posTotal) + if counts.recUnresolved > 0 { + fmt.Fprintln(stdout, "note: rec-feed stars without a matching labeled original were skipped") + } + + nc, err := c.exportNegatives(client, seenIDs, seenURLs, posTotal, negTotal, now, stdout) + if err != nil { + return err + } + return writeRunSummary(c.RunOut, runSummary{ + RanAt: now, DurationMs: time.Since(started).Milliseconds(), + PositivesSeen: counts.seen, PositivesNew: counts.appended, + PosTotal: posTotal, NegCandidates: nc.candidates, NegNew: nc.appended, + NegTotal: negTotal + nc.appended, IDDup: counts.idDup, URLDup: counts.urlDup, + RecHit: counts.recHit, RecUnresolved: counts.recUnresolved, + }) +} + +// negCounts is what the negatives sampler reports back. +type negCounts struct { + candidates, appended int +} + +// exportNegatives samples read, unstarred entries published within the +// --neg-days window down to labels.jsonl as LabelNeg rows, topping the +// negative pool up to --neg-ratio × positives. Anything already labeled +// (either polarity) is excluded; sampling is seeded so the same candidate +// pool + seed reproduce the same sample. +// +// The window is enforced server-side via published_after (Miniflux >= 2.0.49, +// the same floor the rest of export requires): only in-window entries are +// fetched, so the deep archive is never walked and no client-side order +// assumption is needed (Miniflux's default order is published_at asc). +func (c *ExportCommand) exportNegatives(client *miniflux.Client, seenIDs map[int64]bool, seenURLs map[string]bool, posTotal, negTotal int, now time.Time, stdout io.Writer) (negCounts, error) { + var nc negCounts + need := c.NegRatio*posTotal - negTotal + if need <= 0 || posTotal == 0 { + fmt.Fprintf(stdout, "negatives: skipped (need=%d pos=%d)\n", need, posTotal) + return nc, nil + } + + cutoff := now.Add(-time.Duration(c.NegDays) * 24 * time.Hour) + unstarred := false + f := miniflux.Filters{ + Starred: &unstarred, Status: "read", Limit: 200, + PublishedAfter: cutoff.Unix(), + } + + candIDs := make(map[int64]bool) + var cands []miniflux.Entry + for { + set, err := client.Entries(f) + if err != nil { + return nc, err + } + if len(set.Entries) == 0 { + break + } + for _, e := range set.Entries { + if seenIDs[e.ID] || candIDs[e.ID] || seenURLs[core.NormalizeURL(e.URL)] { + continue + } + candIDs[e.ID] = true + cands = append(cands, e) + } + f.Offset += len(set.Entries) + if f.Offset >= set.Total { + break + } + } + nc.candidates = len(cands) + + rng := rand.New(rand.NewSource(c.NegSeed)) + perm := rng.Perm(len(cands)) + take := min(need, len(cands)) + rows := make([]core.LabelRow, 0, take) + for i := range take { + e := cands[perm[i]] + rows = append(rows, core.LabelRow{ + EntryID: e.ID, URL: e.URL, Title: e.Title, + FeedID: e.Feed.ID, FeedTitle: e.Feed.Title, + Label: core.LabelNeg, CapturedAt: now, + }) + seenIDs[e.ID] = true + seenURLs[core.NormalizeURL(e.URL)] = true + } + if err := core.AppendLabels(c.Labels, rows); err != nil { + return nc, err + } + nc.appended = len(rows) + + fmt.Fprintf(stdout, "negatives: candidates=%d appended=%d (target=%d:%d)\n", + nc.candidates, nc.appended, c.NegRatio, posTotal) + if nc.appended < need { + fmt.Fprintf(stdout, "note: only %d usable candidates in the last %d days; negatives short by %d\n", + nc.candidates, c.NegDays, need-nc.appended) + } + return nc, nil +} + +// runSummary is the per-run JSON summary written for visibility; kept to +// the latest run (overwrite, not append). +type runSummary struct { + RanAt time.Time `json:"ran_at"` + DurationMs int64 `json:"duration_ms"` + PositivesSeen int `json:"positives_seen"` + PositivesNew int `json:"positives_new"` + PosTotal int `json:"pos_total"` + NegCandidates int `json:"neg_candidates"` + NegNew int `json:"neg_new"` + NegTotal int `json:"neg_total"` + IDDup int `json:"id_dup"` + URLDup int `json:"url_dup"` + RecHit int `json:"rec_hit"` + RecUnresolved int `json:"rec_unresolved"` +} + +func writeRunSummary(path string, s runSummary) error { + body, err := json.MarshalIndent(s, "", " ") + if err != nil { + return err + } + if err := os.WriteFile(path, body, 0o644); err != nil { + return fmt.Errorf("writing run summary %s: %w", path, err) + } + return nil +} + +// ── fixture capture (--dump-raw) ── + +// dumpJob is one fixture to capture: output filename + API filters. +type dumpJob struct { + filename string + filters miniflux.Filters +} + +func dumpJobs() []dumpJob { + starred, unstarred := true, false + return []dumpJob{ + {"starred_sample.json", miniflux.Filters{Starred: &starred, Limit: 25}}, + // Negatives fixture must NOT overlap stars; a plain status=read + // sample can be all-starred after a retroactive starring pass. + {"read_sample.json", miniflux.Filters{Starred: &unstarred, Status: "read", Limit: 100}}, + } +} + +func (c *ExportCommand) runDumpRaw(stdout io.Writer) error { + cfg, err := miniflux.ConfigFromEnv() + if err != nil { + return err + } + client := miniflux.NewClient(cfg) + + if err := os.MkdirAll(c.DumpRaw, 0o755); err != nil { + return fmt.Errorf("creating %s: %w", c.DumpRaw, err) + } + + for _, job := range dumpJobs() { + body, err := client.RawEntries(job.filters) + if err != nil { + return fmt.Errorf("fetching %s: %w", job.filename, err) + } + // Decode only to report a count and catch a malformed response + // before it becomes a silent bad fixture. + var set miniflux.EntrySet + if err := json.Unmarshal(body, &set); err != nil { + return fmt.Errorf("%s: response did not decode as entries: %w", job.filename, err) + } + path := filepath.Join(c.DumpRaw, job.filename) + if err := os.WriteFile(path, body, 0o644); err != nil { + return fmt.Errorf("writing %s: %w", path, err) + } + // Contamination guard: the negatives fixture must be star-free, + // or dev data for negative sampling teaches the wrong lesson. + starredN := 0 + for _, e := range set.Entries { + if e.Starred { + starredN++ + } + } + if job.filters.Starred != nil && !*job.filters.Starred && starredN > 0 { + fmt.Fprintf(stdout, "WARNING: %s contains %d starred entries (API ignored starred=false?)\n", job.filename, starredN) + } + fmt.Fprintf(stdout, "wrote %s (%d entries, %d bytes)\n", path, len(set.Entries), len(body)) + } + return nil +} diff --git a/cmds/export_test.go b/cmds/export_test.go new file mode 100644 index 0000000..0190d39 --- /dev/null +++ b/cmds/export_test.go @@ -0,0 +1,441 @@ +// Tests for `fluxrec export` against an httptest fake Miniflux. Env vars +// point the command's config resolution at the fake — no real credentials. +package cmds + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "fluxrec/core" + "fluxrec/miniflux" +) + +// testEntry mirrors the entry fields export consumes. +type testFeed struct { + ID int64 `json:"id"` + Title string `json:"title"` +} + +type testEntry struct { + ID int64 `json:"id"` + URL string `json:"url"` + Title string `json:"title"` + Content string `json:"content"` + Status string `json:"status"` + Starred bool `json:"starred"` + PublishedAt time.Time `json:"published_at"` + ChangedAt time.Time `json:"changed_at"` + Feed testFeed `json:"feed"` +} + +func makeEntries(n int, startID int64, starred bool, feedID int64) []testEntry { + base := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + entries := make([]testEntry, n) + for i := range entries { + entries[i] = testEntry{ + ID: startID + int64(i), + URL: fmt.Sprintf("https://example.com/post/%d", startID+int64(i)), + Title: fmt.Sprintf("Title %d", startID+int64(i)), + Status: "read", + Starred: starred, + PublishedAt: base, + ChangedAt: base.Add(time.Duration(i) * time.Hour), + Feed: testFeed{ID: feedID, Title: "Fixture Feed"}, + } + } + return entries +} + +func entriesJSON(total int, entries []testEntry) string { + b, _ := json.Marshal(entries) + return fmt.Sprintf(`{"total":%d,"entries":%s}`, total, b) +} + +// pagedServer serves a starred pool and a read pool (chosen per request by +// the starred filter) with limit/offset paging, and records every query. +type pagedServer struct { + starred []testEntry + read []testEntry + queries []string +} + +func (p *pagedServer) handler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + p.queries = append(p.queries, r.URL.RawQuery) + q := r.URL.Query() + pool := p.read + if q.Get("starred") == "true" { + pool = p.starred + } + // Honor published_after server-side, as real Miniflux does, so the + // negatives sampler's window is exercised by these tests. + if pa, _ := strconv.ParseInt(q.Get("published_after"), 10, 64); pa > 0 { + cutoff := time.Unix(pa, 0) + filtered := pool[:0:0] + for _, e := range pool { + if !e.PublishedAt.Before(cutoff) { + filtered = append(filtered, e) + } + } + pool = filtered + } + limit, _ := strconv.Atoi(q.Get("limit")) + offset, _ := strconv.Atoi(q.Get("offset")) + if limit <= 0 { + limit = len(pool) + } + if offset > len(pool) { + offset = len(pool) + } + end := offset + limit + if end > len(pool) { + end = len(pool) + } + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, entriesJSON(len(pool), pool[offset:end])) + } +} + +// runExportAgainst wires a fake Miniflux via env, then runs the export +// command with the given extra args, returning stdout. +func runExportAgainst(t *testing.T, p *pagedServer, dir string, extra ...string) string { + t.Helper() + srv := httptest.NewServer(p.handler()) + defer srv.Close() + t.Setenv("MINIFLUX_URL", srv.URL) + t.Setenv("MINIFLUX_TOKEN", "test-token") + + args := append([]string{ + "--labels", filepath.Join(dir, "labels.jsonl"), + "--state", filepath.Join(dir, "export_state.json"), + "--run-out", filepath.Join(dir, "export_run.json"), + }, extra...) + + cmd := &ExportCommand{} + if err := cmd.Init(args); err != nil { + t.Fatalf("Init: %v", err) + } + var out bytes.Buffer + if err := cmd.Run(nil, &out); err != nil { + t.Fatalf("Run: %v", err) + } + return out.String() +} + +func TestExportPositivesIdempotent(t *testing.T) { + p := &pagedServer{starred: makeEntries(25, 1000, true, 7)} + dir := t.TempDir() + + out1 := runExportAgainst(t, p, dir) + if !strings.Contains(out1, "appended=25") { + t.Errorf("first run should append 25, got:\n%s", out1) + } + labels, err := core.ReadLabels(filepath.Join(dir, "labels.jsonl")) + if err != nil { + t.Fatalf("read labels: %v", err) + } + if len(labels) != 25 { + t.Fatalf("expected 25 rows, got %d", len(labels)) + } + for _, row := range labels { + if row.Label != core.LabelPos { + t.Errorf("row label = %q, want pos", row.Label) + } + if row.EntryID < 1000 || row.EntryID >= 1025 { + t.Errorf("unexpected entry id %d", row.EntryID) + } + } + + // State cursor must have advanced to the newest changed_at. + stateBody, err := os.ReadFile(filepath.Join(dir, "export_state.json")) + if err != nil { + t.Fatalf("state not written: %v", err) + } + var st exportState + if err := json.Unmarshal(stateBody, &st); err != nil { + t.Fatalf("state unparseable: %v", err) + } + if st.ChangedAfter <= 0 { + t.Errorf("state cursor = %d, want a positive timestamp", st.ChangedAfter) + } + + nQueriesAfterRun1 := len(p.queries) + out2 := runExportAgainst(t, p, dir) + if !strings.Contains(out2, "appended=0") { + t.Errorf("second run must append nothing, got:\n%s", out2) + } + labels2, err := core.ReadLabels(filepath.Join(dir, "labels.jsonl")) + if err != nil { + t.Fatalf("read labels after run 2: %v", err) + } + if len(labels2) != 25 { + t.Fatalf("second run changed the file: %d rows", len(labels2)) + } + // Second run must have gone incremental: changed_after in some query + // issued by that run (the first run, with no cursor, sends none). + incremental := false + for _, qq := range p.queries[nQueriesAfterRun1:] { + decoded, _ := url.QueryUnescape(qq) + if strings.Contains(decoded, "changed_after=") { + incremental = true + } + } + if !incremental { + t.Errorf("second run sent no changed_after query: %v", p.queries[nQueriesAfterRun1:]) + } +} + +func TestExportURLDedup(t *testing.T) { + p := &pagedServer{starred: makeEntries(3, 2000, true, 7)} + dir := t.TempDir() + + // Pre-seed a label whose URL is a normalized twin of entry 2000's URL. + pre := core.LabelRow{ + EntryID: 99999, URL: "HTTPS://www.example.com/post/2000/", Title: "Old", + FeedID: 1, FeedTitle: "F", Label: core.LabelPos, + CapturedAt: time.Now(), + } + if err := core.AppendLabels(filepath.Join(dir, "labels.jsonl"), []core.LabelRow{pre}); err != nil { + t.Fatal(err) + } + + out := runExportAgainst(t, p, dir) + if !strings.Contains(out, "appended=2") || !strings.Contains(out, "url-dup=1") { + t.Errorf("expected appended=2 url-dup=1, got:\n%s", out) + } +} + +func TestExportRecFeedResolution(t *testing.T) { + // Two starred entries ON the recommendations feed (feed id 99). + rec := makeEntries(2, 3000, true, 99) + rec[0].URL = "https://example.com/post/known" + rec[1].URL = "https://example.com/post/unknown" + p := &pagedServer{starred: rec} + dir := t.TempDir() + + // The "known" URL is already labeled (as the original entry's star). + pre := core.LabelRow{ + EntryID: 1234, URL: "https://example.com/post/known", Title: "Known", + FeedID: 7, FeedTitle: "Orig", Label: core.LabelPos, + CapturedAt: time.Now(), + } + if err := core.AppendLabels(filepath.Join(dir, "labels.jsonl"), []core.LabelRow{pre}); err != nil { + t.Fatal(err) + } + + out := runExportAgainst(t, p, dir, "--rec-feed-id", "99") + if !strings.Contains(out, "rec-hit=1") || !strings.Contains(out, "rec-unresolved=1") { + t.Errorf("expected rec-hit=1 rec-unresolved=1, got:\n%s", out) + } + if !strings.Contains(out, "appended=0") { + t.Errorf("rec-feed entries must never append rows, got:\n%s", out) + } + labels, err := core.ReadLabels(filepath.Join(dir, "labels.jsonl")) + if err != nil { + t.Fatal(err) + } + if len(labels) != 1 { + t.Errorf("labels file should be untouched, has %d rows", len(labels)) + } +} + +// negFixture builds a server with `nPos` fresh starred entries and a read +// pool of `nFresh` entries inside the typical 45-day window plus `nOld` +// entries far outside it (sorted after the fresh ones, as published-desc +// would order them). +func negFixture(nPos, nFresh, nOld int) *pagedServer { + now := time.Now() + starred := makeEntries(nPos, 1000, true, 7) + for i := range starred { + starred[i].PublishedAt = now.Add(-24 * time.Hour) + } + fresh := makeEntries(nFresh, 5000, false, 8) + for i := range fresh { + fresh[i].PublishedAt = now.Add(-10 * 24 * time.Hour) + } + old := makeEntries(nOld, 6000, false, 8) + for i := range old { + old[i].PublishedAt = now.Add(-100 * 24 * time.Hour) + } + return &pagedServer{starred: starred, read: append(fresh, old...)} +} + +func negEntryIDs(t *testing.T, dir string) map[int64]bool { + t.Helper() + labels, err := core.ReadLabels(filepath.Join(dir, "labels.jsonl")) + if err != nil { + t.Fatal(err) + } + ids := map[int64]bool{} + for _, row := range labels { + if row.Label == core.LabelNeg { + ids[row.EntryID] = true + } + } + return ids +} + +func TestExportNegatives(t *testing.T) { + // 5 positives × ratio 20 = 100 negatives from 150 in-window candidates. + p := negFixture(5, 150, 10) + dir := t.TempDir() + + out := runExportAgainst(t, p, dir) + if !strings.Contains(out, "candidates=150") || !strings.Contains(out, "appended=100") { + t.Errorf("expected candidates=150 appended=100, got:\n%s", out) + } + + labels, err := core.ReadLabels(filepath.Join(dir, "labels.jsonl")) + if err != nil { + t.Fatal(err) + } + posIDs := map[int64]bool{} + posURLs := map[string]bool{} + negCount := 0 + for _, row := range labels { + if row.Label == core.LabelPos { + posIDs[row.EntryID] = true + posURLs[core.NormalizeURL(row.URL)] = true + } + } + for _, row := range labels { + if row.Label != core.LabelNeg { + continue + } + negCount++ + if posIDs[row.EntryID] { + t.Errorf("negative %d collides with a positive ID", row.EntryID) + } + if posURLs[core.NormalizeURL(row.URL)] { + t.Errorf("negative %d collides with a positive URL", row.EntryID) + } + if row.EntryID >= 6000 { + t.Errorf("old (out-of-window) entry %d sampled as negative", row.EntryID) + } + } + if negCount != 100 { + t.Errorf("expected 100 negatives, got %d", negCount) + } + + // Run summary (latest run, overwritten each time). + body, err := os.ReadFile(filepath.Join(dir, "export_run.json")) + if err != nil { + t.Fatalf("run summary not written: %v", err) + } + var rs runSummary + if err := json.Unmarshal(body, &rs); err != nil { + t.Fatalf("run summary unparseable: %v", err) + } + if rs.PositivesNew != 5 || rs.NegNew != 100 || rs.PosTotal != 5 || rs.NegTotal != 100 { + t.Errorf("run summary counts wrong: %+v", rs) + } + + // Second run: pool already at target, so no new negatives. + out2 := runExportAgainst(t, p, dir) + if !strings.Contains(out2, "appended=0") { + t.Errorf("second run should append nothing, got:\n%s", out2) + } +} + +func TestExportNegativesSeededDeterministic(t *testing.T) { + dir1, dir2 := t.TempDir(), t.TempDir() + runExportAgainst(t, negFixture(4, 120, 0), dir1) + runExportAgainst(t, negFixture(4, 120, 0), dir2) + + ids1, ids2 := negEntryIDs(t, dir1), negEntryIDs(t, dir2) + if len(ids1) != len(ids2) || len(ids1) == 0 { + t.Fatalf("negative counts differ or zero: %d vs %d", len(ids1), len(ids2)) + } + for id := range ids1 { + if !ids2[id] { + t.Fatalf("same seed + same pool picked different samples; %d missing in run 2", id) + } + } +} + +// ── dump-raw tests ── + +// fakeMiniflux serves small responses for fixture capture and records the +// query of each request. +type fakeMiniflux struct { + queries []string +} + +func (f *fakeMiniflux) handler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + f.queries = append(f.queries, r.URL.RawQuery) + w.Header().Set("Content-Type", "application/json") + if r.URL.Query().Get("starred") == "true" { + fmt.Fprint(w, entriesJSON(2, makeEntries(2, 1000, true, 1))) + } else { + fmt.Fprint(w, entriesJSON(3, makeEntries(3, 2000, false, 1))) + } + } +} + +func TestExportDumpRawWritesFixtures(t *testing.T) { + fake := &fakeMiniflux{} + srv := httptest.NewServer(fake.handler()) + defer srv.Close() + t.Setenv("MINIFLUX_URL", srv.URL) + t.Setenv("MINIFLUX_TOKEN", "test-token") + + dir := t.TempDir() + cmd := &ExportCommand{} + if err := cmd.Init([]string{"--dump-raw", dir}); err != nil { + t.Fatalf("Init: %v", err) + } + var out bytes.Buffer + if err := cmd.Run(nil, &out); err != nil { + t.Fatalf("Run: %v", err) + } + + for _, want := range []struct { + file string + nEntries int + }{ + {"starred_sample.json", 2}, + {"read_sample.json", 3}, + } { + path := filepath.Join(dir, want.file) + body, err := os.ReadFile(path) + if err != nil { + t.Fatalf("fixture %s not written: %v", want.file, err) + } + var set miniflux.EntrySet + if err := json.Unmarshal(body, &set); err != nil { + t.Fatalf("fixture %s is not valid entries JSON: %v", want.file, err) + } + if len(set.Entries) != want.nEntries { + t.Errorf("%s: got %d entries, want %d", want.file, len(set.Entries), want.nEntries) + } + } + + if len(fake.queries) != 2 { + t.Fatalf("expected 2 API calls, got %d", len(fake.queries)) + } + if !strings.Contains(fake.queries[0], "starred=true") { + t.Errorf("first call should filter starred=true, got %q", fake.queries[0]) + } + if !strings.Contains(fake.queries[1], "status=read") { + t.Errorf("second call should filter status=read, got %q", fake.queries[1]) + } + if !strings.Contains(fake.queries[1], "starred=false") { + t.Errorf("negatives call must filter starred=false, got %q", fake.queries[1]) + } + + if !strings.Contains(out.String(), "starred_sample.json") || + !strings.Contains(out.String(), "read_sample.json") { + t.Errorf("stdout should name both fixtures, got:\n%s", out.String()) + } +} diff --git a/cmds/retrostar.go b/cmds/retrostar.go new file mode 100644 index 0000000..37f9313 --- /dev/null +++ b/cmds/retrostar.go @@ -0,0 +1,127 @@ +// Retrostar command: retroactive-starring aid for cold start. Scores the +// recent read+unstarred entry pool with the current model and prints the +// highest-ranked candidates so they can be starred by hand in the Miniflux +// UI. Entries already in labels.jsonl are excluded: an exported negative +// can never be re-labeled (labels never change) and an exported positive +// is already known. Read-only — it never writes to Miniflux; starring +// stays a human action in the UI. +package cmds + +import ( + "flag" + "fmt" + "io" + "sort" + "time" + + "fluxrec/core" + "fluxrec/miniflux" +) + +type RetrostarCommand struct { + ModelPath string + Labels string + Days int + Limit int +} + +func (c *RetrostarCommand) Name() string { return "retrostar" } + +func (c *RetrostarCommand) Init(args []string) error { + fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError) + fs.Usage = func() { + fmt.Fprint(fs.Output(), `Usage: fluxrec retrostar [flags] + +Print the highest-scored recent read+unstarred entries (model-ranked) as +candidates for retroactive starring in the Miniflux UI. Entries already in +labels.jsonl are skipped. Read-only: nothing is written to Miniflux. +`) + fs.PrintDefaults() + } + fs.StringVar(&c.ModelPath, "model", "model.json", "Path to model.json") + fs.StringVar(&c.Labels, "labels", "labels.jsonl", "Path to labels.jsonl (already-labeled entries are excluded)") + fs.IntVar(&c.Days, "days", 60, "Only consider entries published within this many days (inside the deletion horizon)") + fs.IntVar(&c.Limit, "limit", 50, "How many top candidates to print") + if err := fs.Parse(args); err != nil { + return err + } + if c.Days < 1 { + return fmt.Errorf("--days must be >= 1, got %d", c.Days) + } + if c.Limit < 1 { + return fmt.Errorf("--limit must be >= 1, got %d", c.Limit) + } + return nil +} + +func (c *RetrostarCommand) Run(stdin io.Reader, stdout io.Writer) error { + cfg, err := miniflux.ConfigFromEnv() + if err != nil { + return err + } + client := miniflux.NewClient(cfg) + + model, err := loadModel(c.ModelPath) + if err != nil { + return err + } + vec := core.CreateVectorizerFromModel(model) + + existing, err := core.ReadLabels(c.Labels) + if err != nil { + return err + } + labeled := make(map[int64]bool, len(existing)) + for _, row := range existing { + labeled[row.EntryID] = true + } + + cutoff := time.Now().Add(-time.Duration(c.Days) * 24 * time.Hour) + unstarred := false + f := miniflux.Filters{ + Starred: &unstarred, Status: "read", Limit: 200, + PublishedAfter: cutoff.Unix(), + } + var scored []scoredEntry + for { + set, err := client.Entries(f) + if err != nil { + return err + } + if len(set.Entries) == 0 { + break + } + for _, e := range set.Entries { + if labeled[e.ID] { + continue + } + vecs := vec.Transform([]string{core.ScoringText(e.Title, e.Feed.Title)}) + s, err := core.PredictScore(vecs[0], model.Weights) + if err != nil { + return fmt.Errorf("scoring entry %d: %w", e.ID, err) + } + scored = append(scored, scoredEntry{entry: e, score: s}) + } + f.Offset += len(set.Entries) + if f.Offset >= set.Total { + break + } + } + + sort.SliceStable(scored, func(i, j int) bool { + if scored[i].score == scored[j].score { + return scored[i].entry.ID < scored[j].entry.ID + } + return scored[i].score > scored[j].score + }) + + shown := min(c.Limit, len(scored)) + for _, se := range scored[:shown] { + e := se.entry + fmt.Fprintf(stdout, "%.3f %s %s %s\n %s (entry %d)\n", + se.score, e.PublishedAt.Format("2006-01-02"), e.Feed.Title, e.Title, e.URL, e.ID) + } + fmt.Fprintf(stdout, "retrostar: scanned %d unlabeled candidates, showing %d — star in the Miniflux UI, then run export\n", + len(scored), shown) + return nil +} diff --git a/cmds/retrostar_test.go b/cmds/retrostar_test.go new file mode 100644 index 0000000..3ad6dca --- /dev/null +++ b/cmds/retrostar_test.go @@ -0,0 +1,64 @@ +// Test for `fluxrec retrostar` against an httptest fake Miniflux: the +// printed candidates are model-ranked and anything already in labels.jsonl +// is excluded. Helpers (testEntry, makeEntries, pagedServer) come from +// export_test.go. +package cmds + +import ( + "encoding/json" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "fluxrec/core" +) + +func TestRetrostar(t *testing.T) { + // pool has 20 read+unstarred candidates (IDs 100-119) + fake := &pagedServer{read: makeEntries(20, 100, false, 1)} + srv := httptest.NewServer(fake.handler()) + defer srv.Close() + t.Setenv("MINIFLUX_URL", srv.URL) + t.Setenv("MINIFLUX_TOKEN", "test-token") + + dir := t.TempDir() + + // Two candidates are already labeled and must be excluded. + labelsPath := filepath.Join(dir, "labels.jsonl") + now := time.Date(2024, 3, 1, 0, 0, 0, 0, time.UTC) + if err := core.AppendLabels(labelsPath, []core.LabelRow{ + {EntryID: 105, URL: "https://example.com/post/105", Title: "Title 105", + FeedID: 1, FeedTitle: "Fixture Feed", Label: core.LabelPos, CapturedAt: now}, + {EntryID: 110, URL: "https://example.com/post/110", Title: "Title 110", + FeedID: 1, FeedTitle: "Fixture Feed", Label: core.LabelNeg, CapturedAt: now}, + }); err != nil { + t.Fatalf("AppendLabels: %v", err) + } + + modelPath := filepath.Join(dir, "model.json") + body, _ := json.Marshal(testModel()) + if err := os.WriteFile(modelPath, body, 0o644); err != nil { + t.Fatalf("writing model: %v", err) + } + + cmd := &RetrostarCommand{ModelPath: modelPath, Labels: labelsPath, Days: 100000, Limit: 5} + var out strings.Builder + if err := cmd.Run(nil, &out); err != nil { + t.Fatalf("retrostar: %v", err) + } + got := out.String() + if !strings.Contains(got, "entry 100") { + t.Errorf("output missing top candidate 100:\n%s", got) + } + for _, excluded := range []string{"entry 105", "entry 110"} { + if strings.Contains(got, excluded) { + t.Errorf("output contains already-labeled %s:\n%s", excluded, got) + } + } + if !strings.Contains(got, "scanned 18 unlabeled candidates, showing 5") { + t.Errorf("summary line wrong:\n%s", got) + } +} diff --git a/cmds/rss.go b/cmds/rss.go new file mode 100644 index 0000000..3c49513 --- /dev/null +++ b/cmds/rss.go @@ -0,0 +1,82 @@ +// RSS rendering for the recommendations feed: hand-rolled RSS 2.0 via +// encoding/xml, no feed library. The render source is the runs.jsonl tail; +// items appear newest batch first, kept order within a batch. Guid embeds +// the original Miniflux entry ID (mf:<id>) for traceability. +package cmds + +import ( + "encoding/xml" + "fmt" + "time" + + "fluxrec/core" +) + +const rssXMLHeader = `<?xml version="1.0" encoding="UTF-8"?>` + "\n" + +type rssGUID struct { + IsPermaLink bool `xml:"isPermaLink,attr"` + Value string `xml:",chardata"` +} + +type rssItem struct { + Title string `xml:"title"` + Link string `xml:"link"` + GUID rssGUID `xml:"guid"` + PubDate string `xml:"pubDate"` + Description string `xml:"description"` +} + +type rssChannel struct { + Title string `xml:"title"` + Description string `xml:"description"` + Items []rssItem `xml:"item"` +} + +type rssDocument struct { + XMLName xml.Name `xml:"rss"` + Version string `xml:"version,attr"` + Channel rssChannel `xml:"channel"` +} + +// RenderRSS renders runs (oldest to newest, as read from runs.jsonl) as an +// RSS 2.0 document. The source feed title prefixes each item description so +// provenance is visible in-reader; item content was already cleaned when +// the run row was written. +func RenderRSS(runs []core.RunRow) ([]byte, error) { + doc := rssDocument{ + Version: "2.0", + Channel: rssChannel{ + Title: "fluxrec recommendations", + Description: "Top-scored Miniflux entries, refreshed on a schedule.", + }, + } + for i := len(runs) - 1; i >= 0; i-- { + for _, k := range runs[i].Kept { + doc.Channel.Items = append(doc.Channel.Items, rssItem{ + Title: k.Title, + Link: k.URL, + GUID: rssGUID{IsPermaLink: false, Value: fmt.Sprintf("mf:%d", k.EntryID)}, + PubDate: k.PublishedAt.Format(time.RFC1123Z), + Description: rssDescription(k), + }) + } + } + body, err := xml.MarshalIndent(doc, "", " ") + if err != nil { + return nil, fmt.Errorf("marshaling RSS: %w", err) + } + return append([]byte(rssXMLHeader), body...), nil +} + +// rssDescription builds the item body: source feed name on the first line, +// then the cleaned content (if any). +func rssDescription(k core.KeptItem) string { + if k.Content == "" { + return k.FeedTitle + } + if k.FeedTitle == "" { + return k.Content + } + return k.FeedTitle + "\n\n" + k.Content +} diff --git a/cmds/rss_test.go b/cmds/rss_test.go new file mode 100644 index 0000000..406f853 --- /dev/null +++ b/cmds/rss_test.go @@ -0,0 +1,73 @@ +// Golden test for the RSS renderer: fixed runs in, exact XML out. Also +// pins batch ordering (newest first) and XML escaping of item text. +package cmds + +import ( + "testing" + "time" + + "fluxrec/core" +) + +func TestRenderRSSGolden(t *testing.T) { + at := func(day, h, m int) time.Time { + return time.Date(2024, 3, day, h, m, 0, 0, time.UTC) + } + runs := []core.RunRow{ + { + RanAt: at(1, 6, 0), EntriesSeen: 40, + Kept: []core.KeptItem{ + {EntryID: 101, URL: "https://example.com/a?x=1&y=2", Title: "Alpha & Omega", + FeedTitle: "Feed A", PublishedAt: at(1, 5, 0), + Content: "x < y", Score: 0.9, Kind: core.KindRanked}, + {EntryID: 102, URL: "https://example.com/b", Title: "Beta", + FeedTitle: "Feed B", PublishedAt: at(1, 5, 30), + Content: "second body", Score: 0.2, Kind: core.KindRandom}, + }, + }, + { + RanAt: at(1, 12, 0), EntriesSeen: 12, + Kept: []core.KeptItem{ + {EntryID: 103, URL: "https://example.com/c", Title: "Gamma", + FeedTitle: "Feed A", PublishedAt: at(1, 11, 0), + Content: "", Score: 0.8, Kind: core.KindRunnerUp}, + }, + }, + } + + got, err := RenderRSS(runs) + if err != nil { + t.Fatalf("RenderRSS: %v", err) + } + want := `<?xml version="1.0" encoding="UTF-8"?> +<rss version="2.0"> + <channel> + <title>fluxrec recommendations</title> + <description>Top-scored Miniflux entries, refreshed on a schedule.</description> + <item> + <title>Gamma</title> + <link>https://example.com/c</link> + <guid isPermaLink="false">mf:103</guid> + <pubDate>Fri, 01 Mar 2024 11:00:00 +0000</pubDate> + <description>Feed A</description> + </item> + <item> + <title>Alpha & Omega</title> + <link>https://example.com/a?x=1&y=2</link> + <guid isPermaLink="false">mf:101</guid> + <pubDate>Fri, 01 Mar 2024 05:00:00 +0000</pubDate> + <description>Feed A

x < y</description> + </item> + <item> + <title>Beta</title> + <link>https://example.com/b</link> + <guid isPermaLink="false">mf:102</guid> + <pubDate>Fri, 01 Mar 2024 05:30:00 +0000</pubDate> + <description>Feed B

second body</description> + </item> + </channel> +</rss>` + if string(got) != want { + t.Errorf("RSS mismatch.\ngot:\n%s", got) + } +} diff --git a/cmds/score.go b/cmds/score.go new file mode 100644 index 0000000..da6167a --- /dev/null +++ b/cmds/score.go @@ -0,0 +1,103 @@ +// Score command: debugging utility, one trained model vs. stdin titles. +// +// Reads one title per line on stdin, writes {"title", "score"} JSONL on +// stdout. Exists to make the vendored ML core exercisable end-to-end before +// the real commands (export/train/serve) land, and stays useful later for +// poking at model behavior by hand. +package cmds + +import ( + "bufio" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "os" + "path/filepath" + "strings" + + "fluxrec/core" +) + +type ScoreCommand struct { + ModelPath string +} + +func (c *ScoreCommand) Name() string { return "score" } + +func (c *ScoreCommand) Init(args []string) error { + fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError) + fs.Usage = func() { + fmt.Fprint(fs.Output(), `Usage: fluxrec score --model MODEL + +Score stdin titles (one per line) with a trained model, JSONL to stdout. +`) + fs.PrintDefaults() + } + fs.StringVar(&c.ModelPath, "model", "", "Path to model.json (required)") + if err := fs.Parse(args); err != nil { + return err + } + if c.ModelPath == "" { + return errors.New("--model flag is required") + } + clean := filepath.Clean(c.ModelPath) + if strings.Contains(clean, "..") { + return fmt.Errorf("invalid model path %q: must not contain '..'", c.ModelPath) + } + c.ModelPath = clean + return nil +} + +type scoreResult struct { + Title string `json:"title"` + Score float64 `json:"score"` +} + +func (c *ScoreCommand) Run(stdin io.Reader, stdout io.Writer) error { + model, err := loadModel(c.ModelPath) + if err != nil { + return err + } + vectorizer := core.CreateVectorizerFromModel(model) + + enc := json.NewEncoder(stdout) + scanner := bufio.NewScanner(stdin) + scanner.Buffer(make([]byte, 64*1024), 1024*1024) + for scanner.Scan() { + title := strings.TrimSpace(scanner.Text()) + if title == "" { + continue + } + if len(title) > 1000 { + return fmt.Errorf("title exceeds 1000 chars") + } + vec := vectorizer.Transform([]string{title}) + score, err := core.PredictScore(vec[0], model.Weights) + if err != nil { + return fmt.Errorf("scoring %q: %w", title, err) + } + if err := enc.Encode(scoreResult{Title: title, Score: score}); err != nil { + return err + } + } + return scanner.Err() +} + +func loadModel(path string) (*core.ModelEnvelope, error) { + f, err := os.Open(path) + if err != nil { + return nil, fmt.Errorf("opening model: %w", err) + } + defer func() { _ = f.Close() }() + + var model core.ModelEnvelope + if err := json.NewDecoder(f).Decode(&model); err != nil { + return nil, fmt.Errorf("parsing model %s: %w", path, err) + } + if len(model.Weights) == 0 || len(model.OrderedVocab) == 0 { + return nil, fmt.Errorf("model %s is missing weights or vocabulary", path) + } + return &model, nil +} diff --git a/cmds/serve.go b/cmds/serve.go new file mode 100644 index 0000000..9d82d38 --- /dev/null +++ b/cmds/serve.go @@ -0,0 +1,452 @@ +// Serve command: the long-running server lane. Polls Miniflux for entries +// newer than a persisted cursor, scores each one exactly once, assembles a +// top-N + exploration batch per refresh into runs.jsonl, and serves the +// recent batches as an RSS feed plus a JSON status endpoint. Strictly +// read-only toward Miniflux (the client has no mutation methods). +// +// Scoring text is the same construction as training: title plus the +// feed-identity token. Batch = top-N by raw score (no threshold cutoff) +// with the exploration budget split at a fixed 1/3 runner-up + 2/3 random. +package cmds + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "math" + "math/rand" + "net/http" + "os" + "os/signal" + "sort" + "strconv" + "strings" + "sync/atomic" + "syscall" + "time" + + "fluxrec/core" + "fluxrec/miniflux" +) + +type ServeCommand struct { + ModelPath string + Cursor string + Runs string + Addr string + RecFeedID int64 + TopN int + ExploreFrac float64 + RSSBatches int + RefreshAt string + + pollsOK atomic.Int64 + pollsFailed atomic.Int64 +} + +func (c *ServeCommand) Name() string { return "serve" } + +func (c *ServeCommand) Init(args []string) error { + fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError) + fs.Usage = func() { + fmt.Fprint(fs.Output(), `Usage: fluxrec serve [flags] + +Poll Miniflux for new entries, score each once, keep a top-N + exploration +batch per refresh in runs.jsonl, and serve the recent batches as RSS at +/recommendations.xml plus JSON status at /api/status. Read-only toward +Miniflux. Refresh times are wall-clock in the process timezone ($TZ, +default UTC). Exits immediately if the model file is missing or corrupt. +`) + fs.PrintDefaults() + } + fs.StringVar(&c.ModelPath, "model", "model.json", "Path to model.json") + fs.StringVar(&c.Cursor, "cursor", "cursor.json", "Path to the poll cursor file") + fs.StringVar(&c.Runs, "runs", "runs.jsonl", "Path to the run log (doubles as the RSS render source)") + fs.StringVar(&c.Addr, "addr", ":8080", "Listen address for RSS + status") + fs.Int64Var(&c.RecFeedID, "rec-feed-id", 0, "Miniflux feed ID of the recommendations feed itself; its entries are skipped so serve never re-recommends its own output (0 = not subscribed yet)") + fs.IntVar(&c.TopN, "top-n", 15, "Batch size: ranked picks plus exploration") + fs.Float64Var(&c.ExploreFrac, "explore-frac", 0.2, "Fraction of the batch spent on exploration (1/3 runner-up, 2/3 random)") + fs.IntVar(&c.RSSBatches, "rss-batches", 21, "How many recent batches the RSS feed renders") + fs.StringVar(&c.RefreshAt, "refresh-at", "06:00,12:00,18:00", "Comma-separated daily refresh times (HH:MM, local time)") + if err := fs.Parse(args); err != nil { + return err + } + if c.TopN < 1 { + return fmt.Errorf("--top-n must be >= 1, got %d", c.TopN) + } + if c.ExploreFrac < 0 || c.ExploreFrac >= 1 { + return fmt.Errorf("--explore-frac must be in [0,1), got %v", c.ExploreFrac) + } + if c.RSSBatches < 1 { + return fmt.Errorf("--rss-batches must be >= 1, got %d", c.RSSBatches) + } + if _, err := parseRefreshAt(c.RefreshAt); err != nil { + return err + } + return nil +} + +func (c *ServeCommand) Run(stdin io.Reader, stdout io.Writer) error { + cfg, err := miniflux.ConfigFromEnv() + if err != nil { + return err + } + client := miniflux.NewClient(cfg) + + model, err := loadModel(c.ModelPath) + if err != nil { + return err + } + vec := core.CreateVectorizerFromModel(model) + mins, _ := parseRefreshAt(c.RefreshAt) // validated in Init + + rng := rand.New(rand.NewSource(time.Now().UnixNano())) + poll := func() { + if err := c.pollOnce(client, vec, model, rng, stdout); err != nil { + c.pollsFailed.Add(1) + fmt.Fprintf(os.Stderr, "serve: poll failed (cursor not advanced, will retry next cycle): %v\n", err) + } else { + c.pollsOK.Add(1) + } + } + + // Boot catch-up: if the last logged run is older than half the smallest + // gap between refreshes (e.g. the container was down across a slot), + // poll immediately rather than waiting for the next scheduled time. + due, err := catchUpDue(c.Runs, mins, time.Now()) + if err != nil { + return err + } + if due { + poll() + } + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + srv := &http.Server{ + Addr: c.Addr, + Handler: c.routes(mins, model.CreatedAt), + ReadTimeout: core.DefaultReadTimeout, + WriteTimeout: core.DefaultWriteTimeout, + IdleTimeout: core.DefaultIdleTimeout, + } + go func() { + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + fmt.Fprintf(os.Stderr, "serve: http listener failed: %v\n", err) + stop() + } + }() + + go func() { + for { + timer := time.NewTimer(time.Until(nextRefresh(time.Now(), mins))) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + poll() + } + } + }() + + fmt.Fprintf(stdout, "serve: listening on %s (refresh at %s, model %s)\n", c.Addr, c.RefreshAt, c.ModelPath) + <-ctx.Done() + + shutdownCtx, cancel := context.WithTimeout(context.Background(), core.DefaultShutdownTimeout) + defer cancel() + return srv.Shutdown(shutdownCtx) +} + +// pollCursor is the persisted serve cursor: the largest entry ID fully +// processed so far. +type pollCursor struct { + AfterEntryID int64 `json:"after_entry_id"` +} + +func loadCursor(path string) (int64, error) { + body, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return 0, nil + } + return 0, fmt.Errorf("reading cursor %s: %w", path, err) + } + var c pollCursor + if err := json.Unmarshal(body, &c); err != nil { + return 0, fmt.Errorf("parsing cursor %s: %w", path, err) + } + return c.AfterEntryID, nil +} + +func saveCursor(path string, id int64) error { + body, err := json.Marshal(pollCursor{AfterEntryID: id}) + if err != nil { + return err + } + if err := os.WriteFile(path, body, 0o644); err != nil { + return fmt.Errorf("writing cursor %s: %w", path, err) + } + return nil +} + +// pollOnce fetches every entry newer than the cursor, scores each one, +// assembles the batch, appends one row to runs.jsonl, and only then +// advances the cursor — process → append log → save cursor, so a crash at +// any point re-processes at worst one batch. Nothing is written for an +// empty poll (a run of empty batches would starve the count-based RSS +// tail). +func (c *ServeCommand) pollOnce(client *miniflux.Client, vec *core.TFIDFVectorizer, model *core.ModelEnvelope, rng *rand.Rand, stdout io.Writer) error { + started := time.Now() + cursor, err := loadCursor(c.Cursor) + if err != nil { + return err + } + + var scored []scoredEntry + seen, recSkipped := 0, 0 + maxID := cursor + f := miniflux.Filters{AfterEntryID: cursor, Limit: 200, Direction: "asc"} + for { + set, err := client.Entries(f) + if err != nil { + return err + } + if len(set.Entries) == 0 { + break + } + for _, e := range set.Entries { + seen++ + if e.ID > maxID { + maxID = e.ID + } + // The recommendations feed's own items are our output, not + // candidates: skip them or the loop feeds itself. + if c.RecFeedID != 0 && e.Feed.ID == c.RecFeedID { + recSkipped++ + continue + } + vecs := vec.Transform([]string{core.ScoringText(e.Title, e.Feed.Title)}) + s, err := core.PredictScore(vecs[0], model.Weights) + if err != nil { + return fmt.Errorf("scoring entry %d: %w", e.ID, err) + } + scored = append(scored, scoredEntry{entry: e, score: s}) + } + if len(set.Entries) < f.Limit || maxID == f.AfterEntryID { + // Short page: done. Unchanged max ID despite a full page means + // the server ignored the cursor — break rather than spin. + break + } + f.AfterEntryID = maxID + } + + if seen == 0 { + fmt.Fprintln(stdout, "serve: no new entries") + return nil + } + + row := core.RunRow{ + RanAt: started.UTC(), + EntriesSeen: seen, + RecFeedSkipped: recSkipped, + Kept: assembleBatch(scored, c.TopN, c.ExploreFrac, rng), + DurationMs: time.Since(started).Milliseconds(), + ModelCreatedAt: model.CreatedAt, + } + if err := core.AppendRun(c.Runs, row); err != nil { + return err + } + if err := saveCursor(c.Cursor, maxID); err != nil { + return err + } + fmt.Fprintf(stdout, "serve: seen=%d kept=%d skipped-rec=%d cursor=%d\n", + seen, len(row.Kept), recSkipped, maxID) + return nil +} + +// scoredEntry is an entry with its model score; batch assembly input. +type scoredEntry struct { + entry miniflux.Entry + score float64 +} + +// assembleBatch ranks scored entries by score descending and selects the +// batch: the top (topN minus the exploration budget) are ranked; the +// budget splits at a fixed 1/3 : 2/3 into runner-ups (highest-scored just +// below the cut — the model's borderline cases) and random picks from the +// remaining tail (serendipity and a recall-ceiling estimator). With fewer +// candidates than the ranked cut, everything is kept as ranked. +func assembleBatch(entries []scoredEntry, topN int, exploreFrac float64, rng *rand.Rand) []core.KeptItem { + ranked := make([]scoredEntry, len(entries)) + copy(ranked, entries) + sort.SliceStable(ranked, func(i, j int) bool { + if ranked[i].score == ranked[j].score { + return ranked[i].entry.ID < ranked[j].entry.ID + } + return ranked[i].score > ranked[j].score + }) + + explore := int(math.Round(float64(topN) * exploreFrac)) + nRanked := topN - explore + if len(ranked) <= nRanked { + return keptFrom(ranked, core.KindRanked) + } + kept := keptFrom(ranked[:nRanked], core.KindRanked) + + nRunnerUp := min(explore/3, len(ranked)-nRanked) + kept = append(kept, keptFrom(ranked[nRanked:nRanked+nRunnerUp], core.KindRunnerUp)...) + + rest := ranked[nRanked+nRunnerUp:] + nRandom := min(explore-nRunnerUp, len(rest)) + if nRandom > 0 { + picks := make([]scoredEntry, 0, nRandom) + for _, i := range rng.Perm(len(rest))[:nRandom] { + picks = append(picks, rest[i]) + } + // Log random picks in ID order so runs.jsonl stays diff-friendly. + sort.Slice(picks, func(i, j int) bool { return picks[i].entry.ID < picks[j].entry.ID }) + kept = append(kept, keptFrom(picks, core.KindRandom)...) + } + return kept +} + +// keptFrom converts scored entries into run-log items, cleaning content on +// the way in so RSS rendering never sees raw feed HTML. +func keptFrom(entries []scoredEntry, kind string) []core.KeptItem { + out := make([]core.KeptItem, len(entries)) + for i, se := range entries { + out[i] = core.KeptItem{ + EntryID: se.entry.ID, + URL: se.entry.URL, + Title: se.entry.Title, + FeedTitle: se.entry.Feed.Title, + PublishedAt: se.entry.PublishedAt.UTC(), + Content: core.CleanFeedContent(se.entry.Content), + Score: se.score, + Kind: kind, + } + } + return out +} + +// parseRefreshAt parses "06:00,12:00,18:00" into ascending +// minutes-after-midnight. +func parseRefreshAt(spec string) ([]int, error) { + var mins []int + for _, part := range strings.Split(spec, ",") { + part = strings.TrimSpace(part) + hm := strings.Split(part, ":") + if len(hm) != 2 { + return nil, fmt.Errorf("--refresh-at: %q is not HH:MM", part) + } + h, errH := strconv.Atoi(hm[0]) + m, errM := strconv.Atoi(hm[1]) + if errH != nil || errM != nil || h < 0 || h > 23 || m < 0 || m > 59 { + return nil, fmt.Errorf("--refresh-at: %q is not a valid time", part) + } + mins = append(mins, h*60+m) + } + sort.Ints(mins) + return mins, nil +} + +// nextRefresh returns the next occurrence of any refresh time strictly +// after now, in now's location, wrapping to tomorrow past the last slot. +func nextRefresh(now time.Time, mins []int) time.Time { + var best time.Time + for _, m := range mins { + t := time.Date(now.Year(), now.Month(), now.Day(), m/60, m%60, 0, 0, now.Location()) + if !t.After(now) { + t = t.Add(24 * time.Hour) + } + if best.IsZero() || t.Before(best) { + best = t + } + } + return best +} + +// catchUpDue reports whether startup should trigger an immediate poll: +// nothing logged yet, or the last run is older than half the smallest gap +// between refresh times (a container down across a slot catches up on boot +// instead of waiting for the next one). +func catchUpDue(runsPath string, mins []int, now time.Time) (bool, error) { + rows, err := core.ReadRunsTail(runsPath, 1) + if err != nil { + return false, err + } + if len(rows) == 0 { + return true, nil + } + half := time.Duration(minGapMinutes(mins)) * time.Minute / 2 + return now.Sub(rows[len(rows)-1].RanAt) > half, nil +} + +// minGapMinutes returns the smallest gap between consecutive refresh times, +// including the wrap across midnight (a single time has a 24h gap). +func minGapMinutes(mins []int) int { + if len(mins) < 2 { + return 24 * 60 + } + sorted := make([]int, len(mins)) + copy(sorted, mins) + sort.Ints(sorted) + gap := 24*60 - sorted[len(sorted)-1] + sorted[0] + for i := 1; i < len(sorted); i++ { + gap = min(gap, sorted[i]-sorted[i-1]) + } + return gap +} + +// routes wires the two surfaces: the RSS feed and the JSON status. +func (c *ServeCommand) routes(mins []int, modelCreatedAt time.Time) http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/recommendations.xml", func(w http.ResponseWriter, r *http.Request) { + runs, err := core.ReadRunsTail(c.Runs, c.RSSBatches) + if err != nil { + http.Error(w, "reading runs: "+err.Error(), http.StatusInternalServerError) + return + } + body, err := RenderRSS(runs) + if err != nil { + http.Error(w, "rendering RSS: "+err.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/rss+xml; charset=utf-8") + _, _ = w.Write(body) + }) + mux.HandleFunc("/api/status", func(w http.ResponseWriter, r *http.Request) { + runs, err := core.ReadRunsTail(c.Runs, 1) + if err != nil { + http.Error(w, "reading runs: "+err.Error(), http.StatusInternalServerError) + return + } + status := map[string]any{ + "model_created_at": modelCreatedAt.UTC(), + "model_age_hours": math.Round(time.Since(modelCreatedAt).Hours()), + "next_refresh_at": nextRefresh(time.Now(), mins), + "polls_ok": c.pollsOK.Load(), + "polls_failed": c.pollsFailed.Load(), + "last_run": nil, + } + if len(runs) > 0 { + last := runs[len(runs)-1] + status["last_run"] = map[string]any{ + "ran_at": last.RanAt, + "entries_seen": last.EntriesSeen, + "rec_feed_skipped": last.RecFeedSkipped, + "kept": len(last.Kept), + "duration_ms": last.DurationMs, + "model_created_at": last.ModelCreatedAt, + } + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(status) + }) + return mux +} diff --git a/cmds/serve_test.go b/cmds/serve_test.go new file mode 100644 index 0000000..d7df2dd --- /dev/null +++ b/cmds/serve_test.go @@ -0,0 +1,407 @@ +// Tests for `fluxrec serve`: the poll loop against an httptest fake +// Miniflux (multi-page, mid-run crash), batch assembly split/kinds, and +// the refresh-time computation including the midnight wrap. Helpers +// (testEntry, makeEntries, entriesJSON) come from export_test.go. +package cmds + +import ( + "encoding/json" + "fmt" + "math/rand" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "fluxrec/core" + "fluxrec/miniflux" +) + +// testModel is a minimal hand-built model: one unigram ("title"), so every +// makeEntries title scores identically and batch ties break by entry ID. +func testModel() *core.ModelEnvelope { + return &core.ModelEnvelope{ + CreatedAt: time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC), + Meta: map[string]any{}, + Vectorizer: map[string]float64{"title": 1.0}, + OrderedVocab: []string{"title"}, + Weights: []float64{2.0, 0.1}, + } +} + +// idServer pages its entry pool by after_entry_id ascending, like the +// serve poll expects, and can be told to die on the Nth request. +type idServer struct { + entries []testEntry // ascending IDs + queries []string + calls int + failOn int // 1-based request number to 500 on; 0 = never +} + +func (s *idServer) handler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + s.calls++ + s.queries = append(s.queries, r.URL.RawQuery) + if s.failOn == s.calls { + http.Error(w, "boom", http.StatusInternalServerError) + return + } + q := r.URL.Query() + after, _ := strconv.ParseInt(q.Get("after_entry_id"), 10, 64) + limit, _ := strconv.Atoi(q.Get("limit")) + if limit <= 0 { + limit = len(s.entries) + } + var page []testEntry + for _, e := range s.entries { + if e.ID > after { + page = append(page, e) + } + } + if len(page) > limit { + page = page[:limit] + } + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, entriesJSON(len(page), page)) + } +} + +func serveTestClient(t *testing.T, s http.Handler) *miniflux.Client { + t.Helper() + srv := httptest.NewServer(s) + t.Cleanup(srv.Close) + return miniflux.NewClientWithHTTP(miniflux.Config{BaseURL: srv.URL, Token: "t"}, srv.Client()) +} + +func TestPollMultiPage(t *testing.T) { + // 450 entries => three 200/200/50 pages off one ascending cursor. + fake := &idServer{entries: makeEntries(450, 1, false, 1)} + client := serveTestClient(t, fake.handler()) + + dir := t.TempDir() + model := testModel() + cmd := &ServeCommand{ + Cursor: filepath.Join(dir, "cursor.json"), Runs: filepath.Join(dir, "runs.jsonl"), + TopN: 15, ExploreFrac: 0.2, + } + var out strings.Builder + rng := rand.New(rand.NewSource(1)) + if err := cmd.pollOnce(client, core.CreateVectorizerFromModel(model), model, rng, &out); err != nil { + t.Fatalf("pollOnce: %v", err) + } + + if fake.calls != 3 { + t.Fatalf("expected 3 paged requests, got %d", fake.calls) + } + if !strings.Contains(fake.queries[0], "direction=asc") || !strings.Contains(fake.queries[0], "limit=200") { + t.Errorf("first query missing direction/limit: %s", fake.queries[0]) + } + if strings.Contains(fake.queries[0], "after_entry_id") { + t.Errorf("first query should have no cursor: %s", fake.queries[0]) + } + if !strings.Contains(fake.queries[1], "after_entry_id=200") || + !strings.Contains(fake.queries[2], "after_entry_id=400") { + t.Errorf("cursor paging wrong: %q", fake.queries) + } + + cursor, err := loadCursor(cmd.Cursor) + if err != nil { + t.Fatalf("loadCursor: %v", err) + } + if cursor != 450 { + t.Errorf("cursor = %d, want 450", cursor) + } + + rows, err := core.ReadRuns(cmd.Runs) + if err != nil { + t.Fatalf("ReadRuns: %v", err) + } + if len(rows) != 1 { + t.Fatalf("expected 1 run row, got %d", len(rows)) + } + row := rows[0] + if row.EntriesSeen != 450 { + t.Errorf("entries_seen = %d, want 450", row.EntriesSeen) + } + if len(row.Kept) != 15 { + t.Errorf("kept = %d, want 15", len(row.Kept)) + } + if !row.ModelCreatedAt.Equal(model.CreatedAt) { + t.Errorf("model_created_at = %v, want %v", row.ModelCreatedAt, model.CreatedAt) + } +} + +func TestPollCrashLeavesNoState(t *testing.T) { + fake := &idServer{entries: makeEntries(450, 1, false, 1), failOn: 2} + client := serveTestClient(t, fake.handler()) + + dir := t.TempDir() + model := testModel() + cmd := &ServeCommand{ + Cursor: filepath.Join(dir, "cursor.json"), Runs: filepath.Join(dir, "runs.jsonl"), + TopN: 15, ExploreFrac: 0.2, + } + err := cmd.pollOnce(client, core.CreateVectorizerFromModel(model), model, + rand.New(rand.NewSource(1)), &strings.Builder{}) + if err == nil { + t.Fatal("expected pollOnce to fail on the second page") + } + // Crash-safe ordering: the page-1 work must not be half-persisted. + if _, statErr := os.Stat(cmd.Cursor); !os.IsNotExist(statErr) { + t.Error("cursor file exists despite the failed poll") + } + if _, statErr := os.Stat(cmd.Runs); !os.IsNotExist(statErr) { + t.Error("runs file exists despite the failed poll") + } +} + +func TestPollEmptyWritesNothing(t *testing.T) { + fake := &idServer{} + client := serveTestClient(t, fake.handler()) + + dir := t.TempDir() + model := testModel() + cmd := &ServeCommand{ + Cursor: filepath.Join(dir, "cursor.json"), Runs: filepath.Join(dir, "runs.jsonl"), + TopN: 15, ExploreFrac: 0.2, + } + if err := cmd.pollOnce(client, core.CreateVectorizerFromModel(model), model, + rand.New(rand.NewSource(1)), &strings.Builder{}); err != nil { + t.Fatalf("pollOnce: %v", err) + } + if rows, _ := core.ReadRuns(cmd.Runs); len(rows) != 0 { + t.Errorf("empty poll wrote %d rows, want 0 (RSS tail starvation)", len(rows)) + } +} + +func TestPollSkipsRecFeed(t *testing.T) { + fake := &idServer{ + entries: append(makeEntries(10, 1, false, 1), makeEntries(5, 100, false, 99)...), + } + client := serveTestClient(t, fake.handler()) + + dir := t.TempDir() + model := testModel() + cmd := &ServeCommand{ + Cursor: filepath.Join(dir, "cursor.json"), Runs: filepath.Join(dir, "runs.jsonl"), + TopN: 30, ExploreFrac: 0, RecFeedID: 99, + } + if err := cmd.pollOnce(client, core.CreateVectorizerFromModel(model), model, + rand.New(rand.NewSource(1)), &strings.Builder{}); err != nil { + t.Fatalf("pollOnce: %v", err) + } + rows, _ := core.ReadRuns(cmd.Runs) + if rows[0].EntriesSeen != 15 || rows[0].RecFeedSkipped != 5 { + t.Errorf("seen=%d skipped=%d, want 15 / 5", rows[0].EntriesSeen, rows[0].RecFeedSkipped) + } + for _, k := range rows[0].Kept { + if k.EntryID >= 100 { + t.Errorf("rec-feed entry %d was kept; serve must not re-recommend its own output", k.EntryID) + } + } +} + +func TestAssembleBatchSplitAndKinds(t *testing.T) { + // 30 candidates, scores increasing with ID => deterministic ranks. + entries := make([]scoredEntry, 30) + for i := range entries { + entries[i] = scoredEntry{ + entry: miniflux.Entry{ID: int64(i + 1), Title: fmt.Sprintf("T%d", i+1)}, + score: float64(i + 1), + } + } + kept := assembleBatch(entries, 15, 0.2, rand.New(rand.NewSource(7))) + if len(kept) != 15 { + t.Fatalf("kept = %d, want 15", len(kept)) + } + counts := map[string]int{} + var rankedIDs, runnerUpIDs []int64 + random := map[int64]bool{} + for _, k := range kept { + counts[k.Kind]++ + switch k.Kind { + case core.KindRanked: + rankedIDs = append(rankedIDs, k.EntryID) + case core.KindRunnerUp: + runnerUpIDs = append(runnerUpIDs, k.EntryID) + case core.KindRandom: + random[k.EntryID] = true + } + } + if counts[core.KindRanked] != 12 || counts[core.KindRunnerUp] != 1 || counts[core.KindRandom] != 2 { + t.Errorf("split = %v, want ranked=12 runnerup=1 random=2", counts) + } + // Ranked are the top 12 by score, in score order. + for i, id := range rankedIDs { + want := int64(30 - i) + if id != want { + t.Fatalf("ranked[%d] = %d, want %d", i, id, want) + } + } + // The runner-up is the highest-scored entry below the cut. + if len(runnerUpIDs) != 1 || runnerUpIDs[0] != 18 { + t.Errorf("runner-up = %v, want [18]", runnerUpIDs) + } + // Random picks come from the tail below the runner-up, no duplicates. + for id := range random { + if id > 17 { + t.Errorf("random pick %d not below the runner-up cut", id) + } + } +} + +func TestAssembleBatchFewCandidates(t *testing.T) { + entries := []scoredEntry{ + {entry: miniflux.Entry{ID: 1}, score: 0.1}, + {entry: miniflux.Entry{ID: 2}, score: 0.9}, + {entry: miniflux.Entry{ID: 3}, score: 0.5}, + } + kept := assembleBatch(entries, 15, 0.2, rand.New(rand.NewSource(1))) + if len(kept) != 3 { + t.Fatalf("kept = %d, want all 3 as ranked", len(kept)) + } + for _, k := range kept { + if k.Kind != core.KindRanked { + t.Errorf("kind = %s, want ranked when candidates < cut", k.Kind) + } + } + if kept[0].EntryID != 2 { // highest score first + t.Errorf("first kept ID = %d, want 2 (highest score)", kept[0].EntryID) + } +} + +func TestParseRefreshAt(t *testing.T) { + mins, err := parseRefreshAt("18:00,06:00, 12:30") + if err != nil { + t.Fatalf("parseRefreshAt: %v", err) + } + want := []int{360, 750, 1080} + if fmt.Sprint(mins) != fmt.Sprint(want) { + t.Errorf("mins = %v, want %v (sorted)", mins, want) + } + for _, bad := range []string{"24:00", "abc", "12", "12:60", ""} { + if _, err := parseRefreshAt(bad); err == nil { + t.Errorf("parseRefreshAt(%q) unexpectedly succeeded", bad) + } + } +} + +func TestNextRefresh(t *testing.T) { + mins := []int{360, 720, 1080} // 06:00, 12:00, 18:00 + at := func(day, h, m int) time.Time { + return time.Date(2024, 3, day, h, m, 0, 0, time.UTC) + } + cases := []struct { + now time.Time + want time.Time + }{ + {at(1, 13, 0), at(1, 18, 0)}, // same-day next slot + {at(1, 6, 0), at(1, 12, 0)}, // exactly on a slot: strictly after + {at(1, 23, 30), at(2, 6, 0)}, // past last slot: midnight wrap + {at(1, 0, 0), at(1, 6, 0)}, // before all slots + } + for _, tc := range cases { + if got := nextRefresh(tc.now, mins); !got.Equal(tc.want) { + t.Errorf("nextRefresh(%v) = %v, want %v", tc.now, got, tc.want) + } + } +} + +func TestMinGapMinutes(t *testing.T) { + if got := minGapMinutes([]int{360, 720, 1080}); got != 360 { + t.Errorf("minGap = %d, want 360", got) + } + if got := minGapMinutes([]int{1380, 60}); got != 120 { + t.Errorf("minGap across midnight = %d, want 120", got) + } + if got := minGapMinutes([]int{360}); got != 1440 { + t.Errorf("single slot minGap = %d, want 1440", got) + } +} + +func TestCatchUpDue(t *testing.T) { + mins := []int{360, 720, 1080} // min gap 6h => half = 3h + now := time.Date(2024, 3, 2, 12, 0, 0, 0, time.UTC) + dir := t.TempDir() + runs := filepath.Join(dir, "runs.jsonl") + + // No log yet: first boot polls immediately. + if due, err := catchUpDue(runs, mins, now); err != nil || !due { + t.Errorf("no runs: due=%v err=%v, want true", due, err) + } + recent := core.RunRow{RanAt: now.Add(-time.Hour), EntriesSeen: 3} + if err := core.AppendRun(runs, recent); err != nil { + t.Fatalf("AppendRun: %v", err) + } + if due, _ := catchUpDue(runs, mins, now); due { + t.Error("last run 1h ago: due=true, want false (next slot will cover it)") + } + old := core.RunRow{RanAt: now.Add(-5 * time.Hour), EntriesSeen: 3} + if err := core.AppendRun(runs, old); err != nil { + t.Fatalf("AppendRun: %v", err) + } + if due, _ := catchUpDue(runs, mins, now); !due { + t.Error("last run 5h ago: due=false, want true (downtime across slots)") + } +} + +func TestServeRoutes(t *testing.T) { + dir := t.TempDir() + runsPath := filepath.Join(dir, "runs.jsonl") + row := core.RunRow{ + RanAt: time.Date(2024, 3, 1, 6, 0, 0, 0, time.UTC), + EntriesSeen: 42, + Kept: []core.KeptItem{ + {EntryID: 101, URL: "https://example.com/a", Title: "Alpha", FeedTitle: "Feed A", + PublishedAt: time.Date(2024, 3, 1, 5, 0, 0, 0, time.UTC), Content: "body a", Score: 0.9, Kind: core.KindRanked}, + {EntryID: 102, URL: "https://example.com/b", Title: "Beta", FeedTitle: "Feed B", + PublishedAt: time.Date(2024, 3, 1, 5, 30, 0, 0, time.UTC), Content: "body b", Score: 0.2, Kind: core.KindRandom}, + }, + } + if err := core.AppendRun(runsPath, row); err != nil { + t.Fatalf("AppendRun: %v", err) + } + + cmd := &ServeCommand{Runs: runsPath, RSSBatches: 21} + handler := cmd.routes([]int{360, 720, 1080}, time.Now().Add(-48*time.Hour)) + + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/recommendations.xml", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("RSS status = %d", rec.Code) + } + if ct := rec.Header().Get("Content-Type"); !strings.Contains(ct, "application/rss+xml") { + t.Errorf("RSS content-type = %q", ct) + } + body := rec.Body.String() + if !strings.Contains(body, "<guid isPermaLink=\"false\">mf:101</guid>") { + t.Error("RSS missing mf:101 guid") + } + if strings.Count(body, "<item>") != 2 { + t.Errorf("RSS item count = %d, want 2", strings.Count(body, "<item>")) + } + + rec = httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/status", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d", rec.Code) + } + var status map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &status); err != nil { + t.Fatalf("status not JSON: %v", err) + } + last, ok := status["last_run"].(map[string]any) + if !ok { + t.Fatalf("status last_run missing: %v", status) + } + if last["entries_seen"].(float64) != 42 || last["kept"].(float64) != 2 { + t.Errorf("last_run = %v, want entries_seen=42 kept=2", last) + } + if status["model_age_hours"].(float64) <= 0 { + t.Errorf("model_age_hours = %v, want positive", status["model_age_hours"]) + } +} diff --git a/cmds/train.go b/cmds/train.go new file mode 100644 index 0000000..234ef06 --- /dev/null +++ b/cmds/train.go @@ -0,0 +1,430 @@ +// Train command: labels.jsonl → model.json + report.json. +// +// Time-split validation (train on older labels, evaluate on the newest +// window — what serve actually faces). Headline metric is precision@15 +// (rank-based: of the 15 highest-scored validation entries, what fraction +// are positives). Youden's J is computed and reported for information only; +// it is NOT used as a cutoff — serve ranks by raw score. +// +// Reproducible: a fixed seed makes the downsampling deterministic, so the +// same labels produce the same weights/vocab. IDF is fit on the full +// (unbalanced) train set to preserve document frequencies, then the +// majority class is downsampled to 1:1 before LR fit (matches the vendored +// scholscan approach). +package cmds + +import ( + "encoding/json" + "flag" + "fmt" + "io" + "math" + "math/rand" + "os" + "sort" + "time" + + "fluxrec/core" +) + +// trainSeed fixes the downsampling RNG: same labels + same seed → same model. +const trainSeed int64 = 20240301 + +type TrainCommand struct { + Labels string + ModelOut string + ReportOut string + ValFrac float64 + Lambda float64 + MinDF int + MaxDF float64 + NgramMax int +} + +func (c *TrainCommand) Name() string { return "train" } + +func (c *TrainCommand) Init(args []string) error { + fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError) + fs.Usage = func() { + fmt.Fprint(fs.Output(), `Usage: fluxrec train LABELS_JSONL [flags] + +Train a TF-IDF + logistic regression model from labels.jsonl. Writes the +model to --model-out and a validation report (precision@15 etc.) to +--report-out. Time-split: the newest --val-frac of labels (by captured_at) +are held out for validation; the rest train the model. +`) + fs.PrintDefaults() + } + fs.StringVar(&c.ModelOut, "model-out", "model.json", "Path to write the trained model JSON") + fs.StringVar(&c.ReportOut, "report-out", "report.json", "Path to write the validation report JSON") + fs.Float64Var(&c.ValFrac, "val-frac", 0.2, "Fraction of newest labels held out for validation") + fs.Float64Var(&c.Lambda, "lambda", 0.001, "L2 regularization strength") + fs.IntVar(&c.MinDF, "min-df", 2, "Minimum document frequency (absolute)") + fs.Float64Var(&c.MaxDF, "max-df", 0.8, "Maximum document frequency (ratio)") + fs.IntVar(&c.NgramMax, "ngram-max", 2, "Max n-gram size (1=unigrams, 2=+bigrams)") + + for _, a := range args { + if a == "--help" || a == "-h" { + fs.Usage() + return flag.ErrHelp + } + } + if len(args) == 0 { + return fmt.Errorf("labels.jsonl path argument is required") + } + c.Labels = args[0] + if err := fs.Parse(args[1:]); err != nil { + return err + } + if c.ValFrac <= 0 || c.ValFrac >= 1 { + return fmt.Errorf("--val-frac must be between 0 and 1, got %v", c.ValFrac) + } + return nil +} + +func (c *TrainCommand) Run(stdin io.Reader, stdout io.Writer) error { + rows, err := core.ReadLabels(c.Labels) + if err != nil { + return err + } + if len(rows) == 0 { + return fmt.Errorf("no labels in %s", c.Labels) + } + + trainRows, valRows := core.SplitByTime(rows, c.ValFrac) + if len(trainRows) == 0 || len(valRows) == 0 { + return fmt.Errorf("time split produced an empty side (train=%d val=%d); adjust --val-frac or accumulate more labels", + len(trainRows), len(valRows)) + } + valPos := 0 + for _, r := range valRows { + if r.Label == core.LabelPos { + valPos++ + } + } + if valPos < 5 { + return fmt.Errorf("only %d positives in validation (need >=5 for signal, not noise); lower --val-frac or accumulate more labels", + valPos) + } + + model, report, err := c.trainAndEvaluate(trainRows, valRows, time.Now().UTC()) + if err != nil { + return err + } + + if err := writeJSON(c.ModelOut, model); err != nil { + return err + } + if err := writeJSON(c.ReportOut, report); err != nil { + return err + } + + fmt.Fprintf(stdout, "train: pos=%d neg=%d | val: pos=%d neg=%d | precision@15=%.3f youden_j=%.3f | vocab=%d → %s\n", + report.TrainPos, report.TrainNeg, report.ValPos, report.ValNeg, + report.PrecisionAt15, report.YoudensJ, report.VocabSize, c.ModelOut) + return nil +} + +func writeJSON(path string, v any) error { + body, err := json.MarshalIndent(v, "", " ") + if err != nil { + return err + } + if err := os.WriteFile(path, body, 0o644); err != nil { + return fmt.Errorf("writing %s: %w", path, err) + } + return nil +} + +// trainAndEvaluate is the pure training core: given split rows and a clock +// value, returns the model envelope and validation report. Pure w.r.t. the +// inputs (no I/O, no wall clock) so two calls with the same rows + now are +// byte-identical — the reproducibility guarantee. +func (c *TrainCommand) trainAndEvaluate(trainRows, valRows []core.LabelRow, now time.Time) (*core.ModelEnvelope, *trainReport, error) { + trainDocs := make([]string, len(trainRows)) + trainLabels := make([]float64, len(trainRows)) + trainPos, trainNeg := 0, 0 + for i, r := range trainRows { + trainDocs[i] = core.ScoringText(r.Title, r.FeedTitle) + if r.Label == core.LabelPos { + trainLabels[i] = 1.0 + trainPos++ + } else { + trainLabels[i] = 0.0 + trainNeg++ + } + } + valDocs := make([]string, len(valRows)) + valLabels := make([]float64, len(valRows)) + valPos, valNeg := 0, 0 + for i, r := range valRows { + valDocs[i] = core.ScoringText(r.Title, r.FeedTitle) + if r.Label == core.LabelPos { + valLabels[i] = 1.0 + valPos++ + } else { + valLabels[i] = 0.0 + valNeg++ + } + } + + // Fit IDF on the full (unbalanced) train set to preserve document + // frequencies, then downsample the majority to 1:1 before LR fit. + vectorizer := &core.TFIDFVectorizer{ + NgramMin: 1, NgramMax: c.NgramMax, + MinDF: c.MinDF, MaxDF: c.MaxDF, VocabCap: 50000, + Vocabulary: make(map[string]float64), + } + vectorizer.Fit(trainDocs) + + balDocs, balLabels := downsampleBalance(trainDocs, trainLabels, trainSeed) + trainVectors := vectorizer.Transform(balDocs) + valVectors := vectorizer.Transform(valDocs) + + lr := &core.LogisticRegression{ + LearningRate: 0.5, Lambda: c.Lambda, Iterations: 500, Tolerance: 0.000001, + } + lr.Validate() + weights, err := lr.Fit(trainVectors, balLabels, map[float64]float64{1.0: 1.0, 0.0: 1.0}) + if err != nil { + return nil, nil, fmt.Errorf("training logistic regression: %w", err) + } + + precisionAt15, youdenJ, posStats, negStats := evaluate(valVectors, valLabels, weights) + + minCap, maxCap := trainRows[0].CapturedAt, trainRows[0].CapturedAt + for _, r := range trainRows { + if r.CapturedAt.Before(minCap) { + minCap = r.CapturedAt + } + if r.CapturedAt.After(maxCap) { + maxCap = r.CapturedAt + } + } + + vecParams := map[string]any{ + "ngram_min": vectorizer.NgramMin, "ngram_max": vectorizer.NgramMax, + "min_df": vectorizer.MinDF, "max_df": vectorizer.MaxDF, "vocab_cap": vectorizer.VocabCap, + } + modelParams := map[string]any{ + "learning_rate": lr.LearningRate, "lambda": lr.Lambda, + "iterations": lr.Iterations, "tolerance": lr.Tolerance, + } + + model := &core.ModelEnvelope{ + Algorithm: "tfidf-go", Impl: "go", Version: "1", + CreatedAt: now, + Meta: map[string]any{ + "scorer": "tfidf", + "seed": trainSeed, + "train_size": len(trainRows), + "val_size": len(valRows), + "train_pos": trainPos, + "train_neg": trainNeg, + "train_window_min": minCap.UTC().Format(time.RFC3339), + "train_window_max": maxCap.UTC().Format(time.RFC3339), + "val_frac": c.ValFrac, + "precision_at_15": precisionAt15, + "youdens_j": youdenJ, + "vectorizer_params": vecParams, + "model_params": modelParams, + }, + Vectorizer: vectorizer.Vocabulary, + OrderedVocab: vectorizer.OrderedVocab, + Weights: weights, + } + + report := &trainReport{ + Scorer: "tfidf", Seed: trainSeed, + TrainSize: len(trainRows), ValSize: len(valRows), + TrainPos: trainPos, TrainNeg: trainNeg, + ValPos: valPos, ValNeg: valNeg, + TrainWindowMin: minCap.UTC(), TrainWindowMax: maxCap.UTC(), + ValFrac: c.ValFrac, VocabSize: len(vectorizer.OrderedVocab), + PrecisionAt15: precisionAt15, YoudensJ: youdenJ, + PositiveScores: posStats, NegativeScores: negStats, + VectorizerParams: vecParams, ModelParams: modelParams, + } + return model, report, nil +} + +// downsampleBalance downsamples the majority class to a 1:1 ratio with the +// minority, using a seeded RNG for reproducibility. If already balanced, +// returns the inputs unchanged. +func downsampleBalance(docs []string, labels []float64, seed int64) ([]string, []float64) { + var posDocs, negDocs []string + var posLabels, negLabels []float64 + for i, label := range labels { + if label == 1.0 { + posDocs = append(posDocs, docs[i]) + posLabels = append(posLabels, label) + } else { + negDocs = append(negDocs, docs[i]) + negLabels = append(negLabels, label) + } + } + if len(posDocs) == len(negDocs) { + return docs, labels + } + var majorityDocs, minorityDocs []string + var majorityLabels, minorityLabels []float64 + if len(negDocs) > len(posDocs) { + majorityDocs, minorityDocs = negDocs, posDocs + majorityLabels, minorityLabels = negLabels, posLabels + } else { + majorityDocs, minorityDocs = posDocs, negDocs + majorityLabels, minorityLabels = posLabels, negLabels + } + minoritySize := len(minorityDocs) + rng := rand.New(rand.NewSource(seed)) + indices := make([]int, len(majorityDocs)) + for i := range indices { + indices[i] = i + } + rng.Shuffle(len(indices), func(i, j int) { + indices[i], indices[j] = indices[j], indices[i] + }) + outDocs := make([]string, 0, minoritySize*2) + outLabels := make([]float64, 0, minoritySize*2) + outDocs = append(outDocs, minorityDocs...) + outLabels = append(outLabels, minorityLabels...) + for i := 0; i < minoritySize; i++ { + outDocs = append(outDocs, majorityDocs[indices[i]]) + outLabels = append(outLabels, majorityLabels[indices[i]]) + } + return outDocs, outLabels +} + +// evaluate scores the validation set and returns precision@15 (headline), +// Youden's J (information only), and per-class score stats. +func evaluate(valVectors [][]float64, valLabels []float64, weights []float64) (precisionAt15, youdenJ float64, posStats, negStats map[string]float64) { + type scored struct { + score float64 + label float64 + idx int + } + ss := make([]scored, len(valVectors)) + for i, v := range valVectors { + s, err := core.PredictScore(v, weights) + if err != nil { + s = 0 + } + ss[i] = scored{s, valLabels[i], i} + } + + // precision@15: rank by score desc (ties by idx), take top 15, count positives. + ranked := make([]scored, len(ss)) + copy(ranked, ss) + sort.SliceStable(ranked, func(i, j int) bool { + if ranked[i].score == ranked[j].score { + return ranked[i].idx < ranked[j].idx + } + return ranked[i].score > ranked[j].score + }) + k := 15 + if k > len(ranked) { + k = len(ranked) + } + topPos := 0 + for i := 0; i < k; i++ { + if ranked[i].label == 1.0 { + topPos++ + } + } + if k > 0 { + precisionAt15 = float64(topPos) / float64(k) + } + + // Youden's J (information only): best sensitivity + specificity - 1 over + // a threshold sweep. Not used as a cutoff anywhere. + bestJ := -1.0 + for t := 5; t <= 95; t++ { + threshold := float64(t) / 100.0 + var tp, fp, tn, fn int + for _, s := range ss { + pred := s.score > threshold + switch { + case pred && s.label == 1.0: + tp++ + case pred && s.label == 0.0: + fp++ + case !pred && s.label == 1.0: + fn++ + default: + tn++ + } + } + sens := 0.0 + if tp+fn > 0 { + sens = float64(tp) / float64(tp+fn) + } + spec := 0.0 + if tn+fp > 0 { + spec = float64(tn) / float64(tn+fp) + } + if j := sens + spec - 1.0; j > bestJ { + bestJ = j + } + } + youdenJ = bestJ + + var pos, neg []float64 + for _, s := range ss { + if s.label == 1.0 { + pos = append(pos, s.score) + } else { + neg = append(neg, s.score) + } + } + posStats = scoreStats(pos) + negStats = scoreStats(neg) + return +} + +// scoreStats returns min/max/mean/std for a slice of scores. +func scoreStats(scores []float64) map[string]float64 { + if len(scores) == 0 { + return map[string]float64{"min": 0, "max": 0, "mean": 0, "std": 0} + } + min, max, sum := scores[0], scores[0], 0.0 + for _, s := range scores { + if s < min { + min = s + } + if s > max { + max = s + } + sum += s + } + mean := sum / float64(len(scores)) + variance := 0.0 + for _, s := range scores { + variance += (s - mean) * (s - mean) + } + variance /= float64(len(scores)) + return map[string]float64{ + "min": min, "max": max, "mean": mean, "std": math.Sqrt(variance), + } +} + +// trainReport is the validation report written to --report-out. +type trainReport struct { + Scorer string `json:"scorer"` + Seed int64 `json:"seed"` + TrainSize int `json:"train_size"` + ValSize int `json:"val_size"` + TrainPos int `json:"train_pos"` + TrainNeg int `json:"train_neg"` + ValPos int `json:"val_pos"` + ValNeg int `json:"val_neg"` + TrainWindowMin time.Time `json:"train_window_min"` + TrainWindowMax time.Time `json:"train_window_max"` + ValFrac float64 `json:"val_frac"` + VocabSize int `json:"vocab_size"` + PrecisionAt15 float64 `json:"precision_at_15"` + YoudensJ float64 `json:"youdens_j"` + PositiveScores map[string]float64 `json:"positive_scores"` + NegativeScores map[string]float64 `json:"negative_scores"` + VectorizerParams map[string]any `json:"vectorizer_params"` + ModelParams map[string]any `json:"model_params"` +} diff --git a/cmds/train_test.go b/cmds/train_test.go new file mode 100644 index 0000000..5a2ecee --- /dev/null +++ b/cmds/train_test.go @@ -0,0 +1,260 @@ +// Tests for `fluxrec train`: reproducibility, metrics, and model envelope +// consumption by the `score` command. +package cmds + +import ( + "bytes" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "fluxrec/core" +) + +// syntheticLabels builds a labels.jsonl file interleaving positives and +// negatives across time (so the newest validation window contains both +// classes), with varied feed titles so the feed-identity token is exercised. +// Timestamps are sequential by position: oldest first, newest last, so the +// time split's "newest val-frac" is a clean tail slice with mixed labels. +func syntheticLabels(t *testing.T, path string, nPos, nNeg int) { + t.Helper() + now := time.Now().UTC() + total := nPos + nNeg + rows := make([]core.LabelRow, total) + posTitles := []string{ + "New evidence on the rate of productivity growth in services", + "Why headline-only models are fine actually, a reanalysis", + "Field notes from sixty days of careful RSS triage at home", + "The economics of attention and the case for reading slower", + "A long retrospective on building small personal recommendation tools", + "Decade-scale trends in academic publishing, a meta-review", + "How I learned to stop worrying and love the title-only classifier", + "Notes on serendipity, exploration budgets, and ranked recall", + "Reflections on a year of minimal-information inference methods", + "The quiet resurgence of logistic regression in small data settings", + } + negTitles := []string{ + "Roundup of this week's product launches and minor updates", + "Local weather forecast for the coming weekend and beyond", + "Five ways to organize your desk for slightly better focus", + "Celebrity interview about their new streaming television show", + "Sports scores and highlights from last night's important games", + "Markets close mostly unchanged as investors await fresh data", + "Travel deals for the autumn season across several destinations", + "Cooking tips for weeknight dinners under thirty minutes flat", + "Opinion piece on city traffic and the eternal parking problem", + "Best of the app store this month, ten small utilities reviewed", + } + feeds := []string{"Marginal Revolution", "ACM Tech News", "Some Other Blog", "Hacker News", "The Atlantic"} + pi, ni := 0, 0 + for i := 0; i < total; i++ { + // Place a positive on even indices while positives remain, so + // labels interleave across the whole time range including the tail. + label := core.LabelNeg + title := "" + if i%2 == 0 && pi < nPos { + label, title = core.LabelPos, posTitles[pi%len(posTitles)] + pi++ + } else { + label, title = core.LabelNeg, negTitles[ni%len(negTitles)] + ni++ + } + rows[i] = core.LabelRow{ + EntryID: int64(i + 1), URL: "https://example.com/" + label + "/" + itoa(i), + Title: title, FeedID: int64(i%len(feeds) + 1), FeedTitle: feeds[i%len(feeds)], + Label: label, CapturedAt: now.Add(-time.Duration(total-i) * 24 * time.Hour), + } + } + if err := core.AppendLabels(path, rows); err != nil { + t.Fatalf("AppendLabels: %v", err) + } +} + +func itoa(i int) string { + b := []byte{} + if i == 0 { + return "0" + } + for i > 0 { + b = append([]byte{byte('0' + i%10)}, b...) + i /= 10 + } + return string(b) +} + +func runTrain(t *testing.T, labelsPath, dir string) (modelPath, reportPath string) { + t.Helper() + modelPath = filepath.Join(dir, "model.json") + reportPath = filepath.Join(dir, "report.json") + cmd := &TrainCommand{} + if err := cmd.Init([]string{labelsPath, "--model-out", modelPath, "--report-out", reportPath, "--val-frac", "0.3"}); err != nil { + t.Fatalf("Init: %v", err) + } + var out bytes.Buffer + if err := cmd.Run(nil, &out); err != nil { + t.Fatalf("Run: %v\nstdout: %s", err, out.String()) + } + return +} + +// B4: same labels → byte-identical model.json (seeds make training deterministic). +func TestTrainReproducible(t *testing.T) { + dir := t.TempDir() + labelsPath := filepath.Join(dir, "labels.jsonl") + // Enough rows that a 0.3 val split has >=5 positives in val. + syntheticLabels(t, labelsPath, 50, 50) + + m1, r1 := runTrain(t, labelsPath, dir) + body1, err := os.ReadFile(m1) + if err != nil { + t.Fatalf("read model1: %v", err) + } + report1, err := os.ReadFile(r1) + if err != nil { + t.Fatalf("read report1: %v", err) + } + + // Second train to a second pair of paths. + dir2 := t.TempDir() + m2, r2 := runTrain(t, labelsPath, dir2) + body2, err := os.ReadFile(m2) + if err != nil { + t.Fatalf("read model2: %v", err) + } + report2, err := os.ReadFile(r2) + if err != nil { + t.Fatalf("read report2: %v", err) + } + + // Strip CreatedAt (wall clock) before comparing models — the rest must + // be byte-identical. Compare reports fully. + var env1, env2 core.ModelEnvelope + if err := json.Unmarshal(body1, &env1); err != nil { + t.Fatal(err) + } + if err := json.Unmarshal(body2, &env2); err != nil { + t.Fatal(err) + } + env1.CreatedAt = time.Time{} + env2.CreatedAt = time.Time{} + c1, _ := json.Marshal(env1) + c2, _ := json.Marshal(env2) + if string(c1) != string(c2) { + t.Fatal("model not reproducible: weights/vocab differ across runs with same labels") + } + if string(report1) != string(report2) { + t.Fatal("report not reproducible: differs across runs with same labels") + } +} + +// B5: report.json parses and precision@15 is a valid probability. +func TestTrainReportValid(t *testing.T) { + dir := t.TempDir() + labelsPath := filepath.Join(dir, "labels.jsonl") + syntheticLabels(t, labelsPath, 50, 50) + + modelPath, reportPath := runTrain(t, labelsPath, dir) + + var report trainReport + body, err := os.ReadFile(reportPath) + if err != nil { + t.Fatalf("read report: %v", err) + } + if err := json.Unmarshal(body, &report); err != nil { + t.Fatalf("report unparseable: %v", err) + } + if report.PrecisionAt15 < 0 || report.PrecisionAt15 > 1 { + t.Errorf("precision@15 = %v, want in [0,1]", report.PrecisionAt15) + } + if report.VocabSize <= 0 { + t.Errorf("vocab size = %d, want > 0", report.VocabSize) + } + if report.TrainPos == 0 || report.TrainNeg == 0 { + t.Errorf("train should have both classes: pos=%d neg=%d", report.TrainPos, report.TrainNeg) + } + if report.ValPos < 5 { + t.Errorf("val pos = %d, guard should have required >=5", report.ValPos) + } + if report.Seed != trainSeed { + t.Errorf("seed = %d, want %d", report.Seed, trainSeed) + } + + // Model file exists and has non-empty vocab + weights. + mbody, err := os.ReadFile(modelPath) + if err != nil { + t.Fatalf("read model: %v", err) + } + var model core.ModelEnvelope + if err := json.Unmarshal(mbody, &model); err != nil { + t.Fatalf("model unparseable: %v", err) + } + if len(model.Weights) == 0 || len(model.OrderedVocab) == 0 { + t.Fatal("model missing weights or vocab") + } +} + +// B6: the produced model.json is consumable by the score command's loader +// (same ModelEnvelope type). Round-trips through CreateVectorizerFromModel. +func TestTrainModelConsumedByScore(t *testing.T) { + dir := t.TempDir() + labelsPath := filepath.Join(dir, "labels.jsonl") + syntheticLabels(t, labelsPath, 50, 50) + + modelPath, _ := runTrain(t, labelsPath, dir) + + model, err := loadModel(modelPath) // same loader the score command uses + if err != nil { + t.Fatalf("loadModel (score's loader): %v", err) + } + vec := core.CreateVectorizerFromModel(model) + title := "A long retrospective on building small personal recommendation tools" + out := vec.Transform([]string{core.ScoringText(title, "Marginal Revolution")}) + score, err := core.PredictScore(out[0], model.Weights) + if err != nil { + t.Fatalf("PredictScore: %v", err) + } + if score < 0 || score > 1 { + t.Errorf("score = %v, want in [0,1]", score) + } +} + +// Guard: too few validation positives must error loudly. +func TestTrainGuardFewValPositives(t *testing.T) { + dir := t.TempDir() + labelsPath := filepath.Join(dir, "labels.jsonl") + // 5 positives total, val-frac 0.3 → 1-2 val positives < 5. + syntheticLabels(t, labelsPath, 5, 20) + + cmd := &TrainCommand{} + if err := cmd.Init([]string{labelsPath, "--val-frac", "0.3", "--model-out", + filepath.Join(dir, "m.json"), "--report-out", filepath.Join(dir, "r.json")}); err != nil { + t.Fatalf("Init: %v", err) + } + var out bytes.Buffer + err := cmd.Run(nil, &out) + if err == nil { + t.Fatal("expected error for <5 validation positives, got nil") + } + if !strings.Contains(err.Error(), "positives in validation") { + t.Errorf("error should mention validation positives: %v", err) + } +} + +// Guard: empty labels file errors. +func TestTrainEmptyLabels(t *testing.T) { + dir := t.TempDir() + labelsPath := filepath.Join(dir, "empty.jsonl") + if err := os.WriteFile(labelsPath, []byte{}, 0o644); err != nil { + t.Fatal(err) + } + cmd := &TrainCommand{} + if err := cmd.Init([]string{labelsPath}); err != nil { + t.Fatalf("Init: %v", err) + } + if err := cmd.Run(nil, &bytes.Buffer{}); err == nil { + t.Fatal("expected error for empty labels") + } +} diff --git a/core/constants.go b/core/constants.go new file mode 100644 index 0000000..baba926 --- /dev/null +++ b/core/constants.go @@ -0,0 +1,23 @@ +// Vendored from scholscan/core/constants.go (rev 7c1a5ef, 2026-08-28). +// Owned here; no upstream sync. +// Default configuration constants. +// +// Timeouts are defensive: 30s for HTTP requests, 5s for graceful shutdown. +// Score threshold 0.5 is neutral; models should learn their own. +// MinTitleLength filters junk/broken titles (<15 chars rarely meaningful). +// ChunkSize 50 balances memory usage vs batch efficiency. +package core + +import "time" + +const ( + DefaultHTTPTimeout = 30 * time.Second + DefaultContextTimeout = 10 * time.Second + DefaultReadTimeout = 30 * time.Second + DefaultWriteTimeout = 30 * time.Second + DefaultIdleTimeout = 120 * time.Second + DefaultShutdownTimeout = 5 * time.Second + DefaultScoreThreshold = 0.5 + MinTitleLength = 15 + DefaultChunkSize = 50 +) diff --git a/core/labels.go b/core/labels.go new file mode 100644 index 0000000..bdff515 --- /dev/null +++ b/core/labels.go @@ -0,0 +1,127 @@ +// labels.jsonl read/write and URL normalization. +// +// labels.jsonl is the canonical label store: one JSON LabelRow per line, +// append-only. ReadLabels is strict (per-line errors with line numbers); +// AppendLabels is the only writer — rows are never rewritten in place. +package core + +import ( + "bufio" + "encoding/json" + "errors" + "fmt" + "net/url" + "os" + "sort" + "strings" +) + +// NormalizeURL collapses trivially-equivalent article links into one dedup +// key: scheme/host lowercased, leading "www." and trailing "/" dropped. +// Unparseable input degrades to a plain lowercase-trim. +func NormalizeURL(u string) string { + u = strings.TrimSpace(u) + pr, err := url.Parse(u) + if err != nil || pr.Host == "" { + return strings.TrimSuffix(strings.ToLower(u), "/") + } + pr.Scheme = strings.ToLower(pr.Scheme) + pr.Host = strings.TrimPrefix(strings.ToLower(pr.Host), "www.") + return strings.TrimSuffix(pr.String(), "/") +} + +// ReadLabels parses path strictly. A missing file is not an error — it just +// means no labels yet (first export run). +func ReadLabels(path string) ([]LabelRow, error) { + f, err := os.Open(path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + return nil, fmt.Errorf("opening labels: %w", err) + } + defer func() { _ = f.Close() }() + + var rows []LabelRow + scanner := bufio.NewScanner(f) + scanner.Buffer(make([]byte, 64*1024), 4*1024*1024) + line := 0 + for scanner.Scan() { + line++ + text := strings.TrimSpace(scanner.Text()) + if text == "" { + continue + } + var row LabelRow + if err := json.Unmarshal([]byte(text), &row); err != nil { + return nil, fmt.Errorf("%s:%d: invalid JSON: %w", path, line, err) + } + if row.EntryID <= 0 { + return nil, fmt.Errorf("%s:%d: entry_id must be positive", path, line) + } + if row.URL == "" { + return nil, fmt.Errorf("%s:%d: url is required", path, line) + } + if row.Label != LabelPos && row.Label != LabelNeg { + return nil, fmt.Errorf("%s:%d: label must be %q or %q, got %q", path, line, LabelPos, LabelNeg, row.Label) + } + if row.CapturedAt.IsZero() { + return nil, fmt.Errorf("%s:%d: captured_at is required", path, line) + } + rows = append(rows, row) + } + if err := scanner.Err(); err != nil { + return nil, fmt.Errorf("reading %s: %w", path, err) + } + return rows, nil +} + +// SplitByTime partitions rows into train (older) and val (newest valFrac) +// by CapturedAt, the time-split validation strategy: train on older labels, +// evaluate on the most recent window, which is what serve actually faces. +// Sort is stable; ties on CapturedAt break by EntryID for determinism. +// valFrac is clamped: <=0 → all train, >=1 → all val. nVal is computed by +// truncation, so a tiny set may yield an empty val side (callers guard). +func SplitByTime(rows []LabelRow, valFrac float64) (train, val []LabelRow) { + if len(rows) == 0 { + return nil, nil + } + if valFrac <= 0 { + return rows, nil + } + if valFrac >= 1 { + return nil, rows + } + sorted := make([]LabelRow, len(rows)) + copy(sorted, rows) + sort.SliceStable(sorted, func(i, j int) bool { + if sorted[i].CapturedAt.Equal(sorted[j].CapturedAt) { + return sorted[i].EntryID < sorted[j].EntryID + } + return sorted[i].CapturedAt.Before(sorted[j].CapturedAt) + }) + nVal := int(float64(len(sorted)) * valFrac) + split := len(sorted) - nVal + return sorted[:split], sorted[split:] +} + +// AppendLabels appends rows to path (created if needed). Rows already in +// the file are never touched. +func AppendLabels(path string, rows []LabelRow) error { + if len(rows) == 0 { + return nil + } + f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) + if err != nil { + return fmt.Errorf("opening labels for append: %w", err) + } + defer func() { _ = f.Close() }() + + enc := json.NewEncoder(f) + for _, row := range rows { + if err := enc.Encode(row); err != nil { + return fmt.Errorf("writing label row: %w", err) + } + } + return nil +} diff --git a/core/labels_test.go b/core/labels_test.go new file mode 100644 index 0000000..d8abb43 --- /dev/null +++ b/core/labels_test.go @@ -0,0 +1,192 @@ +// Tests for labels.jsonl handling and URL normalization. +package core + +import ( + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestNormalizeURL(t *testing.T) { + cases := []struct{ in, want string }{ + {"HTTPS://WWW.Example.com/Path/", "https://example.com/Path"}, + {"https://example.com/a?x=1&y=2", "https://example.com/a?x=1&y=2"}, + {"https://example.com", "https://example.com"}, + {" https://EXAMPLE.com/ ", "https://example.com"}, + {"https://www.www.example.com/x", "https://www.example.com/x"}, // only one leading www. + {"not a url", "not a url"}, + } + for _, c := range cases { + if got := NormalizeURL(c.in); got != c.want { + t.Errorf("NormalizeURL(%q) = %q, want %q", c.in, got, c.want) + } + } +} + +func sampleRow() LabelRow { + return LabelRow{ + EntryID: 42, URL: "https://example.com/post", Title: "A title", + FeedID: 7, FeedTitle: "Feed", Label: LabelPos, + CapturedAt: time.Date(2024, 3, 1, 12, 0, 0, 0, time.UTC), + } +} + +func TestReadLabelsMissingFile(t *testing.T) { + rows, err := ReadLabels(filepath.Join(t.TempDir(), "nope.jsonl")) + if err != nil { + t.Fatalf("missing file should not error: %v", err) + } + if len(rows) != 0 { + t.Fatalf("expected no rows, got %d", len(rows)) + } +} + +func TestAppendThenReadRoundTrip(t *testing.T) { + path := filepath.Join(t.TempDir(), "labels.jsonl") + r1, r2 := sampleRow(), sampleRow() + r2.EntryID, r2.Label = 43, LabelNeg + + if err := AppendLabels(path, []LabelRow{r1}); err != nil { + t.Fatalf("append 1: %v", err) + } + if err := AppendLabels(path, []LabelRow{r2}); err != nil { + t.Fatalf("append 2: %v", err) + } + + rows, err := ReadLabels(path) + if err != nil { + t.Fatalf("read: %v", err) + } + if len(rows) != 2 || rows[0].EntryID != 42 || rows[1].Label != LabelNeg { + t.Fatalf("round trip mismatch: %+v", rows) + } +} + +func TestAppendLabelsNoopOnEmpty(t *testing.T) { + path := filepath.Join(t.TempDir(), "labels.jsonl") + if err := AppendLabels(path, nil); err != nil { + t.Fatalf("noop append: %v", err) + } + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Fatal("empty append should not create the file") + } +} + +func TestSplitByTime(t *testing.T) { + mk := func(id int64, day int, label string) LabelRow { + return LabelRow{ + EntryID: id, URL: "https://example.com/" + fmt.Sprint(id), + Title: "T", FeedID: 1, FeedTitle: "F", Label: label, + CapturedAt: time.Date(2024, 1, day, 0, 0, 0, 0, time.UTC), + } + } + // 10 rows, days 1..10. val-frac 0.3 → newest 3 (days 8,9,10) are val. + rows := []LabelRow{ + mk(1, 1, LabelPos), mk(2, 2, LabelNeg), mk(3, 3, LabelPos), + mk(4, 4, LabelNeg), mk(5, 5, LabelPos), mk(6, 6, LabelNeg), + mk(7, 7, LabelPos), mk(8, 8, LabelNeg), mk(9, 9, LabelPos), mk(10, 10, LabelNeg), + } + train, val := SplitByTime(rows, 0.3) + if len(val) != 3 { + t.Fatalf("val size = %d, want 3", len(val)) + } + if len(train) != 7 { + t.Fatalf("train size = %d, want 7", len(train)) + } + // Val must be the newest (days 8,9,10). + wantValDays := map[int]bool{8: true, 9: true, 10: true} + for _, r := range val { + d := r.CapturedAt.Day() + if !wantValDays[d] { + t.Errorf("val contains day %d, expected 8/9/10", d) + } + } + // Train must be the older days 1..7. + for _, r := range train { + if r.CapturedAt.Day() > 7 { + t.Errorf("train contains day %d, expected <=7", r.CapturedAt.Day()) + } + } +} + +func TestSplitByTimeEdgeCases(t *testing.T) { + rows := []LabelRow{ + {EntryID: 1, CapturedAt: time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)}, + {EntryID: 2, CapturedAt: time.Date(2024, 1, 2, 0, 0, 0, 0, time.UTC)}, + } + if train, val := SplitByTime(rows, 0); len(train) != 2 || len(val) != 0 { + t.Errorf("valFrac 0: train=%d val=%d, want 2/0", len(train), len(val)) + } + if train, val := SplitByTime(rows, 1); len(train) != 0 || len(val) != 2 { + t.Errorf("valFrac 1: train=%d val=%d, want 0/2", len(train), len(val)) + } + if train, val := SplitByTime(nil, 0.3); len(train) != 0 || len(val) != 0 { + t.Errorf("empty: train=%d val=%d, want 0/0", len(train), len(val)) + } +} + +func TestSplitByTimeTieBreakDeterministic(t *testing.T) { + // Same CapturedAt, different EntryIDs: tie broken by EntryID ascending. + day := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + rows := []LabelRow{ + {EntryID: 30, CapturedAt: day}, {EntryID: 10, CapturedAt: day}, {EntryID: 20, CapturedAt: day}, + } + // Two runs must produce identical splits. + train1, val1 := SplitByTime(rows, 0.34) + _, val2 := SplitByTime(rows, 0.34) + if len(val1) != len(val2) { + t.Fatalf("non-deterministic val size: %d vs %d", len(val1), len(val2)) + } + for i := range val1 { + if val1[i].EntryID != val2[i].EntryID { + t.Fatalf("non-deterministic tie break at %d: %d vs %d", i, val1[i].EntryID, val2[i].EntryID) + } + } + // Val (newest) should carry the highest EntryIDs. + if len(val1) > 0 { + maxTrain := int64(0) + for _, r := range train1 { + if r.EntryID > maxTrain { + maxTrain = r.EntryID + } + } + for _, r := range val1 { + if r.EntryID <= maxTrain { + t.Errorf("val id %d not greater than max train id %d (tie break ascending)", r.EntryID, maxTrain) + } + } + } +} + +func TestReadLabelsStrictErrors(t *testing.T) { + cases := []struct { + name, body, wantSub string + }{ + {"bad json", `{not json`, "invalid JSON"}, + {"bad id", `{"entry_id":0,"url":"https://x","label":"pos","captured_at":"2024-01-01T00:00:00Z"}`, "entry_id"}, + {"no url", `{"entry_id":1,"label":"pos","captured_at":"2024-01-01T00:00:00Z"}`, "url"}, + {"bad label", `{"entry_id":1,"url":"https://x","label":"maybe","captured_at":"2024-01-01T00:00:00Z"}`, "label"}, + {"no captured_at", `{"entry_id":1,"url":"https://x","label":"pos"}`, "captured_at"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "labels.jsonl") + if err := os.WriteFile(path, []byte(c.body+"\n"), 0o644); err != nil { + t.Fatal(err) + } + _, err := ReadLabels(path) + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), c.wantSub) { + t.Errorf("error %q should mention %q", err, c.wantSub) + } + if !strings.Contains(err.Error(), ":1") { + t.Errorf("error %q should carry the line number", err) + } + }) + } +} diff --git a/core/ml.go b/core/ml.go new file mode 100644 index 0000000..8a29f9f --- /dev/null +++ b/core/ml.go @@ -0,0 +1,435 @@ +// Vendored from scholscan/core/ml.go (rev 7c1a5ef, 2026-08-28). +// Owned here; no upstream sync. +// ML implementation: TF-IDF + Logistic Regression for article filtering. +// +// Why title-only: Avoids content scraping overhead, titles are already informative. +// MinDF=2: Removes typos and rare terms that don't generalize. +// MaxDF=0.8: Removes common words that appear in >80% of documents. +// λ=0.001: Light L2 regularization to prevent overfitting on small datasets. +// +// Public API: +// - TFIDFVectorizer.Fit(): Learn vocabulary from documents +// - TFIDFVectorizer.Transform(): Convert documents to TF-IDF vectors +// - LogisticRegression.Fit(): Train classifier on vectors +// - CreateVectorizerFromModel(): Reconstruct vectorizer from saved model +// - PredictScore(): Score article using trained weights +package core + +import ( + "fmt" + "math" + "regexp" + "sort" + "strings" +) + +// ============================================================================ +// ╻ ╻┏━╸┏━╸╺┳╸┏━┓┏━┓╻┏━┓┏━╸┏━┓ +// ┃┏┛┣╸ ┃ ┃ ┃ ┃┣┳┛┃┗━┓┣╸ ┣┳┛ +// ┗┛ ┗━╸┗━╸ ╹ ┗━┛╹┗╸╹┗━┛┗━╸╹┗╸ +// ============================================================================ + +var wordHyphenRegex = regexp.MustCompile("[^a-zA-Z0-9-]+") + +// StopWords: Common words that don't help distinguish articles. +// Why: Reduces noise and improves model generalization. +var stopWords = map[string]struct{}{ + // Single letters and symbols + "s": {}, "-": {}, "0": {}, "1": {}, "2": {}, "3": {}, "4": {}, "5": {}, "6": {}, "7": {}, "8": {}, "9": {}, + + // Common English stop words + "a": {}, "about": {}, "above": {}, "after": {}, "again": {}, "against": {}, "al": {}, "all": {}, "am": {}, "an": {}, "and": {}, "any": {}, "are": {}, "aren't": {}, "as": {}, "at": {}, "be": {}, "because": {}, "been": {}, "before": {}, "being": {}, "below": {}, "between": {}, "both": {}, "but": {}, "by": {}, "can't": {}, "cannot": {}, "could": {}, "couldn't": {}, "did": {}, "didn't": {}, "do": {}, "does": {}, "doesn't": {}, "doing": {}, "don't": {}, "down": {}, "during": {}, "each": {}, "et": {}, "few": {}, "for": {}, "from": {}, "further": {}, "had": {}, "hadn't": {}, "has": {}, "hasn't": {}, "have": {}, "haven't": {}, "having": {}, "he": {}, "he'd": {}, "he'll": {}, "he's": {}, "her": {}, "here": {}, "here's": {}, "hers": {}, "herself": {}, "him": {}, "himself": {}, "his": {}, "how": {}, "how's": {}, "i": {}, "i'd": {}, "i'll": {}, "i'm": {}, "i've": {}, "if": {}, "in": {}, "into": {}, "is": {}, "isn't": {}, "it": {}, "it's": {}, "its": {}, "itself": {}, "let's": {}, "me": {}, "more": {}, "most": {}, "mustn't": {}, "my": {}, "myself": {}, "no": {}, "nor": {}, "not": {}, "of": {}, "off": {}, "on": {}, "once": {}, "only": {}, "or": {}, "other": {}, "ought": {}, "our": {}, "ours": {}, "ourselves": {}, "out": {}, "over": {}, "own": {}, "same": {}, "shan't": {}, "she": {}, "she'd": {}, "she'll": {}, "she's": {}, "should": {}, "shouldn't": {}, "so": {}, "some": {}, "such": {}, "than": {}, "that": {}, "that's": {}, "the": {}, "their": {}, "theirs": {}, "them": {}, "themselves": {}, "then": {}, "there": {}, "there's": {}, "these": {}, "they": {}, "they'd": {}, "they'll": {}, "they're": {}, "they've": {}, "this": {}, "those": {}, "through": {}, "to": {}, "too": {}, "under": {}, "until": {}, "up": {}, "very": {}, "was": {}, "wasn't": {}, "we": {}, "we'd": {}, "we'll": {}, "we're": {}, "we've": {}, "were": {}, "weren't": {}, "what": {}, "what's": {}, "when": {}, "when's": {}, "where": {}, "where's": {}, "which": {}, "while": {}, "who": {}, "who's": {}, "whom": {}, "why": {}, "why's": {}, "with": {}, "won't": {}, "would": {}, "wouldn't": {}, "you": {}, "you'd": {}, "you'll": {}, "you're": {}, "you've": {}, "your": {}, "yours": {}, "yourself": {}, "yourselves": {}, +} + +type TFIDFVectorizer struct { + Vocabulary map[string]float64 + OrderedVocab []string + NgramMin int + NgramMax int + MinDF int // Minimum document frequency (absolute) + MaxDF float64 // Maximum document frequency (ratio) + VocabCap int +} + +func CreateVectorizerFromModel(model *ModelEnvelope) *TFIDFVectorizer { + v := &TFIDFVectorizer{ + Vocabulary: model.Vectorizer, + OrderedVocab: model.OrderedVocab, + } + // Restore the n-gram range recorded at fit time so scoring text is + // vectorized exactly as training/validation saw it (zero values would + // silently degrade bigram models to unigram scoring). Absent meta — + // older or hand-built models — leaves zeros, i.e. unigrams. + if params, ok := model.Meta["vectorizer_params"].(map[string]any); ok { + if n, ok := params["ngram_min"].(float64); ok { + v.NgramMin = int(n) + } + if n, ok := params["ngram_max"].(float64); ok { + v.NgramMax = int(n) + } + } + return v +} + +// Learns vocabulary and IDF from documents +func (v *TFIDFVectorizer) Fit(documents []string) { + numDocs := len(documents) + docFreqs := make(map[string]int) + + for _, doc := range documents { + unigrams := Tokenize(doc) + ngrams := generateNgrams(unigrams, v.NgramMin, v.NgramMax) + seenInDoc := make(map[string]struct{}) + for _, ngram := range ngrams { + if _, seen := seenInDoc[ngram]; !seen { + docFreqs[ngram]++ + seenInDoc[ngram] = struct{}{} + } + } + } + + maxDocs := int(v.MaxDF * float64(numDocs)) + filteredVocab := make(map[string]int) + for term, freq := range docFreqs { + if freq >= v.MinDF && freq <= maxDocs { + filteredVocab[term] = freq + } + } + + if v.VocabCap > 0 && len(filteredVocab) > v.VocabCap { + type termFreq struct { + term string + freq int + } + terms := make([]termFreq, 0, len(filteredVocab)) + for term, freq := range filteredVocab { + terms = append(terms, termFreq{term, freq}) + } + sort.Slice(terms, func(i, j int) bool { + return terms[i].freq > terms[j].freq + }) + + cappedTerms := terms[:v.VocabCap] + filteredVocab = make(map[string]int, v.VocabCap) + for _, tf := range cappedTerms { + filteredVocab[tf.term] = tf.freq + } + } + + v.OrderedVocab = make([]string, 0, len(filteredVocab)) + for term := range filteredVocab { + v.OrderedVocab = append(v.OrderedVocab, term) + } + sort.Strings(v.OrderedVocab) // deterministic order + + v.Vocabulary = make(map[string]float64, len(v.OrderedVocab)) + for _, term := range v.OrderedVocab { + // IDF = log(total num of docs / num of docs with term) + idf := math.Log(float64(numDocs) / float64(filteredVocab[term])) + v.Vocabulary[term] = idf + } +} + +// Converts documents to TF-IDF vectors using learned vocabulary +func (v *TFIDFVectorizer) Transform(documents []string) [][]float64 { + vectors := make([][]float64, len(documents)) + + for i, doc := range documents { + unigrams := Tokenize(doc) + ngrams := generateNgrams(unigrams, v.NgramMin, v.NgramMax) + vector := make([]float64, len(v.OrderedVocab)) + + if len(ngrams) > 0 { + // tf: term frequency (normalized count of each n-gram in document) + tf := make(map[string]float64) + for _, ngram := range ngrams { + tf[ngram]++ + } + numNgrams := float64(len(ngrams)) + for ngram, count := range tf { + tf[ngram] = count / numNgrams + } + + for j, term := range v.OrderedVocab { + if tfValue, ok := tf[term]; ok { + // only score terms that were in our training vocabulary + if idfValue, inVocab := v.Vocabulary[term]; inVocab { + vector[j] = tfValue * idfValue + } + } + } + } + vectors[i] = vector + } + + return vectors +} + +func Tokenize(text string) []string { + text = strings.ToLower(text) + words := wordHyphenRegex.Split(text, -1) + tokens := make([]string, 0, len(words)) + for _, word := range words { + if word == "" { + continue + } + if _, isStopWord := stopWords[word]; isStopWord { + continue + } + tokens = append(tokens, word) + } + return tokens +} + +func generateNgrams(tokens []string, minN, maxN int) []string { + if minN <= 0 { + minN = 1 + } + if maxN < minN { + maxN = minN + } + + numTokens := len(tokens) + + estimatedCap := 0 + for n := minN; n <= maxN; n++ { + if numTokens >= n { + estimatedCap += numTokens - n + 1 + } + } + ngrams := make([]string, 0, estimatedCap) + + for n := minN; n <= maxN; n++ { + if numTokens < n { + continue + } + for i := 0; i <= numTokens-n; i++ { + ngrams = append(ngrams, strings.Join(tokens[i:i+n], " ")) + } + } + return ngrams +} + +// ============================================================================ +// ┏━╸╻ ┏━┓┏━┓┏━┓╻┏━╸╻┏━╸┏━┓ +// ┃ ┃ ┣━┫┗━┓┗━┓┃┣╸ ┃┣╸ ┣┳┛ +// ┗━╸┗━╸╹ ╹┗━┛┗━┛╹╹ ╹┗━╸╹┗╸ +// ============================================================================ + +// Binary logistic regression with L2 regularization +// Bias term stored separately (not regularized) +type LogisticRegression struct { + LearningRate float64 + Lambda float64 // L2 regularization parameter + Iterations int + Tolerance float64 // Convergence tolerance on loss improvement +} + +// validate checks and clamps hyperparams to reasonable bounds. +func (lr *LogisticRegression) Validate() *LogisticRegression { + const ( + defaultLearningRate = 0.5 + defaultIterations = 500 + defaultTolerance = 0.000001 + ) + + if lr.LearningRate <= 0 { + lr.LearningRate = defaultLearningRate + } + if lr.LearningRate > 10 { + lr.LearningRate = 10.0 + } + if lr.Lambda < 0 { + lr.Lambda = 0.0 + } + if lr.Iterations <= 0 { + lr.Iterations = defaultIterations + } + if lr.Tolerance <= 0 { + lr.Tolerance = defaultTolerance + } + return lr +} + +// Fit trains via SGD with L2 regularization on feature weights (not bias). +// Class weights reweight samples; unused in our pipeline (we downsample instead). +// Returns weights with bias as last element. +func (lr *LogisticRegression) Fit(vectors [][]float64, labels []float64, classWeights map[float64]float64) ([]float64, error) { + if len(vectors) == 0 { + return nil, fmt.Errorf("cannot train on empty dataset") + } + if len(vectors) != len(labels) { + return nil, fmt.Errorf( + "mismatch between number of vectors (%d) and labels (%d)", + len(vectors), + len(labels), + ) + } + + for i, y := range labels { + if y != 0 && y != 1 { + return nil, fmt.Errorf("invalid label at %d: %v (expected 0 or 1)", i, y) + } + } + + numFeatures := len(vectors[0]) + if numFeatures == 0 { + return nil, fmt.Errorf("cannot train with zero-length feature vectors") + } + for i := 1; i < len(vectors); i++ { + if len(vectors[i]) != numFeatures { + return nil, fmt.Errorf( + "inconsistent feature vector length at index %d: got %d, expected %d", + i, + len(vectors[i]), + numFeatures, + ) + } + } + useUniformWeights := classWeights == nil + if useUniformWeights { + classWeights = map[float64]float64{0.0: 1.0, 1.0: 1.0} + } + + numSamples := float64(len(vectors)) + var totalWeight float64 + if useUniformWeights { + totalWeight = numSamples + } else { + for _, y := range labels { + totalWeight += classWeights[y] + } + } + if totalWeight == 0 { + totalWeight = numSamples // Fallback + } + + weights := make([]float64, numFeatures) + var bias float64 + + prevLoss := math.MaxFloat64 + + for i := 0; i < lr.Iterations; i++ { + gradWeights := make([]float64, numFeatures) + var gradBias float64 + var currentLoss float64 + + for j, x := range vectors { + y := labels[j] + sampleWeight := classWeights[y] + + z, err := dot(weights, x) + if err != nil { + return nil, fmt.Errorf("error calculating dot product for vector %d: %w", j, err) + } + p := Sigmoid(z + bias) + + // Compute prediction error. This term gets multiplied by each feature value + // to accumulate gradients (higher error pushes weights harder). + errTerm := p - y + for k := 0; k < numFeatures; k++ { + gradWeights[k] += sampleWeight * errTerm * x[k] + } + gradBias += sampleWeight * errTerm + + cp := clamp(p) + currentLoss += sampleWeight * (-(y*math.Log(cp) + (1-y)*math.Log(1-cp))) + } + + // Update weights with L2 regularization (only on feature weights, not bias). + // This pulls weights toward zero, preventing overfitting on small datasets. + for k := 0; k < numFeatures; k++ { + regularizedGrad := (gradWeights[k] / totalWeight) + (lr.Lambda * weights[k]) + weights[k] -= lr.LearningRate * regularizedGrad + } + gradBias /= totalWeight + bias -= lr.LearningRate * gradBias + + // Check convergence: if loss change is below tolerance, we're done. + // We include the L2 penalty in total loss to assess true convergence. + avgLoss := currentLoss / totalWeight + var l2Penalty float64 + for _, w := range weights { + l2Penalty += w * w + } + totalLoss := avgLoss + 0.5*lr.Lambda*l2Penalty + if math.Abs(prevLoss-totalLoss) < lr.Tolerance { + break + } + prevLoss = totalLoss + } + + // bias is stored as the last element + return append(weights, bias), nil +} + +// PredictScore computes the probability for a single vec given weights. +// the last element of weights is the bias. +func PredictScore(vector []float64, weights []float64) (float64, error) { + if len(weights) == 0 { + return 0, fmt.Errorf("weights cannot be empty") + } + if len(vector) != len(weights)-1 { + return 0, fmt.Errorf( + "vector length mismatch: expected %d features, got %d", + len(weights)-1, + len(vector), + ) + } + + for i, v := range vector { + if math.IsNaN(v) || math.IsInf(v, 0) { + return 0, fmt.Errorf("invalid value at vector[%d]: %v", i, v) + } + } + for i, w := range weights { + if math.IsNaN(w) || math.IsInf(w, 0) { + return 0, fmt.Errorf("invalid value at weights[%d]: %v", i, w) + } + } + + featureWeights := weights[:len(weights)-1] + bias := weights[len(weights)-1] + + z, err := dot(featureWeights, vector) + if err != nil { + return 0, fmt.Errorf("failed to compute dot product: %w", err) + } + return Sigmoid(z + bias), nil +} + +// ============================================================================ +// ┏┳┓┏━┓╺┳╸╻ ╻┏━┓ +// ┃┃┃┣━┫ ┃ ┣━┫┗━┓ +// ╹ ╹╹ ╹ ╹ ╹ ╹┗━┛ +// ============================================================================ + +func Sigmoid(z float64) float64 { + if z >= 0 { + return 1.0 / (1.0 + math.Exp(-z)) + } + ez := math.Exp(z) + return ez / (1.0 + ez) +} + +func dot(a, b []float64) (float64, error) { + if len(a) != len(b) { + return 0, fmt.Errorf("vector length mismatch: %d != %d", len(a), len(b)) + } + var sum float64 + for i := range a { + sum += a[i] * b[i] + } + return sum, nil +} + +func clamp(p float64) float64 { + const probabilityClamp = 1e-15 + if p < probabilityClamp { + return probabilityClamp + } + if p > 1.0-probabilityClamp { + return 1.0 - probabilityClamp + } + return p +} diff --git a/core/ml_test.go b/core/ml_test.go new file mode 100644 index 0000000..3623b39 --- /dev/null +++ b/core/ml_test.go @@ -0,0 +1,76 @@ +// Roundtrip test for the vendored TF-IDF + LR pipeline: fit a tiny model, +// confirm scores separate the two synthetic topics. Guards against a broken +// copy more than against algorithm regressions. +package core + +import ( + "testing" +) + +func TestTFIDFLRRoundtrip(t *testing.T) { + // Two separable synthetic "topics": optics vs cooking. + docs := []string{ + "quantum laser optics photon", + "laser photon quantum measurement", + "photon optics quantum laser", + "optics quantum laser photon", + "recipe cooking knife onion", + "onion recipe cooking garlic", + "cooking garlic onion skillet", + "garlic recipe onion cooking", + } + labels := []float64{1, 1, 1, 1, 0, 0, 0, 0} + + v := &TFIDFVectorizer{NgramMin: 1, NgramMax: 2, MinDF: 1, MaxDF: 0.9} + v.Fit(docs) + if len(v.OrderedVocab) == 0 { + t.Fatal("vectorizer learned empty vocabulary") + } + + vecs := v.Transform(docs) + lr := (&LogisticRegression{Lambda: 0.001}).Validate() + weights, err := lr.Fit(vecs, labels, nil) + if err != nil { + t.Fatalf("Fit: %v", err) + } + + // Persist and reload through the envelope, as serve does. + model := &ModelEnvelope{ + Vectorizer: v.Vocabulary, + OrderedVocab: v.OrderedVocab, + Weights: weights, + } + rv := CreateVectorizerFromModel(model) + + opticsScore := mustScore(t, rv, model, "quantum photon laser experiment") + cookingScore := mustScore(t, rv, model, "onion garlic skillet dinner") + if !(opticsScore > 0.5 && cookingScore < 0.5) { + t.Fatalf("expected separation, got optics=%.3f cooking=%.3f", opticsScore, cookingScore) + } + if opticsScore <= cookingScore { + t.Fatalf("expected optics score above cooking score, got %.3f <= %.3f", opticsScore, cookingScore) + } +} + +func mustScore(t *testing.T, v *TFIDFVectorizer, m *ModelEnvelope, title string) float64 { + t.Helper() + vec := v.Transform([]string{title}) + s, err := PredictScore(vec[0], m.Weights) + if err != nil { + t.Fatalf("PredictScore(%q): %v", title, err) + } + return s +} + +func TestTokenize(t *testing.T) { + got := Tokenize("The Fall of the House of Usher!") + want := []string{"fall", "house", "usher"} + if len(got) != len(want) { + t.Fatalf("Tokenize: got %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("Tokenize: got %v, want %v", got, want) + } + } +} diff --git a/core/model.go b/core/model.go new file mode 100644 index 0000000..4afcdae --- /dev/null +++ b/core/model.go @@ -0,0 +1,22 @@ +// Vendored from scholscan/core/model.go (rev 7c1a5ef, 2026-08-28). +// Owned here; no upstream sync. +// Model envelope persists trained model to JSON. Contains Vectorizer for IDF values, +// OrderedVocab for feature ordering, and Weights for logistic regression. +// To score: recreate TFIDFVectorizer, transform, then PredictScore. +package core + +import ( + "time" +) + +// ModelEnvelope - complete trained model for scoring articles +type ModelEnvelope struct { + Algorithm string `json:"algorithm"` + Impl string `json:"impl"` + Version string `json:"version"` + CreatedAt time.Time `json:"created_at"` + Meta map[string]any `json:"meta"` + Vectorizer map[string]float64 `json:"vectorizer"` + OrderedVocab []string `json:"ordered_vocab"` + Weights []float64 `json:"weights"` +} diff --git a/core/runs.go b/core/runs.go new file mode 100644 index 0000000..38e6710 --- /dev/null +++ b/core/runs.go @@ -0,0 +1,167 @@ +// runs.jsonl read/write: the serve-side run log. One JSON RunRow appended +// per non-empty poll batch; kept items double as the RSS render source, so +// each carries everything a feed item needs (title/link/date/cleaned +// content). Rotation is generous: past maxRunsBytes the file is rewritten +// keeping the newest rows, via temp + rename so a crash mid-rotate can't +// leave a torn log behind. +package core + +import ( + "bufio" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "time" +) + +// Kept-item kinds. Short strings so runs.jsonl stays human-greppable. +const ( + KindRanked = "ranked" + KindRunnerUp = "runnerup" + KindRandom = "random" +) + +// Rotation knobs. 20MB of runs is years of batches at expected sizes, and +// 500 retained rows is still far more than the count-based RSS tail (~21) +// will ever render. +const ( + maxRunsBytes = 20 << 20 + keepRunsOnRotate = 500 +) + +// KeptItem is one entry selected into a digest batch, with everything the +// RSS renderer needs so rendering never touches Miniflux. +type KeptItem struct { + EntryID int64 `json:"entry_id"` + URL string `json:"url"` + Title string `json:"title"` + FeedTitle string `json:"feed_title"` + PublishedAt time.Time `json:"published_at"` + Content string `json:"content"` // cleaned at poll time: HTML-stripped, <=5KB + Score float64 `json:"score"` + Kind string `json:"kind"` // KindRanked | KindRunnerUp | KindRandom +} + +// RunRow is one poll batch: what was seen, what was kept, when, and how +// old the model was. EntriesSeen counts every fetched entry including any +// skipped before scoring (see RecFeedSkipped). +type RunRow struct { + RanAt time.Time `json:"ran_at"` + EntriesSeen int `json:"entries_seen"` + RecFeedSkipped int `json:"rec_feed_skipped,omitempty"` + Kept []KeptItem `json:"kept"` + DurationMs int64 `json:"duration_ms"` + ModelCreatedAt time.Time `json:"model_created_at"` +} + +// ReadRuns parses all of path strictly. A missing file is not an error — +// it just means no runs logged yet (first serve boot). +func ReadRuns(path string) ([]RunRow, error) { + f, err := os.Open(path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + return nil, fmt.Errorf("opening runs: %w", err) + } + defer func() { _ = f.Close() }() + + var rows []RunRow + scanner := bufio.NewScanner(f) + scanner.Buffer(make([]byte, 64*1024), 4*1024*1024) + line := 0 + for scanner.Scan() { + line++ + text := strings.TrimSpace(scanner.Text()) + if text == "" { + continue + } + var row RunRow + if err := json.Unmarshal([]byte(text), &row); err != nil { + return nil, fmt.Errorf("%s:%d: invalid JSON: %w", path, line, err) + } + if row.RanAt.IsZero() { + return nil, fmt.Errorf("%s:%d: ran_at is required", path, line) + } + for i, k := range row.Kept { + if k.EntryID <= 0 { + return nil, fmt.Errorf("%s:%d: kept[%d]: entry_id must be positive", path, line, i) + } + if k.Kind != KindRanked && k.Kind != KindRunnerUp && k.Kind != KindRandom { + return nil, fmt.Errorf("%s:%d: kept[%d]: invalid kind %q", path, line, i, k.Kind) + } + } + rows = append(rows, row) + } + if err := scanner.Err(); err != nil { + return nil, fmt.Errorf("reading %s: %w", path, err) + } + return rows, nil +} + +// ReadRunsTail returns the newest k runs in file order (oldest to newest). +// k <= 0 reads everything. +func ReadRunsTail(path string, k int) ([]RunRow, error) { + rows, err := ReadRuns(path) + if err != nil { + return nil, err + } + if k > 0 && len(rows) > k { + rows = rows[len(rows)-k:] + } + return rows, nil +} + +// AppendRun appends one row, then rotates the file if it has grown past +// maxRunsBytes. Appending is the only mutation; existing rows are never +// edited, only dropped by rotation. +func AppendRun(path string, row RunRow) error { + f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) + if err != nil { + return fmt.Errorf("opening runs for append: %w", err) + } + if err := json.NewEncoder(f).Encode(row); err != nil { + _ = f.Close() + return fmt.Errorf("writing run row: %w", err) + } + if err := f.Close(); err != nil { + return fmt.Errorf("closing runs: %w", err) + } + + info, err := os.Stat(path) + if err != nil { + return fmt.Errorf("stat runs: %w", err) + } + if info.Size() <= maxRunsBytes { + return nil + } + rows, err := ReadRuns(path) + if err != nil { + return err + } + if len(rows) > keepRunsOnRotate { + rows = rows[len(rows)-keepRunsOnRotate:] + } + tmp := filepath.Clean(path) + ".tmp" + w, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644) + if err != nil { + return fmt.Errorf("rotating runs: %w", err) + } + enc := json.NewEncoder(w) + for _, r := range rows { + if err := enc.Encode(r); err != nil { + _ = w.Close() + return fmt.Errorf("rotating runs: %w", err) + } + } + if err := w.Close(); err != nil { + return fmt.Errorf("rotating runs: %w", err) + } + if err := os.Rename(tmp, path); err != nil { + return fmt.Errorf("rotating runs: %w", err) + } + return nil +} diff --git a/core/scoring.go b/core/scoring.go new file mode 100644 index 0000000..66b3a3b --- /dev/null +++ b/core/scoring.go @@ -0,0 +1,16 @@ +// Vendored from scholscan/core/scoring.go (rev 7c1a5ef, 2026-08-28). +// Owned here; no upstream sync. +// Score conversion utilities. +// +// ScoreToScale: Maps probability (0-1) to user-friendly 1-10 scale. +// Why: Users understand "8/10" better than "0.82 probability". +package core + +import "math" + +// ScoreToScale turns probability into 1-10 display score +func ScoreToScale(rawScore, threshold float64) int { + k := 10.0 + adjustedScore := 1.0 / (1.0 + math.Exp(-k*(rawScore-threshold))) + return int(math.Round(1.0 + (adjustedScore * 9.0))) +} diff --git a/core/scoring_text.go b/core/scoring_text.go new file mode 100644 index 0000000..84cd758 --- /dev/null +++ b/core/scoring_text.go @@ -0,0 +1,43 @@ +// Scoring-text construction: the document fed to the vectorizer. +// +// fluxrec is title-only by design (no body/excerpt ever enters the scoring +// text). On top of the raw title we inject one synthetic feed-identity token +// `feed-<slug>` built from the entry's feed_title, so LR can learn a per-feed +// prior inside the existing TF-IDF + LR model. MinDF shrinks low-data feeds +// toward zero for free (a feed with <2 labeled entries never enters the +// vocabulary), so no separate shrinkage-prior machinery. +// +// Hyphens, not underscores: the tokenizer keeps [a-zA-Z0-9-] and splits on +// everything else, so `feed-marginal-revolution` survives as ONE token while +// `feed_marginal_revolution` would fragment into three. Identical slug +// construction at train (from labels.jsonl feed_title) and serve (from the +// entry's feed.title) keeps the feature consistent across both sides. +package core + +import ( + "regexp" + "strings" +) + +var slugSep = regexp.MustCompile(`[^a-z0-9]+`) + +// FeedSlug normalizes a feed title into a stable slug of lowercase +// letters/digits joined by single hyphens, trimmed. Empty/whitespace input +// returns "". Idempotent: slug(slug(x)) == slug(x). +func FeedSlug(feedTitle string) string { + s := strings.ToLower(strings.TrimSpace(feedTitle)) + s = slugSep.ReplaceAllString(s, "-") + return strings.Trim(s, "-") +} + +// ScoringText builds the document fed to the vectorizer: the entry title plus +// one synthetic token "feed-<slug>". If the feed title is empty (no slug), the +// token is omitted and the title alone is returned. No body/excerpt is ever +// added — title-only is the design. +func ScoringText(title, feedTitle string) string { + slug := FeedSlug(feedTitle) + if slug == "" { + return title + } + return title + " feed-" + slug +} diff --git a/core/scoring_text_test.go b/core/scoring_text_test.go new file mode 100644 index 0000000..87b3948 --- /dev/null +++ b/core/scoring_text_test.go @@ -0,0 +1,99 @@ +// Tests for feed-identity token and scoring-text construction. +package core + +import ( + "strings" + "testing" +) + +func TestFeedSlug(t *testing.T) { + cases := []struct{ in, want string }{ + {"Marginal Revolution", "marginal-revolution"}, + {"ACM / IEEE", "acm-ieee"}, + {" Already-Hyphenated ", "already-hyphenated"}, + {"---", ""}, + {"", ""}, + {"MIT Tech Review #2", "mit-tech-review-2"}, + {"Über Wörter", "ber-w-rter"}, // non-ascii stripped to hyphens + } + for _, c := range cases { + if got := FeedSlug(c.in); got != c.want { + t.Errorf("FeedSlug(%q) = %q, want %q", c.in, got, c.want) + } + } +} + +func TestFeedSlugIdempotent(t *testing.T) { + for _, in := range []string{"Marginal Revolution", "ACM / IEEE", "X---Y", "a b c"} { + once := FeedSlug(in) + twice := FeedSlug(once) + if once != twice { + t.Errorf("slug not idempotent: %q → %q → %q", in, once, twice) + } + } +} + +func TestScoringTextIncludesToken(t *testing.T) { + got := ScoringText("A Title", "Marginal Revolution") + if !strings.HasSuffix(got, " feed-marginal-revolution") { + t.Errorf("expected feed-identity token suffix, got %q", got) + } +} + +func TestScoringTextEmptyFeedOmitsToken(t *testing.T) { + if got := ScoringText("A Title", ""); got != "A Title" { + t.Errorf("empty feed should yield title only, got %q", got) + } + if got := ScoringText("A Title", " --- "); got != "A Title" { + t.Errorf("whitespace-only feed should yield title only, got %q", got) + } +} + +// The feed-identity token must round-trip through Tokenize as a SINGLE token +// (hyphens survive; underscores would fragment). Same feed → shared token; +// different feeds → different tokens. +func TestFeedTokenRoundTripsAsSingleToken(t *testing.T) { + doc := ScoringText("some title here", "Marginal Revolution") + toks := Tokenize(doc) + found := false + for _, tk := range toks { + if tk == "feed-marginal-revolution" { + found = true + break + } + } + if !found { + t.Fatalf("feed token not present as a single token in %v", toks) + } + // And it must not have fragmented into feed/marginal/revolution. + for _, bad := range []string{"feed", "marginal", "revolution"} { + for _, tk := range toks { + if tk == bad { + t.Errorf("token fragmented: %q appeared standalone", bad) + } + } + } + + // Same feed → shared token; different feed → different token. + a := ScoringText("x", "Marginal Revolution") + b := ScoringText("y", "Marginal Revolution") + c := ScoringText("z", "Other Blog") + tokA, tokC := feedToken(t, a), feedToken(t, c) + if feedToken(t, b) != tokA { + t.Error("same feed should share the token") + } + if tokA == tokC { + t.Error("different feeds should differ") + } +} + +func feedToken(t *testing.T, doc string) string { + t.Helper() + for _, tk := range Tokenize(doc) { + if strings.HasPrefix(tk, "feed-") { + return tk + } + } + t.Fatalf("no feed token in %q", doc) + return "" +} diff --git a/core/text.go b/core/text.go new file mode 100644 index 0000000..563abb3 --- /dev/null +++ b/core/text.go @@ -0,0 +1,38 @@ +// Vendored from scholscan/core/text.go (rev 7c1a5ef, 2026-08-28). +// Owned here; no upstream sync. +// Text processing for RSS feed content. +// Used for web UI previews and search indexing - not ML (title-only scoring). +package core + +import ( + "regexp" + "strings" +) + +// CleanFeedContent strips HTML, normalizes whitespace, truncates to 5KB +func CleanFeedContent(content string) string { + if content == "" { + return "" + } + + content = StripHTMLTags(content) + content = NormalizeSpace(content) + + maxLength := 5000 + if len(content) > maxLength { + content = content[:maxLength] + "..." + } + + return content +} + +// StripHTMLTags removes HTML tags +func StripHTMLTags(content string) string { + re := regexp.MustCompile(`<[^>]*>`) + return re.ReplaceAllString(content, "") +} + +// NormalizeSpace collapses whitespace and trims +func NormalizeSpace(s string) string { + return strings.Join(strings.Fields(strings.TrimSpace(s)), " ") +} diff --git a/core/types.go b/core/types.go new file mode 100644 index 0000000..073ee8f --- /dev/null +++ b/core/types.go @@ -0,0 +1,59 @@ +// Core type definitions for fluxrec. +// +// Based on scholscan/core/types.go (vendored rev 7c1a5ef) but trimmed hard: +// academic metadata (authors, journal, DOI, enrichment config) is out. +// fluxrec's world is Miniflux entries, labels, and scores. +package core + +import ( + "io" + "time" +) + +// Article is a title-carrying item moving through the pipeline +// (Miniflux entry at serve time, scored candidate in the digest). +type Article struct { + Title string `json:"title"` + Content string `json:"content,omitempty"` + URL string `json:"url"` + Source string `json:"source,omitempty"` // feed title + PublishedAt *time.Time `json:"published_at,omitempty"` + Score *float64 `json:"score,omitempty"` +} + +// Label values for LabelRow.Label. Short strings so labels.jsonl stays +// human-greppable. +const ( + LabelPos = "pos" + LabelNeg = "neg" +) + +// LabelRow is one row of labels.jsonl, the canonical label store. +// Written by `export`, read by `train`. +// +// EntryID is Miniflux's entry ID; for stars made on rec-feed items it resolves +// to the *original* entry via the guid contract, so a row always describes the +// original article. +type LabelRow struct { + EntryID int64 `json:"entry_id"` + URL string `json:"url"` + Title string `json:"title"` + FeedID int64 `json:"feed_id"` + FeedTitle string `json:"feed_title"` + Label string `json:"label"` // LabelPos or LabelNeg + CapturedAt time.Time `json:"captured_at"` +} + +// Command defines the interface that all CLI subcommands must implement. +type Command interface { + // Name returns the command name (e.g., "export", "train"). + Name() string + + // Init parses command-line arguments and initializes the command. + // It should return flag.ErrHelp if --help was requested. + Init(args []string) error + + // Run executes the command, reading from stdin and writing to stdout. + // The command should handle its own error reporting. + Run(stdin io.Reader, stdout io.Writer) error +} @@ -0,0 +1,3 @@ +module fluxrec + +go 1.25.1 diff --git a/justfile b/justfile new file mode 100644 index 0000000..6921ac4 --- /dev/null +++ b/justfile @@ -0,0 +1,68 @@ +# fluxrec task automation (planned commands per PLAN.md, only implemented ones run) + +build: + go build -o fluxrec . + +test: + go test ./... + +fmt: + go fmt ./... + +vet: + go vet ./... + +clean: + rm -f fluxrec + +# Quick smoke: train-free, needs a model.json (from `fluxrec train`, future) +score TITLE MODEL="model.json": + echo "{{TITLE}}" | go run . score --model {{MODEL}} + +# ── auth (same pattern as the mflux justfile) ── + +# Print the two export lines to paste in your shell (just can't set them for you) +auth: + @echo 'export MFLUX_URL="https://your-miniflux-host" # base URL, no trailing slash' + @echo 'read -s MFLUX_TOKEN && export MFLUX_TOKEN # Settings -> API Keys; input hidden' + +# Guard for recipes that hit the live API (token never echoed) +check-auth: + @test -n "${MINIFLUX_URL:-${MFLUX_URL:-}}" -a -n "${MINIFLUX_TOKEN:-${MFLUX_TOKEN:-}}" || { echo "not authed; run: just auth"; exit 1; } + @echo "auth ok (URL=${MINIFLUX_URL:-${MFLUX_URL:-}})" + +# Capture dev fixtures from the live API; run once, then offline dev +dump-fixtures DIR="testdata": check-auth + go run . export --dump-raw {{DIR}} + +# Retroactive-starring aid: model-ranked read+unstarred candidates to star by hand +retrostar LIMIT="50": + go run . retrostar --limit {{LIMIT}} + +# ── retrain loop ── +# Export new labels, train, print the report path. Eyeball report.json +# (precision@15 >= 0.10) before `just ship SERVER` — retrain never deploys. + +retrain: check-auth + go run . export + go run . train labels.jsonl + @echo "report: report.json — check precision@15 >= 0.10 before shipping" + +# ── deploy lane (local PC → server) ── +# SERVER is an ssh config alias; the agent does auth. No hostnames in this +# file. Run `ship` only after eyeballing report.json — retrain never deploys. + +# Ship a freshly trained model.json to the server +deploy SERVER: + scp model.json {{SERVER}}:/srv/fluxrec/ + +# Restart mechanism pinned at D5 once the container runtime is known; +# adjust the ssh line below then. + +# Bounce the serve container so it picks up the new model +restart SERVER: + ssh {{SERVER}} 'docker restart fluxrec' + +# deploy + restart in one verb +ship SERVER: (deploy SERVER) (restart SERVER) + @echo "model shipped; serve restarted — watch /api/status next run"
\ No newline at end of file @@ -0,0 +1,76 @@ +// fluxrec command-line tool. +// Personal Miniflux article recommender; see PLAN.md for the design. +package main + +import ( + "errors" + "flag" + "fmt" + "os" + + "fluxrec/cmds" + "fluxrec/core" +) + +func main() { + if len(os.Args) < 2 { + printHelp() + os.Exit(1) + } + + cmdName := os.Args[1] + args := os.Args[2:] + + if cmdName == "help" || cmdName == "--help" || cmdName == "-h" { + printHelp() + return + } + + var cmd core.Command + switch cmdName { + case "score": + cmd = &cmds.ScoreCommand{} + case "export": + cmd = &cmds.ExportCommand{} + case "train": + cmd = &cmds.TrainCommand{} + case "serve": + cmd = &cmds.ServeCommand{} + case "retrostar": + cmd = &cmds.RetrostarCommand{} + default: + fmt.Fprintf(os.Stderr, "Unknown command: %s\n\n", cmdName) + printHelp() + os.Exit(1) + } + + if err := cmd.Init(args); err != nil { + if errors.Is(err, flag.ErrHelp) { + os.Exit(0) + } + fmt.Fprintf(os.Stderr, "Error initializing %s command: %v\n", cmdName, err) + os.Exit(1) + } + + if err := cmd.Run(os.Stdin, os.Stdout); err != nil { + fmt.Fprintf(os.Stderr, "Error running %s command: %v\n", cmdName, err) + os.Exit(1) + } +} + +func printHelp() { + fmt.Printf(`fluxrec <command> [arguments] + +Personal Miniflux article recommender (see PLAN.md). + +Commands: + score Score stdin titles with a trained model (debugging) + export Pull starred + sampled negatives from Miniflux into labels.jsonl + train Train a TF-IDF + LR model from labels.jsonl + serve Poll Miniflux, score, serve recommendations RSS (server) + retrostar Model-ranked candidates for retroactive starring (cold start) + +Usage: + echo "some title" | fluxrec score --model model.json +`) +} diff --git a/miniflux/client.go b/miniflux/client.go new file mode 100644 index 0000000..ae23fe3 --- /dev/null +++ b/miniflux/client.go @@ -0,0 +1,201 @@ +// Miniflux API client for fluxrec. +// +// Typed over exactly what fluxrec needs: GET /v1/entries with the filter +// subset used by export (starred/status/changed_after) and serve +// (after_entry_id). Auth is the X-Auth-Token header; credentials come from +// the environment (MINIFLUX_* with MFLUX_* fallback, so existing mflux +// exports work). Read-only by construction: there are no mutation methods. +package miniflux + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "os" + "strconv" + "time" +) + +// Env var names, primary then fallback (matches mflux's exports). +var ( + urlVars = [2]string{"MINIFLUX_URL", "MFLUX_URL"} + tokenVars = [2]string{"MINIFLUX_TOKEN", "MFLUX_TOKEN"} +) + +// Config is resolved client configuration. +type Config struct { + BaseURL string + Token string +} + +// ConfigFromEnv resolves MINIFLUX_URL/MINIFLUX_TOKEN, falling back to +// MFLUX_URL/MFLUX_TOKEN. Errors name both variables so a missing export is +// obvious from the message alone. +func ConfigFromEnv() (Config, error) { + baseURL := firstEnv(urlVars) + if baseURL == "" { + return Config{}, fmt.Errorf("miniflux URL not set: export %s (or %s)", urlVars[0], urlVars[1]) + } + token := firstEnv(tokenVars) + if token == "" { + return Config{}, fmt.Errorf("miniflux token not set: export %s (or %s)", tokenVars[0], tokenVars[1]) + } + return Config{BaseURL: baseURL, Token: token}, nil +} + +func firstEnv(names [2]string) string { + for _, n := range names { + if v := os.Getenv(n); v != "" { + return v + } + } + return "" +} + +// Client is a read-only Miniflux API client. +type Client struct { + baseURL string + token string + httpClient *http.Client +} + +// NewClient builds a Client with a sane default HTTP timeout. +func NewClient(cfg Config) *Client { + return &Client{ + baseURL: cfg.BaseURL, + token: cfg.Token, + httpClient: &http.Client{Timeout: 60 * time.Second}, + } +} + +// NewClientWithHTTP is NewClient with a caller-provided http.Client +// (tests point it at httptest servers). +func NewClientWithHTTP(cfg Config, hc *http.Client) *Client { + return &Client{baseURL: cfg.BaseURL, token: cfg.Token, httpClient: hc} +} + +// Feed is the subset of a Miniflux feed fluxrec needs. +type Feed struct { + ID int64 `json:"id"` + Title string `json:"title"` +} + +// Entry is the subset of a Miniflux entry fluxrec needs. JSON field names +// match the Miniflux API verbatim so decoding is a straight mirror. +type Entry struct { + ID int64 `json:"id"` + URL string `json:"url"` + Title string `json:"title"` + Content string `json:"content"` + Status string `json:"status"` // "read" | "unread" | "removed" + Starred bool `json:"starred"` + PublishedAt time.Time `json:"published_at"` + ChangedAt time.Time `json:"changed_at"` + Feed Feed `json:"feed"` +} + +// Filters selects which entries GET /v1/entries returns. Zero values mean +// "no filter" for every field; Starred is a *bool so unset vs. false stays +// distinguishable (we only ever filter starred=true, but keep the type +// honest). +type Filters struct { + Starred *bool // only starred entries when true + Status string // "read" | "unread"; "" = any + ChangedAfter int64 // unix timestamp; incremental export cursor + PublishedAfter int64 // unix timestamp; only entries published after this + AfterEntryID int64 // serve-lane poll cursor + Direction string // "asc" | "desc"; "" = server default + Limit int // page size; 0 = server default + Offset int // paging offset +} + +// EntrySet is one page of GET /v1/entries results. Total is the full +// matching count across pages, which offsets paging loops can stop against. +type EntrySet struct { + Total int `json:"total"` + Entries []Entry `json:"entries"` +} + +// Entries calls GET /v1/entries with the given filters. +func (c *Client) Entries(f Filters) (*EntrySet, error) { + body, err := c.fetchEntries(f) + if err != nil { + return nil, err + } + var set EntrySet + if err := json.Unmarshal(body, &set); err != nil { + return nil, fmt.Errorf("decoding /v1/entries response: %w", err) + } + return &set, nil +} + +// RawEntries returns the /v1/entries response body verbatim. Exists for +// fixture capture (`export --dump-raw`); the typed path stays Entries. +func (c *Client) RawEntries(f Filters) ([]byte, error) { + return c.fetchEntries(f) +} + +// fetchEntries issues the GET and returns the body on HTTP 200. +func (c *Client) fetchEntries(f Filters) ([]byte, error) { + u := entriesURL(c.baseURL, f) + + req, err := http.NewRequest(http.MethodGet, u, nil) + if err != nil { + return nil, fmt.Errorf("building request: %w", err) + } + req.Header.Set("X-Auth-Token", c.token) + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("GET %s: %w", u, err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 256)) + return nil, fmt.Errorf("GET %s: status %d: %s", u, resp.StatusCode, body) + } + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("reading response from %s: %w", u, err) + } + return body, nil +} + +// entriesURL builds /v1/entries?... honoring only the non-zero filters. +func entriesURL(base string, f Filters) string { + q := url.Values{} + if f.Starred != nil { + q.Set("starred", strconv.FormatBool(*f.Starred)) + } + if f.Status != "" { + q.Set("status", f.Status) + } + if f.ChangedAfter > 0 { + q.Set("changed_after", strconv.FormatInt(f.ChangedAfter, 10)) + } + if f.PublishedAfter > 0 { + q.Set("published_after", strconv.FormatInt(f.PublishedAfter, 10)) + } + if f.AfterEntryID > 0 { + q.Set("after_entry_id", strconv.FormatInt(f.AfterEntryID, 10)) + } + if f.Direction != "" { + q.Set("direction", f.Direction) + } + if f.Limit > 0 { + q.Set("limit", strconv.Itoa(f.Limit)) + } + if f.Offset > 0 { + q.Set("offset", strconv.Itoa(f.Offset)) + } + + u := base + "/v1/entries" + if len(q) > 0 { + u += "?" + q.Encode() + } + return u +} diff --git a/miniflux/client_test.go b/miniflux/client_test.go new file mode 100644 index 0000000..8f8c6ad --- /dev/null +++ b/miniflux/client_test.go @@ -0,0 +1,307 @@ +// Tests for the Miniflux client against an httptest fake. Inline fixture +// JSON mirrors a real GET /v1/entries response shape (extra API fields +// included to prove they're harmlessly ignored). +package miniflux + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +const entriesFixture = `{ + "total": 2, + "entries": [ + { + "id": 41001, + "user_id": 3, + "feed_id": 77, + "status": "read", + "hash": "abc123", + "title": "Why headline-only models are fine actually", + "url": "https://example.com/posts/headline-models", + "comments_url": "", + "published_at": "2024-03-02T10:15:00Z", + "created_at": "2024-03-02T11:00:00Z", + "changed_at": "2024-03-05T08:30:00Z", + "content": "<p>Article body here.</p>", + "author": "Jane Doe", + "share_code": "", + "starred": true, + "reading_time": 4, + "enclosures": null, + "feed": { + "id": 77, + "user_id": 3, + "feed_url": "https://example.com/feed.xml", + "site_url": "https://example.com", + "title": "Example Blog", + "checked_at": "2024-03-05T09:00:00Z", + "etag_header": "", + "last_modified_header": "", + "parsing_error_message": "", + "parsing_error_count": 0 + } + }, + { + "id": 41002, + "user_id": 3, + "feed_id": 102, + "status": "unread", + "hash": "def456", + "title": "Sixty days of RSS triage: a field report", + "url": "https://blog.other.example/triage-report", + "comments_url": "", + "published_at": "2024-03-04T14:05:00Z", + "created_at": "2024-03-04T15:20:00Z", + "changed_at": "2024-03-04T15:20:00Z", + "content": "<p>Another body.</p>", + "author": "", + "share_code": "", + "starred": false, + "reading_time": 9, + "enclosures": null, + "feed": { + "id": 102, + "user_id": 3, + "feed_url": "https://blog.other.example/rss", + "site_url": "https://blog.other.example", + "title": "Other Blog", + "checked_at": "2024-03-05T09:00:00Z", + "etag_header": "", + "last_modified_header": "", + "parsing_error_message": "", + "parsing_error_count": 0 + } + } + ] +}` + +// fakeServer records the last request's auth header and query, then replies +// with the fixture (or an error status). +type fakeServer struct { + t *testing.T + status int + body string + gotHeader string + gotQuery map[string]string + sawRequest bool +} + +func (f *fakeServer) handler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + f.sawRequest = true + f.gotHeader = r.Header.Get("X-Auth-Token") + f.gotQuery = map[string]string{} + for k, v := range r.URL.Query() { + f.gotQuery[k] = v[0] + } + if r.URL.Path != "/v1/entries" { + f.t.Errorf("unexpected path: %s", r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(f.status) + _, _ = w.Write([]byte(f.body)) + } +} + +func newFakeClient(t *testing.T, f *fakeServer) *Client { + t.Helper() + srv := httptest.NewServer(f.handler()) + t.Cleanup(srv.Close) + return NewClientWithHTTP(Config{BaseURL: srv.URL, Token: "test-token-123"}, srv.Client()) +} + +func TestEntriesDecodesFixture(t *testing.T) { + f := &fakeServer{t: t, status: http.StatusOK, body: entriesFixture} + client := newFakeClient(t, f) + + set, err := client.Entries(Filters{}) + if err != nil { + t.Fatalf("Entries: %v", err) + } + + if !f.sawRequest { + t.Fatal("fake server saw no request") + } + if f.gotHeader != "test-token-123" { + t.Errorf("X-Auth-Token = %q, want test-token-123", f.gotHeader) + } + if set.Total != 2 { + t.Errorf("Total = %d, want 2", set.Total) + } + if len(set.Entries) != 2 { + t.Fatalf("len(Entries) = %d, want 2", len(set.Entries)) + } + + e := set.Entries[0] + if e.ID != 41001 { + t.Errorf("entry[0].ID = %d, want 41001", e.ID) + } + if e.Title != "Why headline-only models are fine actually" { + t.Errorf("entry[0].Title = %q", e.Title) + } + if !e.Starred { + t.Error("entry[0].Starred = false, want true") + } + if e.Status != "read" { + t.Errorf("entry[0].Status = %q, want read", e.Status) + } + if e.Feed.ID != 77 || e.Feed.Title != "Example Blog" { + t.Errorf("entry[0].Feed = %+v", e.Feed) + } + if e.PublishedAt.IsZero() || e.ChangedAt.IsZero() { + t.Errorf("entry[0] times not parsed: %v / %v", e.PublishedAt, e.ChangedAt) + } +} + +func TestEntriesSendsFilters(t *testing.T) { + f := &fakeServer{t: t, status: http.StatusOK, body: `{"total":0,"entries":[]}`} + client := newFakeClient(t, f) + + starred := true + _, err := client.Entries(Filters{ + Starred: &starred, + Status: "read", + ChangedAfter: 1700000000, + PublishedAfter: 1700000000, + AfterEntryID: 40999, + Limit: 200, + Offset: 400, + }) + if err != nil { + t.Fatalf("Entries: %v", err) + } + + want := map[string]string{ + "starred": "true", + "status": "read", + "changed_after": "1700000000", + "published_after": "1700000000", + "after_entry_id": "40999", + "limit": "200", + "offset": "400", + } + for k, v := range want { + if f.gotQuery[k] != v { + t.Errorf("query %s = %q, want %q", k, f.gotQuery[k], v) + } + } +} + +func TestEntriesZeroFiltersSendNoParams(t *testing.T) { + f := &fakeServer{t: t, status: http.StatusOK, body: `{"total":0,"entries":[]}`} + client := newFakeClient(t, f) + + if _, err := client.Entries(Filters{}); err != nil { + t.Fatalf("Entries: %v", err) + } + if len(f.gotQuery) != 0 { + t.Errorf("zero Filters sent params: %v", f.gotQuery) + } +} + +func TestEntriesHTTPError(t *testing.T) { + f := &fakeServer{t: t, status: http.StatusUnauthorized, body: `{"error_message":"bad token"}`} + client := newFakeClient(t, f) + + _, err := client.Entries(Filters{}) + if err == nil { + t.Fatal("expected error on 401, got nil") + } + if !strings.Contains(err.Error(), "401") { + t.Errorf("error should name the status: %v", err) + } +} + +func TestEntriesBadJSON(t *testing.T) { + f := &fakeServer{t: t, status: http.StatusOK, body: `not json`} + client := newFakeClient(t, f) + + if _, err := client.Entries(Filters{}); err == nil { + t.Fatal("expected decode error, got nil") + } +} + +func TestConfigFromEnv(t *testing.T) { + t.Run("primary vars", func(t *testing.T) { + t.Setenv("MINIFLUX_URL", "https://flux.example") + t.Setenv("MINIFLUX_TOKEN", "tok-primary") + cfg, err := ConfigFromEnv() + if err != nil { + t.Fatalf("ConfigFromEnv: %v", err) + } + if cfg.BaseURL != "https://flux.example" || cfg.Token != "tok-primary" { + t.Errorf("cfg = %+v", cfg) + } + }) + + t.Run("fallback vars", func(t *testing.T) { + t.Setenv("MFLUX_URL", "https://flux.example") + t.Setenv("MFLUX_TOKEN", "tok-fallback") + cfg, err := ConfigFromEnv() + if err != nil { + t.Fatalf("ConfigFromEnv: %v", err) + } + if cfg.Token != "tok-fallback" { + t.Errorf("Token = %q, want tok-fallback", cfg.Token) + } + }) + + t.Run("primary beats fallback", func(t *testing.T) { + t.Setenv("MINIFLUX_TOKEN", "tok-primary") + t.Setenv("MFLUX_TOKEN", "tok-fallback") + t.Setenv("MINIFLUX_URL", "https://flux.example") + cfg, err := ConfigFromEnv() + if err != nil { + t.Fatalf("ConfigFromEnv: %v", err) + } + if cfg.Token != "tok-primary" { + t.Errorf("Token = %q, want tok-primary", cfg.Token) + } + }) + + t.Run("missing URL errors naming both vars", func(t *testing.T) { + t.Setenv("MINIFLUX_TOKEN", "tok") + _, err := ConfigFromEnv() + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "MINIFLUX_URL") || !strings.Contains(err.Error(), "MFLUX_URL") { + t.Errorf("error should name both URL vars: %v", err) + } + }) + + t.Run("missing token errors naming both vars", func(t *testing.T) { + t.Setenv("MINIFLUX_URL", "https://flux.example") + _, err := ConfigFromEnv() + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "MINIFLUX_TOKEN") || !strings.Contains(err.Error(), "MFLUX_TOKEN") { + t.Errorf("error should name both token vars: %v", err) + } + }) +} + +// Compile-time assertion: EntrySet round-trips through encoding/json (a +// guard against tag typos drifting from the API's field names). +func TestEntrySetJSONRoundTrip(t *testing.T) { + var set EntrySet + if err := json.Unmarshal([]byte(entriesFixture), &set); err != nil { + t.Fatalf("fixture does not decode into EntrySet: %v", err) + } + b, err := json.Marshal(set) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var back EntrySet + if err := json.Unmarshal(b, &back); err != nil { + t.Fatalf("re-decode: %v", err) + } + if back.Total != set.Total || len(back.Entries) != len(set.Entries) { + t.Error("round trip mismatch") + } +} |