// 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 }