aboutsummaryrefslogtreecommitdiffstats
path: root/skate/zippy.go
blob: ff836e8a49a4b382e3d561b4dce1ea3ffc5ed836 (plain)
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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
// This file contains various "reducers", e.g. merging data from two streams and
// applying a function on groups of documents with a shared key.
//
// Note: This is a bit repetitive, but we do not want to introduce any other
// abstraction for now. Since most of the logic is in the "grouper" functions,
// we could make them top level values and then assemble the zipkey runner on
// the fly.
//
// The most confusing aspect currently is the variety of schemas hidden within
// the readers (and string groups): release, ref, ref-as-release, open library,
// wikipedia, ...
//
// TODO: [ ] pass release stage through all match types
// TODO: [ ] switch to faster logging, e.g. zerolog, https://github.com/rs/zerolog#benchmarks
package skate

import (
	"fmt"
	"io"
	"log"
	"sort"
	"strings"
	"sync"
	"time"

	"git.archive.org/martin/cgraph/skate/set"
	"git.archive.org/martin/cgraph/skate/zipkey"
	json "github.com/segmentio/encoding/json"
)

var brefPool = sync.Pool{
	New: func() interface{} {
		var bref BiblioRef
		return bref
	},
}

// groupLogf logs a message alongsize a serialized group for debugging.
func groupLogf(g *zipkey.Group, s string, vs ...interface{}) {
	log.Printf(s, vs...)
	b, _ := json.MarshalIndent(g, "", "    ")
	log.Println(string(b))
}

// ZippyExact takes a release and refs reader (key, doc) and assigns a fixed
// match result, e.g. for doi matches.
func ZippyExact(releases, refs io.Reader, matchResult MatchResult, w io.Writer) error {
	var (
		enc     = json.NewEncoder(w)
		keyer   = makeKeyFunc("\t", 1)
		i       = 0
		bref    BiblioRef
		grouper = func(g *zipkey.Group) error {
			i++
			if i%10000 == 0 {
				log.Printf("processed %v groups", i)
			}
			var (
				target *Release
				ref    *Ref
				err    error
			)
			if len(g.G0) == 0 || len(g.G1) == 0 {
				return nil
			}
			if target, err = parseRelease(Cut(g.G0[0], 2)); err != nil {
				groupLogf(g, "[skip] failed to parse release: %v", err)
				return nil
			}
			for _, line := range g.G1 {
				if ref, err = parseRef(Cut(line, 2)); err != nil {
					groupLogf(g, "[skip] failed to parse ref: %v", err)
					continue
				}
				bref = brefPool.Get().(BiblioRef)
				bref.Reset()
				bref.SourceReleaseIdent = ref.ReleaseIdent
				bref.SourceWorkIdent = ref.WorkIdent
				bref.SourceReleaseStage = ref.ReleaseStage
				bref.SourceYear = fmt.Sprintf("%d", ref.ReleaseYear)
				bref.RefIndex = ref.Index + 1 // we want 1-index (also helps with omitempty)
				bref.RefKey = ref.Key
				bref.TargetReleaseIdent = target.Ident
				bref.TargetWorkIdent = target.WorkID
				bref.MatchProvenance = ref.RefSource
				bref.MatchStatus = matchResult.Status.Short()
				bref.MatchReason = matchResult.Reason.Short()
				if err := enc.Encode(bref); err != nil {
					return err
				}
				brefPool.Put(bref)
			}
			return nil
		}
	)
	zipper := zipkey.New(releases, refs, keyer, grouper)
	return zipper.Run()
}

// ZippyExactReleases takes two release readers (key, doc) and assigns a fixed
// match result, e.g. used with release entities converted from open library snapshots.
func ZippyExactReleases(olr, releases io.Reader, matchResult MatchResult, w io.Writer) error {
	var (
		enc     = json.NewEncoder(w)
		keyer   = makeKeyFunc("\t", 1)
		i       = 0
		bref    BiblioRef
		grouper = func(g *zipkey.Group) error {
			i++
			if i%10000 == 0 {
				log.Printf("processed %v groups", i)
			}
			var (
				target, re *Release
				err        error
			)
			if len(g.G0) == 0 || len(g.G1) == 0 {
				return nil
			}
			if target, err = parseRelease(Cut(g.G0[0], 2)); err != nil {
				groupLogf(g, "[skip] failed to parse release: %v", err)
				return nil
			}
			for _, line := range g.G1 {
				if re, err = parseRelease(Cut(line, 2)); err != nil {
					groupLogf(g, "[skip] failed to parse release: %v", err)
					continue
				}
				if target.WorkID == "" {
					continue
				}
				bref = brefPool.Get().(BiblioRef)
				bref.Reset()
				bref.SourceReleaseIdent = re.Ident
				bref.SourceWorkIdent = re.WorkID
				bref.SourceReleaseStage = re.ReleaseStage
				bref.SourceYear = fmt.Sprintf("%d", re.ReleaseYear())
				bref.RefIndex = re.Extra.Skate.Ref.Index + 1 // we want 1-index (also helps with omitempty)
				bref.RefKey = re.Extra.Skate.Ref.Key
				bref.TargetOpenLibraryWork = target.WorkID
				bref.MatchProvenance = re.Extra.Skate.Ref.Source
				bref.MatchStatus = matchResult.Status.Short()
				bref.MatchReason = matchResult.Reason.Short()
				if err := enc.Encode(bref); err != nil {
					return err
				}
				brefPool.Put(bref)
			}
			return nil
		}
	)
	zipper := zipkey.New(olr, releases, keyer, grouper)
	return zipper.Run()
}

// ZippyExactWiki takes a release and wiki reader (key, doc) and assigns a
// fixed match result.
func ZippyExactWiki(releases, wiki io.Reader, mr MatchResult, w io.Writer) error {
	var (
		enc     = json.NewEncoder(w)
		keyer   = makeKeyFunc("\t", 1)
		bref    BiblioRef
		grouper = func(g *zipkey.Group) error {
			var (
				target *Release
				wiki   *MinimalCitations
				err    error
			)
			if len(g.G0) == 0 || len(g.G1) == 0 {
				return nil
			}
			if target, err = parseRelease(Cut(g.G0[0], 2)); err != nil {
				return err
			}
			for _, line := range g.G1 {
				if wiki, err = parseWiki(Cut(line, 2)); err != nil {
					return err
				}
				bref = brefPool.Get().(BiblioRef)
				bref.Reset()
				bref.Key = fmt.Sprintf("%s_%s", slugifyString(wiki.PageTitle), target.Ident) // XXX: what should we use?
				bref.SourceWikipediaArticle = wiki.PageTitle
				bref.TargetReleaseIdent = target.Ident
				bref.TargetWorkIdent = target.WorkID
				bref.MatchProvenance = "wikipedia"
				bref.MatchStatus = mr.Status.Short()
				bref.MatchReason = mr.Reason.Short()
				if err := enc.Encode(bref); err != nil {
					return err
				}
				brefPool.Put(bref)
			}
			return nil
		}
	)
	zipper := zipkey.New(releases, wiki, keyer, grouper)
	return zipper.Run()
}

// ZippyVerifyRefs takes a release and refs (as release) reader (key, doc), run
// fuzzy verification and will emit a biblioref document, if exact or strong
// match.
func ZippyVerifyRefs(releases, refs io.Reader, w io.Writer) error {
	var (
		enc     = json.NewEncoder(w)
		keyer   = makeKeyFunc("\t", 1)
		grouper = func(g *zipkey.Group) error {
			var (
				re, pivot *Release
				err       error
			)
			if len(g.G0) == 0 || len(g.G1) == 0 {
				return nil
			}
			if pivot, err = parseRelease(Cut(g.G0[0], 2)); err != nil {
				return err
			}
			for _, line := range g.G1 {
				if re, err = parseRelease(Cut(line, 2)); err != nil {
					return err
				}
				result := Verify(pivot, re)
				switch result.Status {
				case StatusExact, StatusStrong:
					if result.Reason == ReasonDOI {
						continue
					}
					br := generateBiblioRef(re, pivot, result, "fuzzy")
					if err := enc.Encode(br); err != nil {
						return err
					}
				default:
				}
			}
			return nil
		}
	)
	zipper := zipkey.New(releases, refs, keyer, grouper)
	return zipper.Run()
}

// ZippyVerifyRefsOpenLibraryTable takes OL editions (as release) and refs (as
// release) and emits a match table for manual inspection.
func ZippyVerifyRefsOpenLibraryTable(olr, refs io.Reader, w io.Writer) error {
	var (
		keyer   = makeKeyFunc("\t", 1)
		grouper = func(g *zipkey.Group) error {
			var (
				re, pivot *Release
				err       error
			)
			if len(g.G0) == 0 || len(g.G1) == 0 {
				return nil
			}
			// We take a single edition from OL.
			if pivot, err = parseRelease(Cut(g.G0[0], 2)); err != nil {
				return err
			}
			for _, line := range g.G1 {
				if re, err = parseRelease(Cut(line, 2)); err != nil {
					return err
				}
				// The refs have a container name, but not a title, but here we
				// compare against titles from open library.
				re.Title = re.ContainerName
				result := Verify(pivot, re)
				fmt.Printf("%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n",
					result.Status.Short(),
					result.Reason.Short(),
					pivot.Extra.OpenLibrary.WorkID,
					FindByPrefix(pivot.Extra.OpenLibrary.SourceRecords, "ia:"),
					re.Ident,
					CutSep(g.G0[0], "\t", 1),
					pivot.Title,
					re.Title)
			}
			return nil
		}
	)
	zipper := zipkey.New(olr, refs, keyer, grouper)
	return zipper.Run()
}

// ZippyVerifyRefsOpenLibrary takes OL editions (as release) and refs (as
// release) and writes biblioref.
func ZippyVerifyRefsOpenLibrary(olr, refs io.Reader, w io.Writer) error {
	var (
		enc     = json.NewEncoder(w)
		keyer   = makeKeyFunc("\t", 1)
		bref    BiblioRef
		grouper = func(g *zipkey.Group) error {
			var (
				ref, pivot *Release // ref (reference), pivot (open library)
				err        error
			)
			if len(g.G0) == 0 || len(g.G1) == 0 {
				return nil
			}
			// We take a single edition from OL.
			if pivot, err = parseRelease(Cut(g.G0[0], 2)); err != nil {
				return err
			}
			for _, line := range g.G1 {
				if ref, err = parseRelease(Cut(line, 2)); err != nil {
					return err
				}
				// The refs have a container name, but not a title, but here we
				// compare against titles from open library.
				ref.Title = ref.ContainerName
				result := Verify(pivot, ref)
				switch result.Status {
				case StatusExact, StatusStrong:
					bref = brefPool.Get().(BiblioRef)
					bref.Reset()
					bref.SourceReleaseIdent = ref.Ident
					bref.SourceWorkIdent = ref.WorkID
					bref.SourceReleaseStage = ref.ReleaseStage
					bref.SourceYear = fmt.Sprintf("%d", ref.ReleaseYear())
					bref.RefIndex = ref.Extra.Skate.Ref.Index + 1 // we want 1-index (also helps with omitempty)
					bref.RefKey = ref.Extra.Skate.Ref.Key
					bref.TargetOpenLibraryWork = pivot.WorkID
					bref.MatchProvenance = ref.Extra.Skate.Ref.Source
					bref.MatchStatus = result.Status.Short()
					bref.MatchReason = result.Reason.Short()
					if err := enc.Encode(bref); err != nil {
						return err
					}
					brefPool.Put(bref)
				default:
				}
			}
			return nil
		}
	)
	zipper := zipkey.New(olr, refs, keyer, grouper)
	return zipper.Run()
}

// ZippyBrefAugment takes all matched docs from bref and adds docs from raw
// refs, which have not been matched. It also gets rid of duplicate matches.
// Note: This operates on two streams: raw refs with about 2.5B (07/2021) and
// matches, which will be about 1B; in essence we have to iterate through about
// 3.5B records; small tweak here may be worthwhile.
//
// We can identify, which docs have been matched by checking the source ident,
// ref index and key.
//
// TODO: This needs to be completed and made fast.
func ZippyBrefAugment(bref, raw io.Reader, w io.Writer) error {
	var (
		stats   = statsAugment{}
		enc     = json.NewEncoder(w)
		keyer   = makeKeyFunc("\t", 1)
		grouper = func(g *zipkey.Group) error {
			// g.G0 contains matched docs for a given work id, g.G1 all raw
			// refs, with the same work id.

			// First, iterate over all matches and sort out duplicates, e.g.
			// docs that have the same source and target id.
			log.Printf("group K=%s, G0=%d, G1=%d", g.Key, len(g.G0), len(g.G1))
			matched, err := uniqueMatches(CutBatch(g.G0, 2), &stats)
			if err != nil {
				return err
			}
			var refs = make([]*Ref, len(g.G1))
			for i := 0; i < len(refs); i++ {
				var (
					data []byte = []byte(Cut(g.G1[i], 2))
					ref  Ref
				)
				if err := json.Unmarshal(data, &ref); err != nil {
					return err
				}
				refs[i] = &ref
			}
			// TODO: this slows down this process; be a bit smarter about slices.
			matched = matchedRefsExtend(matched, refs, &stats)
			// At this point, we may have duplicates by "_id", e.g. source
			// release ident and ref index (example:
			// 4kg2dejsgzaf3cszs2lt5hz4by_9, which appears three times, one
			// exact match, and twice unmatched).
			matched = deduplicateBrefs(matched)
			matched = removeSelfLinks(matched)
			for _, bref := range matched {
				stats.total++
				if err := enc.Encode(bref); err != nil {
					return err
				}
			}
			return nil
		}
	)
	zipper := zipkey.New(bref, raw, keyer, grouper)
	err := zipper.Run()
	log.Println(stats)
	return err
}

// removeSelfLinks removes self-referential links. TODO: Those should be caught
// at the root cause.
func removeSelfLinks(brefs []*BiblioRef) (result []*BiblioRef) {
	var i int
	for _, bref := range brefs {
		if bref.SourceReleaseIdent == bref.TargetReleaseIdent {
			continue
		}
		brefs[i] = bref
		i++
	}
	brefs = brefs[:i]
	return brefs
}

// deduplicateBrefs deduplicates by the document id (for elasticsearch), which
// may help filter out some duplicates but not all.
func deduplicateBrefs(brefs []*BiblioRef) []*BiblioRef {
	// Sort by match status, exact first, unmatched last.
	sort.Slice(brefs, func(i, j int) bool {
		switch {
		case brefs[i].MatchStatus == StatusExact.Short():
			return true
		case brefs[i].MatchStatus == StatusStrong.Short():
			return true
		case brefs[i].MatchStatus == StatusWeak.Short():
			return false
		case brefs[i].MatchStatus == StatusAmbiguous.Short():
			return false
		case brefs[i].MatchStatus != StatusUnmatched.Short():
			return true
		default:
			return false
		}
	})
	var (
		seen = set.New()
		i    int
	)
	for _, v := range brefs {
		if seen.Contains(v.Key) {
			continue
		}
		brefs[i] = v
		i++
		seen.Add(v.Key)
	}
	brefs = brefs[:i]
	log.Printf("trimmed brefs from %d to %d", len(brefs), i)
	return brefs
}

// matchedRefsExtend takes a set of (unique) biblioref docs and will emit that
// set of biblioref docs (unchanged) plus raw references as biblioref, which
// did not result in a match (determined by e.g. ref key and index).  XXX: We
// may have duplicate refs as well - how to distinguish them?
func matchedRefsExtend(matched []*BiblioRef, refs []*Ref, stats *statsAugment) []*BiblioRef {
	seen := set.New() // store "key + index" of matched items
	for _, m := range matched {
		s := m.RefKey + fmt.Sprintf("%d", m.RefIndex)
		seen.Add(s)
	}
	for _, r := range refs {
		s := r.Key + fmt.Sprintf("%d", r.Index)
		if seen.Contains(s) {
			stats.skipMatchedRef++
			log.Printf("skip-matched-ref [%d]: from %d matches; ident=%v, title=%s, key=%v, index=%d",
				stats.skipMatchedRef, len(matched), r.ReleaseIdent, r.Biblio.Title, r.Key, r.Index)
			continue
		}
		var bref BiblioRef
		bref.IndexedTs = time.Now().UTC().Format(time.RFC3339)
		bref.Key = fmt.Sprintf("%s_%d", r.ReleaseIdent, r.Index)
		bref.RefIndex = r.Index
		bref.RefKey = r.Key
		bref.SourceReleaseIdent = r.ReleaseIdent
		bref.SourceReleaseStage = r.ReleaseStage
		bref.SourceWorkIdent = r.WorkIdent
		bref.SourceYear = fmt.Sprintf("%d", r.ReleaseYear)
		bref.TargetUnstructured = r.Biblio.Unstructured
		// Reuse fields for debugging, for now.
		bref.MatchStatus = StatusUnmatched.Short()
		bref.MatchReason = ReasonUnknown.Short()
		matched = append(matched, &bref)
	}
	return matched
}

// uniqueMatches takes a list of bref docs (unserialized) and will return a
// list of deserialized bref docs, containing unique matches only (e.g. filter
// out duplicate matches, e.g. from exact and fuzzy). We are including
// "skate-bref-id" post-processing here as well (but there is surely a better
// place for that).
func uniqueMatches(docs []string, stats *statsAugment) (result []*BiblioRef, err error) {
	var brefs []*BiblioRef
	for _, doc := range docs {
		var bref BiblioRef
		if err := json.Unmarshal([]byte(doc), &bref); err != nil {
			return nil, err
		}
		// On-the-fly add elasticsearch "_id" and indexed timestamp, if not already set.
		if bref.Key == "" && bref.SourceReleaseIdent != "" {
			bref.Key = fmt.Sprintf("%s_%d", bref.SourceReleaseIdent, bref.RefIndex)
			bref.IndexedTs = time.Now().UTC().Format(time.RFC3339)
		}
		brefs = append(brefs, &bref)
	}
	// Make sure exact matches come first. XXX: bug?
	sort.Slice(brefs, func(i, j int) bool {
		return brefs[i].MatchStatus == StatusExact.Short()
	})
	seen := set.New()
	for _, doc := range brefs {
		h := doc.LinkHash()
		if seen.Contains(h) {
			stats.skipDuplicatedBref++
			log.Printf("skip-dup-bref [%d]: hash=%v source=%v status=%v reason=%v",
				stats.skipDuplicatedBref, h, doc.SourceReleaseIdent, doc.MatchStatus, doc.MatchReason)
			continue
		}
		seen.Add(h)
		result = append(result, doc)
	}
	return result, nil
}

type statsAugment struct {
	skipDuplicatedBref int64
	skipMatchedRef     int64
	total              int64
}

func (s statsAugment) String() string {
	return fmt.Sprintf("total=%d, skipMatchedRef=%d, skipDuplicatedBref=%d",
		s.total, s.skipMatchedRef, s.skipDuplicatedBref)
}

// CutBatch runs Cut over a list of lines.
func CutBatch(lines []string, column int) (result []string) {
	for _, line := range lines {
		result = append(result, Cut(line, column))
	}
	return result
}

// Cut returns a specific column (1-indexed) from a line, returns empty string
// if column is invalid.
func Cut(line string, column int) string {
	return CutSep(line, "\t", column)
}

// CutSep allows to specify a separator, column is 1-indexed.
func CutSep(line, sep string, column int) string {
	parts := strings.Split(strings.TrimSpace(line), sep)
	if len(parts) < column {
		return ""
	} else {
		return parts[column-1]
	}
}

// FindByPrefix return the first element for a slice of strings, which matches a prefix.
func FindByPrefix(ss []string, prefix string) string {
	for _, s := range ss {
		if strings.HasPrefix(s, prefix) {
			return s
		}
	}
	return ""
}

// makeKeyFunc creates a function that can be used as keyFunc, selecting a
// column from fields separated by sep; column is 1-indexed.
func makeKeyFunc(sep string, column int) func(string) (string, error) {
	return func(s string) (string, error) {
		if k := CutSep(s, sep, column); k != "" {
			return k, nil
		}
		return "", fmt.Errorf("cannot get key from column %d in line (len=%d): %s", column, len(s), s)
	}
}

func parseRelease(s string) (r *Release, err error) {
	err = json.Unmarshal([]byte(s), &r)
	return
}

func parseRef(s string) (r *Ref, err error) {
	err = json.Unmarshal([]byte(s), &r)
	return
}

func parseWiki(s string) (r *MinimalCitations, err error) {
	err = json.Unmarshal([]byte(s), &r)
	return
}

func parseBiblioref(s string) (r *BiblioRef, err error) {
	err = json.Unmarshal([]byte(s), &r)
	return
}