Pontoon: Scripts to extract CHS data from spreadsheets, and load it to DB

extract_chs_backfill.py

"""
Extract Community Health Score (CHS) data from the historical monthly
spreadsheets so it can be backfilled into Pontoon's LocaleHealthSnapshot table.

Each spreadsheet is named YYYY-MM.xlsx and holds a "Summary" sheet with one
row per locale. The score for a month is recorded on the 1st of the *following*
month (matching how the automated snapshots work), so 2025-01.xlsx maps to a
created_at of 2025-02-01.

This script only reads the raw inputs needed to recompute the score in Django:
the loader (load_chs_backfill.py) runs them through pontoon.insights.chs.compute_chs.
The sheet's own total is carried along as "sheet_chs" for verification only.

Usage:
    python3 extract_chs_backfill.py

Writes chs_backfill.json next to this script.
"""

import glob
import json
import os

from datetime import date

import openpyxl


# Directory holding the YYYY-MM.xlsx spreadsheets.
SHEETS_DIR = os.path.expanduser("PATH_TO_SHEETS_DIR")  # TODO: set this to the actual path
OUTPUT_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "chs_backfill.json")

# Column indices in the "Summary" sheet (0-based).
COL_LOCALE = 0
COL_MANAGERS = 1
COL_TRANSLATORS = 2
COL_ACTIVE_CONTRIBUTORS = 3
COL_ALL_CONTRIBUTORS = 4
COL_NEW_SIGNUPS = 5
COL_COMPLETION = 6  # ratio 0-1; also used to identify data rows
COL_ENABLED_PROJECTS_SCORE = 13  # "#P" weighted score
COL_SHEET_TOTAL = 15  # "Total" CHS reported in the sheet


def next_month_first_day(filename):
    """2025-01.xlsx -> date(2025, 2, 1) (first day of the following month)."""
    stem = os.path.splitext(os.path.basename(filename))[0]
    year, month = (int(part) for part in stem.split("-"))
    if month == 12:
        return date(year + 1, 1, 1)
    return date(year, month + 1, 1)


def to_int(value):
    return int(value) if value is not None else 0


def is_data_row(row):
    """Data rows have a completion ratio (0-1) in the completion column."""
    if len(row) <= COL_SHEET_TOTAL:
        return False
    value = row[COL_COMPLETION]
    return isinstance(value, (int, float)) and 0 <= value <= 1


def extract_file(filename):
    created_at = next_month_first_day(filename).isoformat()
    wb = openpyxl.load_workbook(filename, read_only=True, data_only=True)
    ws = wb["Summary"]

    rows = []
    for row in ws.iter_rows(values_only=True):
        if not is_data_row(row):
            continue
        rows.append(
            {
                "created_at": created_at,
                "locale": row[COL_LOCALE].strip(),
                "completion": round(row[COL_COMPLETION] * 100, 2),
                "active_managers": to_int(row[COL_MANAGERS]),
                "active_translators": to_int(row[COL_TRANSLATORS]),
                "active_contributors": to_int(row[COL_ACTIVE_CONTRIBUTORS]),
                "all_contributors": to_int(row[COL_ALL_CONTRIBUTORS]),
                "new_signups": to_int(row[COL_NEW_SIGNUPS]),
                "enabled_projects_score": row[COL_ENABLED_PROJECTS_SCORE] or 0,
                "sheet_chs": row[COL_SHEET_TOTAL],
            }
        )
    wb.close()
    return rows


def main():
    files = sorted(glob.glob(os.path.join(SHEETS_DIR, "*.xlsx")))
    if not files:
        raise SystemExit(f"No spreadsheets found in {SHEETS_DIR}")

    data = []
    for filename in files:
        rows = extract_file(filename)
        print(f"{os.path.basename(filename)}: {len(rows)} locales")
        data.extend(rows)

    with open(OUTPUT_FILE, "w") as f:
        json.dump(data, f, indent=2)
        f.write("\n")

    locales = sorted({row["locale"] for row in data})
    dates = sorted({row["created_at"] for row in data})
    print("")
    print(f"Files processed:  {len(files)}")
    print(f"Rows extracted:   {len(data)}")
    print(f"Distinct locales: {len(locales)} ({', '.join(locales)})")
    print(f"Date range:       {dates[0]} -> {dates[-1]}")
    print(f"Written to:       {OUTPUT_FILE}")


if __name__ == "__main__":
    main()

load_chs_backfill.py

"""
Backfill Community Health Score (CHS) history into Pontoon's
LocaleHealthSnapshot table from the spreadsheet data extracted by
extract_chs_backfill.py.

Scores and the final CHS are RECOMPUTED here from the raw inputs using the live
pontoon.insights.chs.compute_chs, so stored values stay consistent with the
current formula (rather than copying the sheet's precomputed totals). The sheet
total is used only to print a verification diff.

Existing (locale, created_at) snapshots are never overwritten: the script only
inserts missing rows, so it is safe to re-run.

How to run:
    1. Run extract_chs_backfill.py locally to produce chs_backfill.json.
    2. Open a Django shell, e.g.:
           heroku run --app mozilla-pontoon ./manage.py shell
    3. Paste the contents of chs_backfill.json between the triple quotes of
       CHS_JSON below, then paste this whole script into the shell.
"""

# ---------------------------------------------------------------------------
# Paste the contents of chs_backfill.json between the triple quotes below.
# ---------------------------------------------------------------------------
CHS_JSON = r"""
<<< PASTE THE CONTENTS OF chs_backfill.json HERE >>>
"""

import json

from datetime import date

from django.conf import settings

from pontoon.base.models import Locale
from pontoon.insights.chs import compute_chs
from pontoon.insights.models import LocaleHealthSnapshot

data = json.loads(CHS_JSON)

# Resolve locale codes to objects up front; report any unknown codes.
codes = {row["locale"] for row in data}
locales_by_code = {loc.code: loc for loc in Locale.objects.filter(code__in=codes)}
missing_codes = sorted(codes - set(locales_by_code))

# Existing (locale_id, created_at) pairs to skip.
dates = sorted({date.fromisoformat(row["created_at"]) for row in data})
existing = set(
    LocaleHealthSnapshot.objects.filter(created_at__in=dates).values_list(
        "locale_id", "created_at"
    )
)

snapshots = []
skipped_existing = 0
skipped_no_locale = 0
max_diff = 0.0
max_diff_row = None

for row in data:
    locale = locales_by_code.get(row["locale"])
    if locale is None:
        skipped_no_locale += 1
        continue

    created_at = date.fromisoformat(row["created_at"])
    if (locale.pk, created_at) in existing:
        skipped_existing += 1
        continue

    # Back-compute the raw enabled-projects count from its weighted score,
    # using the live settings (mirrors get_key_projects_enabled_by_locale).
    key_projects_enabled = round(
        row["enabled_projects_score"]
        / settings.ENABLED_PROJECT_POINTS
        * len(settings.KEY_PROJECT_SLUGS)
    )

    args = {
        "completion": row["completion"],
        "key_projects_enabled": key_projects_enabled,
        "active_managers": row["active_managers"],
        "active_translators": row["active_translators"],
        "active_contributors": row["active_contributors"],
        "all_contributors": row["all_contributors"],
        "new_signups": row["new_signups"],
    }
    chs_fields = compute_chs(args)

    diff = abs(float(chs_fields["chs"]) - float(row["sheet_chs"]))
    if diff > max_diff:
        max_diff = diff
        max_diff_row = (row["created_at"], row["locale"], chs_fields["chs"], row["sheet_chs"])

    snapshots.append(
        LocaleHealthSnapshot(locale=locale, created_at=created_at, **args, **chs_fields)
    )

LocaleHealthSnapshot.objects.bulk_create(snapshots, ignore_conflicts=True)

print(f"Created:            {len(snapshots)}")
print(f"Skipped (existing): {skipped_existing}")
print(f"Skipped (no locale):{skipped_no_locale}")
if missing_codes:
    print(f"Unknown locale codes: {', '.join(missing_codes)}")
print(f"Max CHS diff vs sheet: {round(max_diff, 3)} at {max_diff_row}")