1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
package dataset
import (
"bufio"
"bytes"
"fmt"
"iter"
"os"
"path/filepath"
"slices"
)
type FileReader struct {
shards []string
delimiter string
err error
}
func NewFileReader(name, pattern string) (*FileReader, error) {
var shards []string
if matches, err := filepath.Glob(filepath.Join(name, pattern)); err != nil {
return nil, err
} else {
shards = matches
}
if len(shards) == 0 {
return nil, fmt.Errorf("no matching files in %q", name)
}
slices.Sort(shards)
return &FileReader{
shards: shards,
delimiter: "\n",
}, nil
}
func (f *FileReader) SetDelimiter(delimiter string) *FileReader {
f.delimiter = delimiter
return f
}
func (f *FileReader) Num() (int, error) {
return len(f.shards), nil
}
func (f *FileReader) Err() error {
return f.err
}
func (f *FileReader) Texts() iter.Seq2[int, string] {
return func(yield func(int, string) bool) {
for n, name := range f.shards {
for text := range f.read(name) {
if f.err != nil {
return
}
if !yield(n, text) {
return
}
}
}
}
}
func (f *FileReader) read(name string) iter.Seq[string] {
return func(yield func(string) bool) {
var file *os.File
if r, err := os.Open(name); err != nil {
f.err = err
return
} else {
file = r
}
defer file.Close()
d := []byte(f.delimiter)
l := len(d)
scanner := bufio.NewScanner(file)
buf := make([]byte, 0, 1024*1024)
scanner.Buffer(buf, 1024*1024)
scanner.Split(func(data []byte, atEOF bool) (advance int, token []byte, err error) {
if atEOF && len(data) == 0 {
return 0, nil, nil
}
if i := bytes.Index(data, d); i >= 0 {
return i + l, data[:i+l], nil // TODO option to consume delimiter
}
if atEOF {
return len(data), data, nil
}
return 0, nil, nil
})
for scanner.Scan() {
if !yield(scanner.Text()) {
return
}
}
if err := scanner.Err(); err != nil {
f.err = err
}
}
}
|
