path: root/cmds/export.go

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