1package fileio23import (4 "encoding/csv"5 "errors"6 "fmt"7 "io"8 "runtime"9 "strings"10 "sync"1112 "src.rest/cat/012/internal/locale"13 "src.rest/cat/012/internal/nuon"14 "src.rest/cat/012/internal/sheet"15)1617// Tails: a text table read as it arrives in pieces, as a file followed18// grows. The text is parsed by the importers' own readers (encoding/csv19// and nuon.Reader) on a goroutine of the tail's, reading from a pipe20// that waits for the next piece: a record or value cut off at the end of21// one piece waits there for the rest, so a partial last line is never22// read as a row. Rows come out typed as the importer types them: CSV and23// TSV fields as if typed in the locale, NUON and JSON values with their24// types (nuonCell).2526// Grows reports whether a file of kind k can be followed as it grows,27// its new rows read without reading it again: text tables, which28// Tail reads.29func (k Kind) Grows() bool { return k == CSV || k == TSV || k == JSON || k == NUON }3031// TailRows are the rows a piece of text completed.32type TailRows struct {33 // Header is the table's first row when it's new or changed: a CSV34 // file's first record, or NUON's column names as they're seen.35 Header sheet.LiveRow36 Rows []sheet.LiveRow37}3839// A Tail reads a table of kind k from text that arrives in pieces.40type Tail struct {41 p *tailPipe42 out *tailOut43}4445// tailPipe hands pieces of text to the tail's goroutine and tells when46// it has read them all: it has then parsed every row they complete.47type tailPipe struct {48 in chan []byte49 starved chan struct{} // the goroutine wants more50 done chan struct{} // the goroutine ended51 buf []byte52 eof bool53 once sync.Once54}5556// tailOut is what the goroutine parsed since the last piece; the Tail57// reads it only while the goroutine waits.58type tailOut struct {59 header sheet.LiveRow60 changed bool // header changed61 rows []sheet.LiveRow62 err error63}6465// Read gives the goroutine's reader the rest of the piece, or waits for66// the next one.67func (p *tailPipe) Read(b []byte) (int, error) {68 for len(p.buf) == 0 {69 if p.eof {70 return 0, io.EOF71 }72 p.starved <- struct{}{}73 chunk, ok := <-p.in74 if !ok {75 p.eof = true76 return 0, io.EOF77 }78 p.buf = chunk79 }80 n := copy(b, p.buf)81 p.buf = p.buf[n:]82 return n, nil83}8485func (p *tailPipe) close() { p.once.Do(func() { close(p.in) }) }8687// NewTail starts reading a table of kind k (see Kind.Grows). head is the88// start of the text, what it's sniffed by (its encoding, and a CSV89// file's delimiter and decimal separator), in the locale loc; the text90// itself comes through Feed, head included.91func NewTail(k Kind, head []byte, loc *locale.Locale) (*Tail, error) {92 if !k.Grows() {93 return nil, fmt.Errorf("can't follow %s files as they grow", k)94 }95 p := &tailPipe{in: make(chan []byte), starved: make(chan struct{}), done: make(chan struct{})}96 out := &tailOut{}97 var parse func()98 switch k {99 case CSV, TSV:100 parse = delimitedTail(p, out, k, head, loc)101 default:102 parse = tableTail(p, out)103 }104 go func() {105 defer close(p.done)106 parse()107 }()108 t := &Tail{p: p, out: out}109 // A tail dropped without Close ends its goroutine too.110 runtime.AddCleanup(t, func(p *tailPipe) { p.close() }, p)111 select {112 case <-p.starved:113 case <-p.done:114 }115 return t, nil116}117118// errTailClosed is Feed's error once the tail is closed or failed.119var errTailClosed = errors.New("the tail is closed")120121// Feed reads the next piece of text and returns the rows it completed.122// After an error the tail reads nothing more.123func (t *Tail) Feed(piece []byte) (TailRows, error) {124 select {125 case <-t.p.done:126 return TailRows{}, t.failure()127 default:128 }129 if len(piece) > 0 {130 t.p.in <- piece131 select {132 case <-t.p.starved:133 case <-t.p.done:134 }135 }136 out := t.out137 res := TailRows{Rows: out.rows}138 if out.changed {139 res.Header = out.header140 }141 out.rows, out.changed = nil, false142 if out.err != nil {143 return res, out.err144 }145 return res, nil146}147148// failure is why the goroutine ended.149func (t *Tail) failure() error {150 if t.out.err != nil {151 return t.out.err152 }153 return errTailClosed154}155156// Close ends the tail's goroutine.157func (t *Tail) Close() { t.p.close() }158159// delimitedTail parses CSV or TSV: fields as if typed, the delimiter and160// the numbers' decimal separator sniffed from head as an import does.161func delimitedTail(p *tailPipe, out *tailOut, k Kind, head []byte, loc *locale.Locale) func() {162 enc, bom := encodingOf(head)163 sample := head[bom:]164 if enc != encUTF8 {165 b, _ := io.ReadAll(decodeAs(strings.NewReader(string(sample)), enc))166 sample = b167 }168 comma := '\t'169 if k == CSV {170 comma = sniff(sample, len(head) >= sniffSize, csvComma(loc))171 }172 loc, _ = numberLocale(sample, comma, loc)173 return func() {174 skip := &skipReader{r: p, n: bom}175 cr := csv.NewReader(decodeAs(skip, enc))176 cr.Comma, cr.LazyQuotes, cr.FieldsPerRecord = comma, true, -1177 first := true178 for {179 rec, err := cr.Read()180 if err != nil {181 if !errors.Is(err, io.EOF) {182 out.err = err183 }184 return185 }186 row := make(sheet.LiveRow, len(rec))187 for i, field := range rec {188 if field != "" {189 v, f := sheet.EntryValue(entryInput(field, loc))190 row[i] = sheet.LiveCell{V: v, F: f}191 }192 }193 if first {194 first, out.header, out.changed = false, row, true195 continue196 }197 out.rows = append(out.rows, row)198 }199 }200}201202// skipReader leaves out the first n bytes of r: a byte order mark.203type skipReader struct {204 r io.Reader205 n int206}207208func (s *skipReader) Read(b []byte) (int, error) {209 for s.n > 0 {210 var one [4]byte211 m, err := s.r.Read(one[:min(s.n, len(one))])212 s.n -= m213 if err != nil {214 return 0, err215 }216 }217 return s.r.Read(b)218}219220// tableTail parses NUON or JSON: the columns named as rows name them,221// the header growing as new names appear.222func tableTail(p *tailPipe, out *tailOut) func() {223 return func() {224 r := nuon.NewReader(p)225 cols := map[string]int{}226 zone := zone()227 col := func(name string) int {228 i, ok := cols[name]229 if !ok {230 i = len(cols)231 cols[name] = i232 out.header = append(out.header, textCell(name))233 out.changed = true234 }235 return i236 }237 for {238 fields, err := r.Next()239 if err != nil {240 if !errors.Is(err, io.EOF) {241 out.err = err242 }243 return244 }245 for _, c := range r.Header() {246 col(c)247 }248 row := make(sheet.LiveRow, len(cols))249 for _, f := range fields {250 i := col(f.Key)251 if i >= len(row) {252 row = append(row, make(sheet.LiveRow, i+1-len(row))...)253 }254 row[i] = nuonCell(f.Value, zone)255 }256 out.rows = append(out.rows, row)257 }258 }259}260261// Rows reads a sheet an import made as a linked region's rows: its first262// row as the header and the rest under it, from column A, values with263// the formats they show in. A file that is read again whole when it264// changes (XLSX, SQLite, Parquet) is read this way.265func Rows(s *sheet.Sheet) TailRows {266 used, ok := s.UsedRange()267 if !ok {268 return TailRows{}269 }270 var res TailRows271 for r := 0; r <= used.To.Row; r++ {272 row := make(sheet.LiveRow, used.To.Col+1)273 for c := range row {274 a := sheet.Addr{Col: c, Row: r}275 if s.Filled(a) {276 row[c] = sheet.LiveCell{V: s.Value(a), F: s.DisplayFormat(a)}277 }278 }279 if r == 0 {280 res.Header = row281 continue282 }283 res.Rows = append(res.Rows, row)284 }285 return res286}287288// textCell is a cell showing s as text, as the builder stores it.289func textCell(s string) sheet.LiveCell {290 v, _ := sheet.EntryValue("'" + flatten(s))291 return sheet.LiveCell{V: v}292}