// 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. A star outranks the read state // that put an entry in the negative pool: it flips the stored negative row // (see relabel.go), or — for a fluxrec pick whose original was never // exported — is recorded as its own positive. // 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 NegFeedCapPct int } 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. No single feed may hold more than --neg-feed-cap-pct percent of the negative target (floored at 50 rows), so a few firehoses can't own the negative class. Each run also writes a JSON summary to --run-out. A star always wins over the read state that produced a negative: a starred entry already labeled negative is relabeled to positive in place (identity kept, fresh captured_at, original copied to LABELS.bak), and a star on a rec-feed item whose original was never exported is added as its own positive (URL is the original's; title/feed identity are the rec copy's). 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)") fs.IntVar(&c.NegFeedCapPct, "neg-feed-cap-pct", 5, "Per-feed negative ceiling as % of target (0 = uncapped; floor 50)") 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, recAdded, relabeled 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 // Every star seen this run, so rows already in the store can be flipped // to positive after the walk (relabelStars). starIDs := make(map[int64]bool) starURLs := make(map[string]bool) 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 } starIDs[e.ID] = true starURLs[core.NormalizeURL(e.URL)] = true // Stars on the recommendations feed are not original entries. // A star outranks whatever the original's read state said, so // resolve by URL against existing labels when possible (the // relabel pass below flips a stored negative), and otherwise // record the star as its own positive: the URL is the original // article's, only the title/feed identity come from the rec copy. if c.RecFeedID != 0 && e.Feed.ID == c.RecFeedID { if seenURLs[core.NormalizeURL(e.URL)] { counts.recHit++ continue } nu := core.NormalizeURL(e.URL) 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++ counts.recAdded++ 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 } } // A star on an entry already in the store as a negative flips that row: // same identity, label positive, captured_at refreshed so the star reads // as recent evidence. Rewritten in place, backup first — the rows the // dedup counters above skipped as duplicates are exactly the ones this // rescues. if flipped := relabelStars(existing, starIDs, starURLs, now); flipped > 0 { if err := rewriteLabels(c.Labels, c.Labels+".bak", existing); err != nil { return err } counts.relabeled = flipped } 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 // negPerFeed is each feed's existing negative count: the cap is on the // feed's total share of the pool, so history counts against it. negPerFeed := make(map[int64]int) for _, row := range existing { if row.Label == core.LabelPos { posTotal++ } else { negTotal++ negPerFeed[row.FeedID]++ } } posTotal += counts.appended fmt.Fprintf(stdout, "export: seen=%d appended=%d relabeled=%d (id-dup=%d url-dup=%d rec-hit=%d rec-added=%d) pos-total=%d\n", counts.seen, counts.appended, counts.relabeled, counts.idDup, counts.urlDup, counts.recHit, counts.recAdded, posTotal) if counts.relabeled > 0 { fmt.Fprintf(stdout, "note: relabeled %d starred entries that were stored as negatives (backup: %s)\n", counts.relabeled, c.Labels+".bak") } if counts.recAdded > 0 { fmt.Fprintf(stdout, "note: %d rec-feed stars had no labeled original and were added as positives\n", counts.recAdded) } nc, err := c.exportNegatives(client, seenIDs, seenURLs, negPerFeed, 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, NegCapped: nc.capped, Relabeled: counts.relabeled, IDDup: counts.idDup, URLDup: counts.urlDup, RecHit: counts.recHit, RecAdded: counts.recAdded, }) } // negCounts is what the negatives sampler reports back. type negCounts struct { candidates, appended, capped 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. // // negPerFeed carries each feed's existing negative count and is updated as // rows are added: the --neg-feed-cap-pct ceiling applies to a feed's total // pool share, so candidates from feeds already at their ceiling are skipped // and the budget flows to other feeds. A capped-out pool can therefore // finish short of the ratio target, which is reported, not hidden. // // 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, negPerFeed map[int64]int, 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 } cap := feedNegCap(c.NegFeedCapPct, c.NegRatio, posTotal) 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) // Seeded permutation walk: same candidate pool + seed = same sample. // The ceiling filters the walk rather than reordering it, so enabling // or disabling it can't perturb the sampled order of accepted rows. rng := rand.New(rand.NewSource(c.NegSeed)) perm := rng.Perm(len(cands)) rows := make([]core.LabelRow, 0, min(need, len(cands))) for _, idx := range perm { if len(rows) >= need { break } e := cands[idx] if cap > 0 && negPerFeed[e.Feed.ID] >= cap { nc.capped++ continue } 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, }) negPerFeed[e.Feed.ID]++ 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 capped=%d (target=%d:%d cap=%s)\n", nc.candidates, nc.appended, nc.capped, c.NegRatio, posTotal, capLabel(cap)) if nc.appended < need { fmt.Fprintf(stdout, "note: negatives short by %d (candidates=%d in the last %d days, capped-out=%d)\n", need-nc.appended, nc.candidates, c.NegDays, nc.capped) } return nc, nil } // capLabel renders a per-feed ceiling for the run line ("off" when uncapped). func capLabel(cap int) string { if cap <= 0 { return "off" } return fmt.Sprintf("%d/feed", cap) } // 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"` NegCapped int `json:"neg_capped"` Relabeled int `json:"relabeled"` IDDup int `json:"id_dup"` URLDup int `json:"url_dup"` RecHit int `json:"rec_hit"` RecAdded int `json:"rec_added"` } 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 }