1package source
  2
  3import (
  4	"bufio"
  5	"encoding/json"
  6	"fmt"
  7	"io"
  8	"math"
  9	"os"
 10	"sort"
 11	"strconv"
 12	"strings"
 13	"time"
 14
 15	"reelreview/internal/model"
 16)
 17
 18func ItemKey(s string) string { return strings.ToLower(strings.ReplaceAll(s, "-", "")) }
 19
 20// Load validates a finalized canonical snapshot; inconsistent duplicates fail closed.
 21func Load(path string) (model.Snapshot, error) {
 22	var s model.Snapshot
 23	f, err := os.Open(path)
 24	if err != nil {
 25		return s, fmt.Errorf("cannot open source snapshot")
 26	}
 27	defer f.Close()
 28	if st, e := f.Stat(); e != nil || st.Size() > 256<<20 {
 29		return s, fmt.Errorf("source snapshot exceeds size limit or cannot be read")
 30	}
 31	d := json.NewDecoder(io.LimitReader(f, 256<<20))
 32	d.DisallowUnknownFields()
 33	if err := d.Decode(&s); err != nil {
 34		return s, fmt.Errorf("invalid source snapshot: %w", err)
 35	}
 36	if d.Decode(new(any)) != io.EOF {
 37		return s, fmt.Errorf("trailing or oversized source data")
 38	}
 39	return s, Validate(&s)
 40}
 41func Validate(s *model.Snapshot) error {
 42	if s.Version != 1 || s.CapturedAt.IsZero() || s.Source == "" || s.Coverage == "" {
 43		return fmt.Errorf("snapshot requires v1, capturedAt, source and explicit coverage")
 44	}
 45	users := map[string]bool{}
 46	for i := range s.Users {
 47		u := &s.Users[i]
 48		id, err := model.ID(u.ID)
 49		if err != nil || users[id] {
 50			return fmt.Errorf("invalid or duplicate source user")
 51		}
 52		u.ID = id
 53		users[id] = true
 54	}
 55	items := map[string]bool{}
 56	for i := range s.Items {
 57		v := &s.Items[i]
 58		v.ID = ItemKey(v.ID)
 59		v.SeriesID = ItemKey(v.SeriesID)
 60		if v.ID == "" || items[v.ID] || v.Runtime < 0 || math.IsNaN(v.Runtime) || math.IsInf(v.Runtime, 0) {
 61			return fmt.Errorf("invalid or duplicate media")
 62		}
 63		items[v.ID] = true
 64	}
 65	seen := map[string]string{}
 66	out := []model.Playback{}
 67	for _, p := range s.Playback {
 68		var err error
 69		p.UserID, err = model.ID(p.UserID)
 70		if err != nil {
 71			return fmt.Errorf("invalid playback user")
 72		}
 73		p.ItemID = ItemKey(p.ItemID)
 74		if p.ID == "" || p.ItemID == "" || math.IsNaN(p.Seconds) || math.IsInf(p.Seconds, 0) || p.Seconds <= 0 || p.Seconds > 366*86400 {
 75			return fmt.Errorf("invalid playback duration or identity")
 76		}
 77		if p.Exact {
 78			if !p.End.After(p.Start) || p.Start.IsZero() || math.Abs(p.End.Sub(p.Start).Seconds()-p.Seconds) > 0.01 {
 79				return fmt.Errorf("exact intervals must match watched seconds")
 80			}
 81		} else if p.ObservedAt.IsZero() {
 82			return fmt.Errorf("aggregate activity requires observedAt")
 83		}
 84		if (p.PositionStart == nil) != (p.PositionEnd == nil) {
 85			return fmt.Errorf("both position bounds required")
 86		}
 87		if p.PositionStart != nil && (math.IsNaN(*p.PositionStart) || math.IsInf(*p.PositionStart, 0) || *p.PositionStart < 0 || *p.PositionEnd < *p.PositionStart || math.IsNaN(*p.PositionEnd) || math.IsInf(*p.PositionEnd, 0)) {
 88			return fmt.Errorf("invalid position range")
 89		}
 90		b, _ := json.Marshal(p)
 91		key := p.UserID + ":" + p.ID
 92		if prev, ok := seen[key]; ok {
 93			if prev != string(b) {
 94				return fmt.Errorf("conflicting duplicate playback ID")
 95			}
 96			continue
 97		}
 98		seen[key] = string(b)
 99		out = append(out, p)
100	}
101	s.Playback = out
102	sort.Slice(s.Users, func(i, j int) bool { return s.Users[i].ID < s.Users[j].ID })
103	sort.Slice(s.Items, func(i, j int) bool { return s.Items[i].ID < s.Items[j].ID })
104	sort.Slice(s.Playback, func(i, j int) bool {
105		a, b := s.Playback[i], s.Playback[j]
106		if a.UserID != b.UserID {
107			return a.UserID < b.UserID
108		}
109		return a.ID < b.ID
110	})
111	return nil
112}
113
114// Jellystat reads the raw table JSONL contract, deliberately rejecting grouped API
115// history: grouped durations and resumed records cannot recover exact watch intervals.
116func Jellystat(path string, users []model.User, items []model.Item, captured time.Time, coverage string) (model.Snapshot, error) {
117	s := model.Snapshot{Version: 1, CapturedAt: captured, Source: "Jellystat raw finalized activity", Coverage: coverage, Users: users, Items: items}
118	f, err := os.Open(path)
119	if err != nil {
120		return s, fmt.Errorf("cannot open Jellystat export")
121	}
122	defer f.Close()
123	if st, e := f.Stat(); e != nil || st.Size() > 256<<20 {
124		return s, fmt.Errorf("Jellystat snapshot exceeds size limit or cannot be read")
125	}
126	scan := bufio.NewScanner(io.LimitReader(f, 256<<20))
127	scan.Buffer(make([]byte, 4096), 1<<20)
128	for scan.Scan() {
129		var row struct {
130			ID       string      `json:"Id"`
131			UserID   string      `json:"UserId"`
132			Item     string      `json:"NowPlayingItemId"`
133			Episode  string      `json:"EpisodeId"`
134			Duration json.Number `json:"PlaybackDuration"`
135			At       time.Time   `json:"ActivityDateInserted"`
136			Device   string      `json:"DeviceName"`
137			Client   string      `json:"Client"`
138			Imported bool        `json:"Imported"`
139		}
140		if err := json.Unmarshal(scan.Bytes(), &row); err != nil {
141			return s, fmt.Errorf("invalid Jellystat raw row")
142		}
143		if row.Imported {
144			return s, fmt.Errorf("Playback Reporting imports are not supported; export native Jellystat activity only")
145		}
146		n, err := strconv.ParseFloat(string(row.Duration), 64)
147		if err != nil {
148			return s, fmt.Errorf("invalid Jellystat PlaybackDuration")
149		}
150		if row.Episode != "" {
151			row.Item = row.Episode
152		}
153		s.Playback = append(s.Playback, model.Playback{ID: row.ID, UserID: row.UserID, ItemID: row.Item, ObservedAt: row.At, Seconds: n, Device: row.Device, Client: row.Client})
154	}
155	if scan.Err() != nil {
156		return s, fmt.Errorf("Jellystat export exceeds row limit or cannot be read")
157	}
158	return s, Validate(&s)
159}