-
Notifications
You must be signed in to change notification settings - Fork 2
/
artworkrecord-score.go
121 lines (102 loc) · 2.5 KB
/
artworkrecord-score.go
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
120
121
package main
import (
"context"
"flag"
"fmt"
"log"
"cloud.google.com/go/datastore"
"github.com/nishanths/scrobble/appengine/artwork"
"golang.org/x/sync/errgroup"
"google.golang.org/api/iterator"
"google.golang.org/api/option"
)
const credFile = "../selective-scrobble-8033d812577b.json"
var (
namespaces = []string{
"6e697368616e74682e6765727261726440676d61696c2e636f6d",
}
)
var (
fParallel = flag.Int("parallel", 64, "number of worker goroutines")
)
func main() {
if err := run(context.Background()); err != nil {
log.Fatal(err)
}
}
type work struct {
namespace string
hash string
}
func run(ctx context.Context) error {
ds, err := datastore.NewClient(ctx, "selective-scrobble", option.WithCredentialsFile(credFile))
if err != nil {
return fmt.Errorf("datastore client: %s", err)
}
defer ds.Close()
g, ctx := errgroup.WithContext(ctx)
workCh := make(chan work)
g.Go(func() error {
defer close(workCh)
for _, namespace := range namespaces {
q := datastore.NewQuery(artwork.KindArtworkRecord).Namespace(namespace).KeysOnly()
it := ds.Run(ctx, q)
for {
key, err := it.Next(nil)
if err == iterator.Done {
break
}
if err != nil {
return fmt.Errorf("iterator: %s", err)
}
workCh <- work{key.Namespace, key.Name}
}
}
return nil
})
for i := 0; i < *fParallel; i++ {
g.Go(func() error {
return handle(ctx, ds, workCh)
})
}
return g.Wait()
}
func handle(ctx context.Context, ds *datastore.Client, workCh chan work) error {
for k := range workCh {
if err := handleOne(ctx, ds, k); err != nil {
return fmt.Errorf("handling %s", err)
}
}
return nil
}
func handleOne(ctx context.Context, ds *datastore.Client, w work) error {
namespace := w.namespace
hash := w.hash
var as artwork.ArtworkScore
err := ds.Get(ctx, artwork.ArtworkScoreKey(hash), &as)
if err == datastore.ErrNoSuchEntity {
log.Printf("no global ArtworkScore for %s", hash)
return nil
}
if err != nil {
return fmt.Errorf("datastore get artwork score: %s", err)
}
_, err = ds.RunInTransaction(ctx, func(tx *datastore.Transaction) error {
var ar artwork.ArtworkRecord
err := tx.Get(artwork.ArtworkRecordKey(namespace, hash), &ar)
if err != nil {
return fmt.Errorf("datastore get artwork record: %s", err)
}
ar.Score = as
_, err = tx.Put(artwork.ArtworkRecordKey(namespace, hash), &ar)
if err != nil {
return fmt.Errorf("datastore put artwork record: %s", err)
}
return nil
})
if err != nil {
return err
}
log.Printf("put %v", w)
return nil
}