A script to join the GitHub's grade CSV with the student list of Kyoto U.

join_github_csv.py

#!/usr/bin/env python3
#######################################################
# Name
#  join_github_csv.py
# Description
#  A script to join the GitHub's grade CSV with the student list of Kyoto U.
# Usage
#  python3 join_github_csv.py kyoto-u.csv github.csv merged.csv ; cut -d , -f 1,2,4,5,33 merged.csv > cleaned.csv
# Author
#  Masaki Waga
# License
#  MIT License
#######################################################


import argparse
import csv
import sys
from collections import defaultdict


def normalize_github(value: str) -> str:
    """
    Normalize GitHub identifiers so that these match:
      maswag
      @maswag
      https://github.com/maswag
      github.com/maswag/
    """
    if value is None:
        return ""

    v = value.strip().strip('"').strip()
    if not v:
        return ""

    if v.startswith("@"):
        v = v[1:]

    lower = v.lower()

    for prefix in [
        "https://github.com/",
        "http://github.com/",
        "github.com/",
    ]:
        if lower.startswith(prefix):
            v = v[len(prefix):]
            break

    v = v.split("?")[0].split("#")[0].strip("/")

    # If someone wrote github.com/user/repo, use only user.
    if "/" in v:
        v = v.split("/", 1)[0]

    return v.lower()


def read_added_csv(path: str, keep_right_key: bool):
    with open(path, newline="", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)

        if reader.fieldnames is None:
            raise ValueError(f"{path}: empty CSV")

        if "github_username" not in reader.fieldnames:
            raise ValueError(f"{path}: no 'github_username' column found")

        if keep_right_key:
            added_columns = reader.fieldnames
        else:
            added_columns = [
                col for col in reader.fieldnames
                if col != "github_username"
            ]

        rows_by_github = defaultdict(list)

        for row in reader:
            key = normalize_github(row.get("github_username", ""))
            if key:
                rows_by_github[key].append(row)

        return added_columns, rows_by_github


def join_csv(
    original_path: str,
    added_path: str,
    output_path: str,
    duplicate_policy: str,
    keep_right_key: bool,
):
    added_columns, rows_by_github = read_added_csv(
        added_path,
        keep_right_key=keep_right_key,
    )

    unmatched_original = []
    matched_count = 0

    with open(original_path, newline="", encoding="utf-8-sig") as fin, \
         open(output_path, "w", newline="", encoding="utf-8-sig") as fout:

        reader = csv.reader(fin)
        writer = csv.writer(fout)

        try:
            original_header = next(reader)
        except StopIteration:
            raise ValueError(f"{original_path}: empty CSV")

        try:
            github_index = original_header.index("github")
        except ValueError:
            raise ValueError(f"{original_path}: no 'github' column found")

        writer.writerow(original_header + added_columns)

        for row_number, original_row in enumerate(reader, start=2):
            # Make sure short rows still have a github column.
            if len(original_row) <= github_index:
                original_row += [""] * (github_index + 1 - len(original_row))

            github_key = normalize_github(original_row[github_index])
            matches = rows_by_github.get(github_key, [])

            if not matches:
                writer.writerow(original_row + [""] * len(added_columns))
                unmatched_original.append((row_number, original_row[github_index]))
                continue

            if duplicate_policy == "error" and len(matches) > 1:
                raise ValueError(
                    f"Multiple rows in added CSV match github={original_row[github_index]!r} "
                    f"at original CSV line {row_number}"
                )

            if duplicate_policy == "first":
                matches = matches[:1]

            # For duplicate_policy == "all", the original row is repeated once
            # for each matching row in the added CSV. The original order is still preserved.
            for added_row in matches:
                writer.writerow(
                    original_row + [added_row.get(col, "") for col in added_columns]
                )
                matched_count += 1

    print(f"Written: {output_path}", file=sys.stderr)
    print(f"Matched rows: {matched_count}", file=sys.stderr)
    print(f"Unmatched original rows: {len(unmatched_original)}", file=sys.stderr)

    if unmatched_original:
        print("Unmatched examples:", file=sys.stderr)
        for row_number, github in unmatched_original[:10]:
            print(f"  line {row_number}: github={github!r}", file=sys.stderr)


def main():
    parser = argparse.ArgumentParser(
        description="Join two CSV files by original.github == added.github_username."
    )
    parser.add_argument("original_csv")
    parser.add_argument("added_csv")
    parser.add_argument("output_csv")

    parser.add_argument(
        "--duplicate-policy",
        choices=["first", "all", "error"],
        default="first",
        help=(
            "What to do if added_csv has multiple rows with the same github_username. "
            "Default: first"
        ),
    )

    parser.add_argument(
        "--keep-right-key",
        action="store_true",
        help="Also copy github_username from the added CSV into the output.",
    )

    args = parser.parse_args()

    join_csv(
        original_path=args.original_csv,
        added_path=args.added_csv,
        output_path=args.output_csv,
        duplicate_policy=args.duplicate_policy,
        keep_right_key=args.keep_right_key,
    )


if __name__ == "__main__":
    main()