1 files changed, 402 insertions, 0 deletions
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
+}