team-qbits / training /normalize_subjects.py

Upload folder using huggingface_hub

e55c8a4 verified 3 days ago

8.49 kB

#!/usr/bin/env python
"""
Make the `subject` field consistent across the dataset.
Why it matters: build-sft puts the subject INTO the prompt ("විෂය: {subject}" / "Subject: {subject}"),
and the organisers' prompt always supplies the SINHALA subject name (metadata.subject_original, e.g.
"චිත්‍ර" for Arts). If some training rows carry an English or short name ("Dancing", "සිංහල") the model
sees a label at train time that never appears at test time, for exactly those subjects.
Three steps:
  1. see what the official test files actually use (the authoritative names):
         python normalize_subjects.py --test_dir TEST/easy
  2. see what your dataset currently uses:
         python normalize_subjects.py --inp data/all_converted.jsonl --list
  3. rewrite with the mapping below (edit MAPPING first if step 1 disagrees):
         python normalize_subjects.py --inp data/all_converted.jsonl --out data/all_converted_norm.jsonl
Nothing else in the record is touched. Use --map_file to pass your own {"from": "to"} JSON instead.
"""
import argparse
import collections
import glob
import json
import os
import sys

# English / short / variant name  ->  the SINHALA name the official test files use as
# metadata.subject_original. These were read off TEST/easy, not guessed.
MAPPING = {
    # --- corrections to names already in the dataset -------------------------------------
    "සිංහල භාෂාව හා සාහිත්‍යය": "සිංහල භාෂාව සහ සාහිත්‍යය",   # හා -> සහ
    "සෞඛ්‍ය හා ශාරීරික අධ්‍යාපනය": "සෞඛ්‍ය හා ශාරීරික විද්‍යාව",  # අධ්‍යාපනය -> විද්‍යාව
    "නැටුම්": "නර්තනය",
    "සිංහල": "සිංහල භාෂාව සහ සාහිත්‍යය",
    # --- English / variant labels from newer extractions ---------------------------------
    "Arts": "චිත්‍ර",
    "Art": "චිත්‍ර",
    "art": "චිත්‍ර",
    "Dancing": "නර්තනය",
    "dancing": "නර්තනය",
    "Buddhism": "බුද්ධ ධර්මය",
    "Sinhala": "සිංහල භාෂාව සහ සාහිත්‍යය",
    "Sinhala language and literature": "සිංහල භාෂාව සහ සාහිත්‍යය",
    "Sinhala_language_and_literature": "සිංහල භාෂාව සහ සාහිත්‍යය",
    "Eastern music": "පෙරදිග සංගීතය",
    "Eastern_music": "පෙරදිග සංගීතය",
    "drama and Theatre": "නාට්‍ය හා රංග කලාව",
    "drama_and_theatre": "නාට්‍ය හා රංග කලාව",
    "drama and theatre": "නාට්‍ය හා රංග කලාව",
    "drama": "නාට්‍ය හා රංග කලාව",
    "drama and theater": "නාට්‍ය හා රංග කලාව",
    "theatre and drama": "නාට්‍ය හා රංග කලාව",
    "eastern music": "පෙරදිග සංගීතය",
    "Health and Physical Science": "සෞඛ්‍ය හා ශාරීරික විද්‍යාව",
    "History": "ඉතිහාසය",
    "Civics": "පුරවැසි අධ්‍යාපනය",
    "Geography": "භූගෝල විද්‍යාව",
    "Catholicism": "කතෝලික ධර්මය",
    "Christianity": "ක්‍රිස්තියානි ධර්මය",
    "Islam": "ඉස්ලාම් ධර්මය",
    "science": "විද්‍යාව",
    "Science": "විද්‍යාව",
}


def load_map(path):
    """Read a {"from": "to"} JSON file, tolerating a UTF-8 BOM or UTF-16 (what Windows editors write)."""
    raw = open(path, "rb").read()
    if not raw.strip():
        sys.exit(f"{path} is empty")
    for enc in ("utf-8-sig", "utf-16", "utf-8"):
        try:
            txt = raw.decode(enc)
        except (UnicodeDecodeError, UnicodeError):
            continue
        try:
            m = json.loads(txt.lstrip("\ufeff"))
        except json.JSONDecodeError:
            continue
        if not isinstance(m, dict):
            sys.exit(f"{path} must contain a JSON object of {{\"from\": \"to\"}} pairs")
        print(f"[map] {len(m)} override(s) read from {path} ({enc})")
        return {str(k): str(v) for k, v in m.items()}
    sys.exit(f"could not parse {path} as JSON in utf-8/utf-8-sig/utf-16. First bytes: {raw[:40]!r}")


def read_jsonl(path):
    rows = []
    with open(path, encoding="utf-8-sig") as f:
        for line in f:
            line = line.strip()
            if line:
                rows.append(json.loads(line))
    return rows


def show_test_names(test_dir):
    """Print the subject / subject_original pairs the official test files use."""
    files = sorted(glob.glob(os.path.join(test_dir, "**", "*.json"), recursive=True))
    if not files:
        sys.exit(f"no .json files under {test_dir}")
    c = collections.Counter()
    for p in files:
        try:
            data = json.load(open(p, encoding="utf-8"))
        except Exception as e:                                           # noqa: BLE001
            print(f"[warn] {os.path.basename(p)}: {type(e).__name__}")
            continue
        if isinstance(data, list):
            for it in data:
                md = it.get("metadata") or {}
                c[(it.get("subject"), md.get("subject_original"))] += 1
    print(f"official test subjects in {test_dir} "
          f"(subject_original is what goes into THEIR prompt):")
    print(f"{'n':>6}  {'subject':<34}{'subject_original'}")
    for (s, so), n in sorted(c.items(), key=lambda x: -x[1]):
        print(f"{n:>6}  {str(s):<34}{so}")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--inp", default="")
    ap.add_argument("--out", default="")
    ap.add_argument("--list", action="store_true", help="just print the current subject counts")
    ap.add_argument("--test_dir", default="", help="print the official test files' subject names")
    ap.add_argument("--map_file", default="", help='JSON file of {"from": "to"} overrides')
    args = ap.parse_args()
    for stream in (sys.stdout, sys.stderr):
        try:
            stream.reconfigure(encoding="utf-8")
        except Exception:                                                # noqa: BLE001
            pass

    if args.test_dir:
        show_test_names(args.test_dir)
        if not args.inp:
            return
        print()

    if not args.inp:
        sys.exit("give --inp (and --out, unless --list)")
    rows = read_jsonl(args.inp)
    before = collections.Counter(r.get("subject") or "?" for r in rows)

    if args.list or not args.out:
        print(f"{len(rows)} rows in {args.inp}; subjects:")
        for s, n in before.most_common():
            flag = "  <-- would be renamed" if s in MAPPING else ""
            print(f"{n:>6}  {s}{flag}")
        if not args.out:
            return

    mapping = dict(MAPPING)
    if args.map_file:
        mapping.update(load_map(args.map_file))

    ci = {k.lower(): v for k, v in mapping.items()}
    changed = collections.Counter()
    for r in rows:
        s = r.get("subject")
        # normalise stray whitespace too: the official files contain "drama and Theatre " with a
        # trailing space, which would otherwise be a separate bucket
        if isinstance(s, str):
            s2 = " ".join(s.split())
            if s2 != s:
                r["subject"] = s = s2
        tgt = mapping.get(s)
        if tgt is None and isinstance(s, str):
            tgt = ci.get(s.lower())          # Sinhala has no case, so lowering is safe
        if tgt and tgt != s:
            changed[(s, tgt)] += 1
            r["subject"] = tgt

    with open(args.out, "w", encoding="utf-8") as f:
        for r in rows:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")

    if changed:
        print("renamed:")
        for (a, b), n in changed.most_common():
            print(f"{n:>6}  {a!r} -> {b!r}")
    else:
        print("nothing to rename (subjects already consistent)")
    after = collections.Counter(r.get("subject") or "?" for r in rows)
    print(f"\n[out] {len(rows)} rows -> {args.out}")
    print(f"distinct subjects: {len(before)} -> {len(after)}")


if __name__ == "__main__":
    main()

	#!/usr/bin/env python
	"""
	Make the `subject` field consistent across the dataset.

	Why it matters: build-sft puts the subject INTO the prompt ("විෂය: {subject}" / "Subject: {subject}"),
	and the organisers' prompt always supplies the SINHALA subject name (metadata.subject_original, e.g.
	"චිත්‍ර" for Arts). If some training rows carry an English or short name ("Dancing", "සිංහල") the model
	sees a label at train time that never appears at test time, for exactly those subjects.

	Three steps:

	1. see what the official test files actually use (the authoritative names):
	python normalize_subjects.py --test_dir TEST/easy
	2. see what your dataset currently uses:
	python normalize_subjects.py --inp data/all_converted.jsonl --list
	3. rewrite with the mapping below (edit MAPPING first if step 1 disagrees):
	python normalize_subjects.py --inp data/all_converted.jsonl --out data/all_converted_norm.jsonl

	Nothing else in the record is touched. Use --map_file to pass your own {"from": "to"} JSON instead.
	"""
	import argparse
	import collections
	import glob
	import json
	import os
	import sys

	# English / short / variant name -> the SINHALA name the official test files use as
	# metadata.subject_original. These were read off TEST/easy, not guessed.
	MAPPING = {
	# --- corrections to names already in the dataset -------------------------------------
	"සිංහල භාෂාව හා සාහිත්‍යය": "සිංහල භාෂාව සහ සාහිත්‍යය", # හා -> සහ
	"සෞඛ්‍ය හා ශාරීරික අධ්‍යාපනය": "සෞඛ්‍ය හා ශාරීරික විද්‍යාව", # අධ්‍යාපනය -> විද්‍යාව
	"නැටුම්": "නර්තනය",
	"සිංහල": "සිංහල භාෂාව සහ සාහිත්‍යය",
	# --- English / variant labels from newer extractions ---------------------------------
	"Arts": "චිත්‍ර",
	"Art": "චිත්‍ර",
	"art": "චිත්‍ර",
	"Dancing": "නර්තනය",
	"dancing": "නර්තනය",
	"Buddhism": "බුද්ධ ධර්මය",
	"Sinhala": "සිංහල භාෂාව සහ සාහිත්‍යය",
	"Sinhala language and literature": "සිංහල භාෂාව සහ සාහිත්‍යය",
	"Sinhala_language_and_literature": "සිංහල භාෂාව සහ සාහිත්‍යය",
	"Eastern music": "පෙරදිග සංගීතය",
	"Eastern_music": "පෙරදිග සංගීතය",
	"drama and Theatre": "නාට්‍ය හා රංග කලාව",
	"drama_and_theatre": "නාට්‍ය හා රංග කලාව",
	"drama and theatre": "නාට්‍ය හා රංග කලාව",
	"drama": "නාට්‍ය හා රංග කලාව",
	"drama and theater": "නාට්‍ය හා රංග කලාව",
	"theatre and drama": "නාට්‍ය හා රංග කලාව",
	"eastern music": "පෙරදිග සංගීතය",
	"Health and Physical Science": "සෞඛ්‍ය හා ශාරීරික විද්‍යාව",
	"History": "ඉතිහාසය",
	"Civics": "පුරවැසි අධ්‍යාපනය",
	"Geography": "භූගෝල විද්‍යාව",
	"Catholicism": "කතෝලික ධර්මය",
	"Christianity": "ක්‍රිස්තියානි ධර්මය",
	"Islam": "ඉස්ලාම් ධර්මය",
	"science": "විද්‍යාව",
	"Science": "විද්‍යාව",
	}


	def load_map(path):
	"""Read a {"from": "to"} JSON file, tolerating a UTF-8 BOM or UTF-16 (what Windows editors write)."""
	raw = open(path, "rb").read()
	if not raw.strip():
	sys.exit(f"{path} is empty")
	for enc in ("utf-8-sig", "utf-16", "utf-8"):
	try:
	txt = raw.decode(enc)
	except (UnicodeDecodeError, UnicodeError):
	continue
	try:
	m = json.loads(txt.lstrip("\ufeff"))
	except json.JSONDecodeError:
	continue
	if not isinstance(m, dict):
	sys.exit(f"{path} must contain a JSON object of {{\"from\": \"to\"}} pairs")
	print(f"[map] {len(m)} override(s) read from {path} ({enc})")
	return {str(k): str(v) for k, v in m.items()}
	sys.exit(f"could not parse {path} as JSON in utf-8/utf-8-sig/utf-16. First bytes: {raw[:40]!r}")


	def read_jsonl(path):
	rows = []
	with open(path, encoding="utf-8-sig") as f:
	for line in f:
	line = line.strip()
	if line:
	rows.append(json.loads(line))
	return rows


	def show_test_names(test_dir):
	"""Print the subject / subject_original pairs the official test files use."""
	files = sorted(glob.glob(os.path.join(test_dir, "*", ".json"), recursive=True))
	if not files:
	sys.exit(f"no .json files under {test_dir}")
	c = collections.Counter()
	for p in files:
	try:
	data = json.load(open(p, encoding="utf-8"))
	except Exception as e: # noqa: BLE001
	print(f"[warn] {os.path.basename(p)}: {type(e).__name__}")
	continue
	if isinstance(data, list):
	for it in data:
	md = it.get("metadata") or {}
	c[(it.get("subject"), md.get("subject_original"))] += 1
	print(f"official test subjects in {test_dir} "
	f"(subject_original is what goes into THEIR prompt):")
	print(f"{'n':>6} {'subject':<34}{'subject_original'}")
	for (s, so), n in sorted(c.items(), key=lambda x: -x[1]):
	print(f"{n:>6} {str(s):<34}{so}")


	def main():
	ap = argparse.ArgumentParser()
	ap.add_argument("--inp", default="")
	ap.add_argument("--out", default="")
	ap.add_argument("--list", action="store_true", help="just print the current subject counts")
	ap.add_argument("--test_dir", default="", help="print the official test files' subject names")
	ap.add_argument("--map_file", default="", help='JSON file of {"from": "to"} overrides')
	args = ap.parse_args()
	for stream in (sys.stdout, sys.stderr):
	try:
	stream.reconfigure(encoding="utf-8")
	except Exception: # noqa: BLE001
	pass

	if args.test_dir:
	show_test_names(args.test_dir)
	if not args.inp:
	return
	print()

	if not args.inp:
	sys.exit("give --inp (and --out, unless --list)")
	rows = read_jsonl(args.inp)
	before = collections.Counter(r.get("subject") or "?" for r in rows)

	if args.list or not args.out:
	print(f"{len(rows)} rows in {args.inp}; subjects:")
	for s, n in before.most_common():
	flag = " <-- would be renamed" if s in MAPPING else ""
	print(f"{n:>6} {s}{flag}")
	if not args.out:
	return

	mapping = dict(MAPPING)
	if args.map_file:
	mapping.update(load_map(args.map_file))

	ci = {k.lower(): v for k, v in mapping.items()}
	changed = collections.Counter()
	for r in rows:
	s = r.get("subject")
	# normalise stray whitespace too: the official files contain "drama and Theatre " with a
	# trailing space, which would otherwise be a separate bucket
	if isinstance(s, str):
	s2 = " ".join(s.split())
	if s2 != s:
	r["subject"] = s = s2
	tgt = mapping.get(s)
	if tgt is None and isinstance(s, str):
	tgt = ci.get(s.lower()) # Sinhala has no case, so lowering is safe
	if tgt and tgt != s:
	changed[(s, tgt)] += 1
	r["subject"] = tgt

	with open(args.out, "w", encoding="utf-8") as f:
	for r in rows:
	f.write(json.dumps(r, ensure_ascii=False) + "\n")

	if changed:
	print("renamed:")
	for (a, b), n in changed.most_common():
	print(f"{n:>6} {a!r} -> {b!r}")
	else:
	print("nothing to rename (subjects already consistent)")
	after = collections.Counter(r.get("subject") or "?" for r in rows)
	print(f"\n[out] {len(rows)} rows -> {args.out}")
	print(f"distinct subjects: {len(before)} -> {len(after)}")


	if __name__ == "__main__":
	main()