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
|
# pylint: disable=C0103
"""
Clustering stage.
"""
import collections
import fileinput
import itertools
import json
import logging
import operator
import os
import re
import subprocess
import sys
import tempfile
import fuzzy
__all__ = [
"release_key_title",
"release_key_title_normalized",
"release_key_title_nysiis",
"sort_by_column",
"group_by",
"Cluster",
]
get_ident_title = operator.itemgetter("ident", "title")
ws_replacer = str.maketrans({"\t": " ", "\n": " "})
non_word_re = re.compile(r'[\W_]+', re.UNICODE)
# Notes: untie from release_entity, as we are only using a few fields. Maybe
# it's a jsob blob, with a pydantic spec and schema.
def release_key_title(release_entity, get_ident_title=get_ident_title):
id, title = get_ident_title(release_entity)
if not title:
raise ValueError('title missing')
title = title.translate(ws_replacer).strip()
return (id, title)
def release_key_title_normalized(release_entity):
id, title = release_key_title(release_entity)
title = re.sub(r'[ ]{2,}', ' ', title)
title = title.lower()
return (id, non_word_re.sub('', title))
def release_key_title_nysiis(release_entity):
id, title = release_key_title(release_entity)
return (id, fuzzy.nysiis(title))
def release_key_title_authors_ngram(release_entity):
"""
Derive a key from title and authors. Authors in contribs list:
"contribs": [
{
"index": 0,
"raw_name": "Meise Botanic Garden",
"role": "author"
}
],
"""
# SS: compare ngram sets?
def sort_by_column(filename, opts="-k 2", fast=True, mode="w", prefix="fuzzycat-", tmpdir=None):
"""
Sort tabular file with sort(1), returns the filename of the sorted file.
TODO: use separate /fast/tmp for sort.
"""
with tempfile.NamedTemporaryFile(delete=False, mode=mode, prefix=prefix) as tf:
env = os.environ.copy()
if tmpdir is not None:
env["TMPDIR"] = tmpdir
if fast:
env["LC_ALL"] = "C"
subprocess.run(["sort"] + opts.split() + [filename], stdout=tf, env=env, check=True)
return tf.name
def group_by(seq, key=None, value=None, comment=""):
"""
Iterate over lines in filename, group by key (a callable deriving the key
from the line), then apply value callable on the same value to emit a
minimal document, containing the key and identifiers belonging to a
cluster.
"""
for k, g in itertools.groupby(seq, key=key):
doc = {
"k": k.strip(),
"v": [value(v) for v in g],
}
if comment:
doc["c"] = comment
yield doc
def cut(f=0, sep='\t', ignore_missing_column=True):
"""
Return a callable, that extracts a given column from a file with a specific
separator. TODO: move this into more generic place.
"""
def func(value):
parts = value.strip().split(sep)
if f >= len(parts):
if ignore_missing_column:
return ""
raise ValueError('cannot split value {} into {} parts'.format(value, f))
return parts[f]
return func
class Cluster:
"""
Cluster scaffold for release entities. XXX: move IO/files out, allow any iterable.
"""
def __init__(self,
files="-",
output=sys.stdout,
keyfunc=lambda v: v,
prefix='fuzzycat-',
tmpdir=None):
"""
Files can be a list of files or "-" for stdin.
"""
self.files = files
self.keyfunc = keyfunc
self.output = output
self.prefix = prefix
self.tmpdir = tmpdir
self.logger = logging.getLogger('fuzzycat.cluster')
def run(self):
"""
Run clustering and write output to given stream or file.
"""
keyfunc = self.keyfunc # Save a lookup in loop.
counter = collections.Counter()
with tempfile.NamedTemporaryFile(delete=False, mode="w", prefix=self.prefix) as tf:
for line in fileinput.input(files=self.files):
try:
id, key = keyfunc(json.loads(line))
print("{}\t{}".format(id, key), file=tf)
except (KeyError, ValueError):
counter["key_extraction_failed"] += 1
else:
counter["key_ok"] += 1
sbc = sort_by_column(tf.name, opts='-k 2', prefix=self.prefix, tmpdir=self.tmpdir)
with open(sbc) as f:
comment = keyfunc.__name__
for doc in group_by(f, key=cut(f=1), value=cut(f=0), comment=comment):
counter["groups"] += 1
json.dump(doc, self.output)
self.output.write("\n")
os.remove(sbc)
os.remove(tf.name)
return counter
|
