File size: 16,981 Bytes
c83c617
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Full storage/span/split/codec audit, plus a stratified machine-review packet."""
import argparse
from collections import Counter, defaultdict
import hashlib
import json
from pathlib import Path
import random
import pickle
import struct
import sys
import time

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "src"))
import numpy as np
from vimeml.posttraining.data import ConversionCodec, ConversionDataset, shifted
from vimeml.posttraining.readings import DualReading, HIRAGANA
from vimeml.training.data import file_sha, write_json


def quarantine_known_ambiguity(directory):
    """Never repair an inferred reading. Remove known shared dictionary errors.

    Standalone 家 tagged ヤ/ケ can be a family suffix, but also mislabels the
    ordinary house noun in unpunctuated web sentences. Quarantine both rather
    than guessing from a pseudo-label. Compounds such as 家族 and readings カ
    are unaffected. No benchmark observations determine this train-review rule.
    """
    manifest = json.loads((directory / "manifest.json").read_text(encoding="utf-8"))
    if manifest.get("quality_refinement", {}).get("version") == 2:
        return
    if not (directory / "annotation-manifest.json").exists():
        write_json(directory / "annotation-manifest.json", manifest)
    prior_refinement = manifest.get("quality_refinement")
    previous_manifest_sha = file_sha(directory / "manifest.json")
    if prior_refinement:
        write_json(directory / "refinement-v1-manifest.json", manifest)
        for split in ("train", "validation"):
            previous = directory / f"{split}.quality-quarantine.jsonl"
            previous.rename(directory / f"{split}.quality-v1-quarantine.jsonl")
    validation_data = ConversionDataset(directory, "validation")
    validation_sequences = {hashlib.sha256(validation_data.ids[int(validation_data.offsets[i]):int(validation_data.offsets[i + 1])].tobytes()).digest()
                            for i in range(len(validation_data))}
    validation_data.close()
    with (directory / "validation.records.jsonl").open(encoding="utf-8") as records:
        validation_cases = {json.loads(raw)["case_sha256"] for raw in records}
    reader = DualReading()
    excluded = {}
    retained_kanji = {}
    for split in ("train", "validation"):
        bad = {}
        kanji = 0
        data = ConversionDataset(directory, split)
        with (ROOT / f"outputs/corpus-fast-v1/{split}.txt").open("rb") as original, \
             (directory / f"{split}.records.jsonl").open(encoding="utf-8") as records:
            for index, raw in enumerate(records):
                ex = json.loads(raw)
                if split == "train":
                    sequence = data.ids[int(data.offsets[index]):int(data.offsets[index + 1])].tobytes()
                    if ex["case_sha256"] in validation_cases or hashlib.sha256(sequence).digest() in validation_sequences:
                        bad[index] = "exact_conversion_validation_case"
                        continue
                ambiguous = False
                if "家" in ex["output"]:
                    original.seek(ex["byte_offset"])
                    text = original.readline().decode("utf-8").rstrip("\r\n")
                    cursor = 0
                    for word in reader.mecab(text):
                        start = cursor + len(word.white_space)
                        end = start + len(word.surface)
                        cursor = end
                        if ex["span"][0] <= start and end <= ex["span"][1] and word.surface == "家" and word.feature.kana in ("ヤ", "ケ"):
                            ambiguous = True
                            break
                if ambiguous:
                    bad[index] = "standalone_house_family_suffix_ambiguity"
                else:
                    kanji += ex["has_kanji"]
        excluded[split], retained_kanji[split] = bad, kanji
        data.close()
    staging = directory / "quality-staging"
    staging.mkdir(exist_ok=True)
    new_stats = {}
    removed_counts = {}
    for split in ("train", "validation"):
        data = ConversionDataset(directory, split)
        stats, hist = Counter(), defaultdict(Counter)
        kana_limit = retained_kanji[split] // 4
        with (directory / f"{split}.records.jsonl").open(encoding="utf-8") as source, \
             (staging / f"{split}.records.jsonl").open("w", encoding="utf-8", newline="\n") as records, \
             (staging / f"{split}.tokens.bin").open("wb") as tokens, \
             (staging / f"{split}.offsets.bin").open("wb") as offsets, \
             (staging / f"{split}.out-start.bin").open("wb") as starts, \
             (staging / f"{split}.reading-dropout.bin").open("wb") as dropout, \
             (directory / f"{split}.quality-quarantine.jsonl").open("w", encoding="utf-8", newline="\n") as rejected:
            offsets.write(struct.pack("<Q", 0))
            for index, raw in enumerate(source):
                ex = json.loads(raw)
                reason = excluded[split].get(index)
                if reason is None and not ex["has_kanji"] and stats["kana_only"] >= kana_limit:
                    reason = "retain_kana_only_cap_after_quarantine"
                if reason:
                    rejected.write(json.dumps({"reason": reason, **ex}, ensure_ascii=False) + "\n")
                    stats["excluded_" + reason] += 1
                    continue
                ids = data.ids[int(data.offsets[index]):int(data.offsets[index + 1])].tolist()
                records.write(raw)
                tokens.write(struct.pack(f"<{len(ids)}H", *ids))
                starts.write(bytes([ex["first_output"]]))
                dropout.write(bytes([ex["reading_dropout_view"]]))
                stats["examples"] += 1
                stats["stored_tokens"] += len(ids)
                stats["supervised_tokens_full_reading"] += len(ids) - ex["first_output"]
                stats["context_cropped"] += bool(ex["context_crop_characters"])
                stats["kana_only"] += not ex["has_kanji"]
                stats["reading_dropout_view"] += ex["reading_dropout_view"]
                offsets.write(struct.pack("<Q", stats["stored_tokens"]))
                for name, value in (("reading_characters", len(ex["reading"])), ("output_characters", len(ex["output"])),
                                    ("context_characters", len(ex["context"])), ("source_id", ex["source_id"]),
                                    ("serialized_tokens", len(ids))):
                    hist[name][value] += 1
        data.close()
        new_stats[split] = {**manifest["splits"][split], **dict(stats), "histograms": dict(hist)}
        for key in list(new_stats[split]):
            if key.startswith("excluded_"):
                new_stats[split][key] = manifest["splits"][split].get(key, 0) + stats.get(key, 0)
        removed_counts[split] = {k: v for k, v in stats.items() if k.startswith("excluded_")}
    write_json(directory / "manifest.json", {**manifest, "status": "refining"})
    for path in staging.iterdir():
        path.replace(directory / path.name)
    manifest["splits"] = new_stats
    if prior_refinement:
        manifest.setdefault("quality_refinement_history", []).append(prior_refinement)
    manifest["quality_refinement"] = {"version": 2, "rule": "exclude exact frozen conversion-validation field tuples or serialized sequences; quarantine standalone 家 with dictionary reading ヤ/ケ; kana-only <=20%; no reading repair or replacement samples",
        "origin": "machine review of original train row14, parent000006e2bfe7819a400b082f1815355e14e445d5d7d8cec234ee855a4b3831b9",
        "counts": removed_counts, "annotation_manifest_sha256": file_sha(directory / "annotation-manifest.json"),
        "previous_manifest_sha256": previous_manifest_sha,
        "code_sha256": file_sha(Path(__file__))}
    manifest["annotation_signature"] = manifest["signature"]
    manifest["signature"] = hashlib.sha256(json.dumps({"annotation": manifest["annotation_signature"],
        "refinement": manifest["quality_refinement"]}, sort_keys=True).encode("utf-8")).hexdigest()
    # The original uniform review packet can contain a quarantined example; retain
    # it as annotation history, and use the new stratified packet for final review.
    for split in ("train", "validation"):
        old_review = directory / f"{split}.review-sample.json"
        if old_review.exists():
            old_review.rename(directory / f"{split}.annotation-review-sample.json")
    manifest["output_sha256"] = {p.name: file_sha(p) for p in sorted(directory.iterdir())
        if p.is_file() and (p.name.endswith(".bin") or p.name.endswith(".records.jsonl") or ".quality" in p.name and p.name.endswith("quarantine.jsonl"))}
    write_json(directory / "manifest.json", manifest)
    print(json.dumps({"quality_refinement": removed_counts, "retained": {s: x["examples"] for s, x in new_stats.items()}}), flush=True)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--directory", type=Path, default=ROOT / "artifacts/posttrain-data/v4-c1")
    args = parser.parse_args()
    directory = args.directory
    quarantine_known_ambiguity(directory)
    manifest = json.loads((directory / "manifest.json").read_text(encoding="utf-8"))
    began = time.monotonic()
    import unidic_lite
    import sudachipy
    dictionary_resources = list(Path(unidic_lite.DICDIR).iterdir())
    dictionary_resources += list((Path(sudachipy.__file__).parent / "resources").iterdir())
    resource_sha = {str(p): file_sha(p) for p in sorted(dictionary_resources)
                    if p.is_file() and p.suffix != ".pdf"}
    for name, expected in manifest["output_sha256"].items():
        if file_sha(directory / name) != expected:
            raise ValueError(f"Damaged conversion file: {name}")
    validation_parents = {hashlib.sha256(x.rstrip(b"\r\n")).hexdigest()
                          for x in (ROOT / "outputs/corpus-fast-v1/validation.txt").open("rb")}
    codec = ConversionCodec(ROOT / "artifacts/tokenizers/v4-a/ja-unigram-8192/tokenizer.model")
    validation_data = ConversionDataset(directory, "validation")
    validation_sequences = {hashlib.sha256(validation_data.ids[int(validation_data.offsets[i]):int(validation_data.offsets[i + 1])].tobytes()).digest()
                            for i in range(len(validation_data))}
    validation_data.close()
    with (directory / "validation.records.jsonl").open(encoding="utf-8") as records:
        validation_cases = {json.loads(raw)["case_sha256"] for raw in records}
    audit, review_buckets, seen = {}, defaultdict(list), Counter()
    for split in ("train", "validation"):
        data = ConversionDataset(directory, split)
        reopened = pickle.loads(pickle.dumps(data))
        assert reopened[0] == data[0]
        reopened.close()
        sources = np.memmap(ROOT / f"artifacts/token-data/corpus-v4-8k/{split}.sources.bin", mode="r", dtype="u1")
        rows = np.memmap(ROOT / f"artifacts/token-data/corpus-v4-8k/{split}.rows.bin", mode="r", dtype="<u8")
        counts = Counter()
        rng = random.Random(42)
        with (ROOT / f"outputs/corpus-fast-v1/{split}.txt").open("rb") as original, \
             (directory / f"{split}.records.jsonl").open(encoding="utf-8") as records:
            for index, raw in enumerate(records):
                ex = json.loads(raw)
                original.seek(ex["byte_offset"])
                text = original.readline().decode("utf-8").rstrip("\r\n")
                parent = hashlib.sha256(text.encode("utf-8")).hexdigest()
                start, end = ex["span"]
                assert parent == ex["parent_sha256"]
                assert text[:start] == ex["context"] and text[start:end] == ex["output"]
                assert int(rows[ex["row"]]) == ex["byte_offset"] and int(sources[ex["row"]]) == ex["source_id"]
                assert split != "train" or parent not in validation_parents
                assert ex["mecab_reading"] == ex["sudachi_reading"] == ex["reading"]
                assert HIRAGANA.fullmatch(ex["reading"])
                encoded, first, removed = codec.encode(ex["context"], ex["reading"], ex["output"])
                stored = data.ids[int(data.offsets[index]):int(data.offsets[index + 1])].tolist()
                assert stored == encoded and int(data.first[index]) == first
                if split == "train":
                    assert ex["case_sha256"] not in validation_cases
                    assert hashlib.sha256(struct.pack(f"<{len(stored)}H", *stored)).digest() not in validation_sequences
                assert removed == ex["context_crop_characters"]
                view = data[index]
                expected_ids, expected_first = encoded, first
                if ex["reading_dropout_view"]:
                    left, right = encoded.index(5) + 1, first - 1
                    expected_ids = encoded[:left] + encoded[right:]
                    expected_first = first - right + left
                assert view == shifted(expected_ids, expected_first)
                assert view["labels"][expected_first - 1:] == codec.sp.encode(ex["output"]) + [3]
                assert len(stored) <= 129 and stored.count(4) == stored.count(5) == stored.count(6) == 1
                counts["examples"] += 1
                counts["supervised_tokens"] += sum(x != -100 for x in view["labels"])
                counts["view_input_tokens"] += len(view["input_ids"])
                counts["reading_dropout"] += bool(ex["reading_dropout_view"])
                counts["no_left_context"] += not ex["context"]
                if split == "train":
                    key = tuple(ex["stratum"])
                    seen[key] += 1
                    bucket = review_buckets[key]
                    if len(bucket) < 10:
                        bucket.append(ex)
                    else:
                        k = rng.randrange(seen[key])
                        if k < 10:
                            bucket[k] = ex
                if (index + 1) % 250000 == 0:
                    print(f"{split}: audited {index + 1}", flush=True)
        assert counts["examples"] == len(data)
        assert counts["supervised_tokens"] == manifest["splits"][split]["supervised_tokens_full_reading"]
        audit[split] = dict(counts)
        data.close()
    # Round-robin occupied strata, followed by independent fresh annotation for review packet.
    packet = []
    for j in range(10):
        for key in sorted(review_buckets):
            if j < len(review_buckets[key]) and len(packet) < 200:
                packet.append(review_buckets[key][j])
    analyzer = DualReading()
    active_mecab_dictionaries = analyzer.mecab.dictionary_info
    frozen_binaries = manifest["contract"]["dictionary"]["binaries"]
    for info in active_mecab_dictionaries:
        if file_sha(info["filename"]) != frozen_binaries.get(info["filename"]):
            raise ValueError("Active MeCab dictionary differs from the frozen identity")
    with (ROOT / "outputs/corpus-fast-v1/train.txt").open("rb") as original:
        for ex in packet:
            original.seek(ex["byte_offset"])
            text = original.readline().decode("utf-8").rstrip("\r\n")
            regenerated, reason = analyzer.example(text, int(ex["parent_sha256"][16:32], 16))
            assert reason is None
            assert all(regenerated[k] == ex[k] for k in ("reading", "output", "span", "context"))
            ex["full_original_sentence"] = text
            ex["reviewer"] = "automated_reannotation; linguistic human review pending"
    write_json(directory / "stratified-review-200.json", packet)
    write_json(directory / "acceptance.json", {"status": "pass", "manifest_sha256": file_sha(directory / "manifest.json"),
               "checker_sha256": file_sha(Path(__file__)), "dictionary_supporting_resources_sha256": resource_sha,
               "reader_code_sha256": file_sha(ROOT / "src/vimeml/posttraining/data.py"),
               "active_mecab_dictionaries": active_mecab_dictionaries,
               "splits": audit, "review_examples": len(packet), "review_strata": len(review_buckets),
               "checks": ["all output file SHA256", "every original TXT row/source/span/hash", "train vs all original validation sentence hashes",
                          "no exact conversion case or serialized sequence overlap with frozen conversion validation",
                          "every structured sequence reencoded", "every shifted loss mask and dropout view", "200 stratified fresh double annotations"],
               "linguistic_review": "machine annotation only; agreement does not establish gold correctness",
               "test_or_blind_read": False, "elapsed_seconds": time.monotonic() - began})
    print(json.dumps({"status": "pass", "splits": audit, "seconds": time.monotonic() - began}), flush=True)


if __name__ == "__main__":
    main()