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