From 088a1e415ed329f4c4c2fc4c70fd0c4d20bd3a0f Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 22 Jul 2026 20:37:14 +0000 Subject: [PATCH 1/3] Shard the generate data workflow per category MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The generate-hacs-data workflow processed each category in a single job. The integration category is by far the largest, so that job dominated runtime and was the most likely to hit GitHub rate limits or time out. Split generation into shards (integration ×3, plugin ×2, everything else ×1) and merge each category's shards back into a single result before the existing summarize/publish jobs run. - generate_category_data: add a stable, salt-free `shard_for` helper and thread a shard `index/total` through generation, slicing both the published data and the default-file repo list by the same lowercased full_name so shards are disjoint and their union equals the unsharded set. Refactor the summary/diff/validation/writing tail into a reusable `finalize_category_output`; a `--shard x/y` CLI selects a shard and writes a partial artifact. With `--shard 1/1` (the default) behaviour is unchanged. - merge_category_data: new script that concatenates the shard partials and runs finalize once on the full merged dataset. - workflow: emit a shard matrix, run generation per shard, add a merge job per category that rebuilds the per-category artifact in the same shape as before, and repoint summarize/publish accordingly. - tests: cover shard_for/_parse_shard/_slice_by_shard and assert a sharded generate + merge matches the unsharded output. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01NL8ieP9vEVJLTZ51esR6VD --- .github/workflows/generate-hacs-data.yml | 95 ++++- scripts/data/generate_category_data.py | 375 ++++++++++++------ scripts/data/merge_category_data.py | 63 +++ tests/conftest.py | 2 + .../data/test_generate_category_data.py | 118 +++++- 5 files changed, 527 insertions(+), 126 deletions(-) create mode 100644 scripts/data/merge_category_data.py diff --git a/.github/workflows/generate-hacs-data.yml b/.github/workflows/generate-hacs-data.yml index 4117d1c5a5a..ed536eb4ff3 100644 --- a/.github/workflows/generate-hacs-data.yml +++ b/.github/workflows/generate-hacs-data.yml @@ -42,26 +42,50 @@ jobs: runs-on: ubuntu-latest if: github.repository == 'hacs/integration' outputs: + shards: ${{ steps.set-matrix.outputs.shards }} categories: ${{ steps.set-matrix.outputs.categories }} steps: - id: set-matrix + env: + EVENT_NAME: ${{ github.event_name }} + INPUT_CATEGORY: ${{ inputs.category }} run: | - if [[ "${{ github.event_name }}" == "workflow_dispatch" ]] && [[ "${{ inputs.category }}" != "None" ]] && [[ "${{ inputs.category }}" != "" ]]; then - echo "categories=['${{ inputs.category }}']" >> $GITHUB_OUTPUT + # Number of shards each category is split into (default 1). + declare -A shard_counts=( [integration]=3 [plugin]=2 ) + + if [[ "$EVENT_NAME" == "workflow_dispatch" ]] && [[ "$INPUT_CATEGORY" != "None" ]] && [[ -n "$INPUT_CATEGORY" ]]; then + categories=("$INPUT_CATEGORY") else - echo "categories=['appdaemon','integration','plugin','python_script','template','theme']" >> $GITHUB_OUTPUT + categories=(appdaemon integration plugin python_script template theme) fi + shards_json="[]" + for category in "${categories[@]}"; do + count=${shard_counts[$category]:-1} + for ((shard=1; shard<=count; shard++)); do + shards_json=$(jq -c \ + --arg c "$category" --argjson s "$shard" --argjson t "$count" \ + '. + [{category: $c, shard: $s, shards: $t}]' <<< "$shards_json") + done + done + + categories_json=$(printf '%s\n' "${categories[@]}" | jq -R . | jq -cs .) + + echo "shards=$shards_json" >> $GITHUB_OUTPUT + echo "categories=$categories_json" >> $GITHUB_OUTPUT + echo "shards: $shards_json" + echo "categories: $categories_json" + category-data: runs-on: ubuntu-latest needs: generate-matrix if: github.repository == 'hacs/integration' - name: Generate ${{ matrix.category }} data + name: Generate ${{ matrix.category }} data (shard ${{ matrix.shard }}/${{ matrix.shards }}) strategy: fail-fast: false max-parallel: 1 matrix: - category: ${{ fromJSON( needs.generate-matrix.outputs.categories )}} + include: ${{ fromJSON( needs.generate-matrix.outputs.shards )}} steps: - name: Checkout the repository uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 @@ -83,12 +107,64 @@ jobs: scripts/install/frontend scripts/install/pip_packages --requirement requirements_generate_data.txt - - name: Generate ${{ matrix.category }} data - run: python3 -m scripts.data.generate_category_data ${{ matrix.category }} + - name: Generate ${{ matrix.category }} data (shard ${{ matrix.shard }}/${{ matrix.shards }}) + run: python3 -m scripts.data.generate_category_data ${{ matrix.category }} --shard ${{ matrix.shard }}/${{ matrix.shards }} env: DATA_GENERATOR_TOKEN: ${{ secrets.DATA_GENERATOR_TOKEN }} FORCE_REPOSITORY_UPDATE: ${{ inputs.forceRepositoryUpdate }} + - name: Upload shard artifacts + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: shard-${{ matrix.category }}-${{ matrix.shard }} + path: outputdata/_shards/${{ matrix.category }}/${{ matrix.shard }} + if-no-files-found: error + retention-days: 1 + + merge: + runs-on: ubuntu-latest + needs: [generate-matrix, category-data] + if: github.repository == 'hacs/integration' + name: Merge ${{ matrix.category }} data + strategy: + fail-fast: false + matrix: + category: ${{ fromJSON( needs.generate-matrix.outputs.categories )}} + steps: + - name: Checkout the repository + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + persist-credentials: false + + - name: Set up Python + uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0 + id: python + with: + python-version: "3.14" + cache: 'pip' + cache-dependency-path: | + requirements_base.txt + requirements_generate_data.txt + + - name: Install dependencies + run: | + scripts/install/frontend + scripts/install/pip_packages --requirement requirements_generate_data.txt + + # Each shard artifact contains data.json + stored.json at its root. Without + # merge-multiple, download-artifact nests each artifact in its own + # name-scoped subdirectory, so the per-shard files never collide. + - name: Download shard artifacts + uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1 + with: + pattern: shard-${{ matrix.category }}-* + path: outputdata/_shards/${{ matrix.category }} + + - name: Merge ${{ matrix.category }} shards + run: python3 -m scripts.data.merge_category_data ${{ matrix.category }} + env: + DATA_GENERATOR_TOKEN: ${{ secrets.DATA_GENERATOR_TOKEN }} + - name: Validate output with JQ run: | jq -c . outputdata/${{ matrix.category }}/data.json @@ -129,7 +205,7 @@ jobs: summarize: name: Summarize runs-on: ubuntu-latest - needs: category-data + needs: merge if: ${{ always() && github.repository == 'hacs/integration' }} outputs: changedCategories: ${{ steps.combined.outputs.changedCategories }} @@ -159,6 +235,7 @@ jobs: const subDirectories = fs.readdirSync("outputdata", { withFileTypes: true }) .filter(entry => entry.isDirectory()) .map(entry => entry.name) + .filter(directory => fs.existsSync(`outputdata/${directory}/summary.json`)) for (const directory of subDirectories) { let changedPctTarget = Number(process.env.HACS_CHANGED_PCT_TARGET) @@ -287,7 +364,7 @@ jobs: notify_on_failure: runs-on: ubuntu-latest name: Trigger Discord notification when jobs fail - needs: ["generate-matrix", "category-data", "summarize", "publish"] + needs: ["generate-matrix", "category-data", "merge", "summarize", "publish"] if: ${{ always() && github.repository == 'hacs/integration' && contains(join(needs.*.result, ','), 'failure') && github.event_name == 'schedule' }} steps: - name: Send notification diff --git a/scripts/data/generate_category_data.py b/scripts/data/generate_category_data.py index fe04c61df21..c9386f905a2 100644 --- a/scripts/data/generate_category_data.py +++ b/scripts/data/generate_category_data.py @@ -2,8 +2,10 @@ from __future__ import annotations +import argparse import asyncio from datetime import datetime +import hashlib import json import logging import os @@ -56,9 +58,37 @@ log_handler.addHandler(stream_handler) OUTPUT_DIR = os.path.join(os.getcwd(), "outputdata") +SHARDS_DIR = os.path.join(OUTPUT_DIR, "_shards") COMPARE_IGNORE = {"etag_releases", "etag_repository", "last_fetched"} +def shard_for(full_name: str, shards: int) -> int: + """Return the (0-based) shard a repository belongs to. + + Uses a stable, salt-free hash so the assignment is identical across + processes and runs (unlike the built-in ``hash``). + """ + if shards <= 1: + return 0 + digest = hashlib.md5(full_name.lower().encode()).digest() + return int.from_bytes(digest[:8], "big") % shards + + +def _slice_by_shard( + current_data: dict[str, dict[str, Any]], + index: int, + shards: int, +) -> dict[str, dict[str, Any]]: + """Return the subset of ``current_data`` that belongs to the given shard.""" + if shards <= 1: + return current_data + return { + key: value + for key, value in current_data.items() + if shard_for(value["full_name"], shards) == index + } + + def jsonprint(data: any): print( json.dumps( @@ -304,8 +334,15 @@ async def generate_data_for_category( repository_name: str | None, current_data: dict[str, dict[str, Any]], force: bool, + index: int = 0, + shards: int = 1, ) -> dict[str, dict[str, Any]]: - """Generate data for category.""" + """Generate data for category. + + When ``shards`` > 1 only the repositories belonging to shard ``index`` + (0-based) are processed and stored, so the result is a disjoint slice of + the full category data. + """ removed = ( [] if repository_name is not None @@ -313,11 +350,11 @@ async def generate_data_for_category( ) await self.data.register_base_data( category, - {} if force else current_data, + {} if force else _slice_by_shard(current_data, index, shards), removed, ) self.queue.clear() - await self.get_category_repositories(category, repository_name, removed) + await self.get_category_repositories(category, repository_name, removed, index, shards) async def _handle_queue(): if not self.queue.pending_tasks: @@ -357,6 +394,8 @@ async def get_category_repositories( category: str, repository_name: str | None, removed: list[str], + index: int = 0, + shards: int = 1, ) -> None: """Get repositories from category.""" repositories = ( @@ -371,6 +410,11 @@ async def get_category_repositories( # hacs/integration i not in the default file, but it's still needed repositories.append("hacs/integration") + if shards > 1: + repositories = [ + repo for repo in repositories if shard_for(repo, shards) == index + ] + for repo in repositories: if repo in removed: self.log.warning("Skipping %s as it's removed from HACS", repo) @@ -453,13 +497,174 @@ async def async_github_get_hacs_default_file(self, filename: str) -> list: return json_loads(decode_content(response.data.content)) -async def generate_category_data(category: str, repository_name: str = None): - """Generate data.""" +def _dump_diff(updated: dict[str, dict[str, Any]], data_file) -> None: + """Write a diff-friendly (COMPARE_IGNORE stripped) JSON dump.""" + json.dump( + { + i: {k: v for k, v in d.items() if k not in COMPARE_IGNORE} + for i, d in updated.items() + }, + data_file, + cls=JSONEncoder, + sort_keys=True, + indent=2, + ) + + +async def finalize_category_output( + hacs: AdjustedHacs, + category: str, + stored_data: dict[str, dict[str, Any]], + current_data: dict[str, dict[str, Any]], + updated_data: dict[str, dict[str, Any]], +) -> None: + """Summarize, validate and write the final per-category output files. + + Runs once on the full (merged) dataset, so the summary, diff and validation + are computed against the complete category data. + """ + os.makedirs(os.path.join(OUTPUT_DIR, category), exist_ok=True) + os.makedirs(os.path.join(OUTPUT_DIR, "diff"), exist_ok=True) + + summary = await hacs.summarize_data(current_data, updated_data) + with open( + os.path.join(OUTPUT_DIR, "summary.json"), + mode="w", + encoding="utf-8", + ) as data_file: + json.dump( + summary, + data_file, + cls=JSONEncoder, + sort_keys=True, + indent=2, + ) + + did_raise = False + if ( + not updated_data + or len(updated_data) == 0 + or not isinstance(updated_data, dict) + ): + print_error_and_exit("Updated data is empty", category) + did_raise = True + + try: + VALIDATE_GENERATED_V2_REPO_DATA[category](updated_data) + except vol.Invalid as error: + did_raise = True + errors = expand_and_humanize_error(updated_data, error) + if isinstance(errors, list): + for err in errors: + print(f"::error::{err}") + sys.exit(1) + + print_error_and_exit(f"Invalid data: {errors}", category) + + if did_raise: + print_error_and_exit( + "Validation did raise but did not exit!", category) + sys.exit(1) # Fallback, should not be reached + + with open( + os.path.join(OUTPUT_DIR, category, "stored.json"), + mode="w", + encoding="utf-8", + ) as data_file: + json.dump( + stored_data, + data_file, + cls=JSONEncoder, + separators=(",", ":"), + ) + with open( + os.path.join(OUTPUT_DIR, category, "data.json"), + mode="w", + encoding="utf-8", + ) as data_file: + json.dump( + updated_data, + data_file, + cls=JSONEncoder, + separators=(",", ":"), + ) + with open( + os.path.join(OUTPUT_DIR, category, "repositories.json"), + mode="w", + encoding="utf-8", + ) as repositories_file: + json.dump( + [v["full_name"] for v in updated_data.values()], + repositories_file, + separators=(",", ":"), + sort_keys=True, + ) + + with open( + os.path.join(OUTPUT_DIR, "diff", f"{category}_before.json"), + mode="w", + encoding="utf-8", + ) as data_file: + _dump_diff(current_data, data_file) + + with open( + os.path.join(OUTPUT_DIR, "diff", f"{category}_after.json"), + mode="w", + encoding="utf-8", + ) as data_file: + _dump_diff(updated_data, data_file) + + +def write_shard_output( + category: str, + shard_number: int, + updated_data: dict[str, dict[str, Any]], + stored_slice: dict[str, dict[str, Any]], +) -> None: + """Write a shard's partial data to a shard-scoped path for later merging.""" + shard_dir = os.path.join(SHARDS_DIR, category, str(shard_number)) + os.makedirs(shard_dir, exist_ok=True) + with open( + os.path.join(shard_dir, "data.json"), + mode="w", + encoding="utf-8", + ) as data_file: + json.dump( + updated_data, + data_file, + cls=JSONEncoder, + separators=(",", ":"), + ) + with open( + os.path.join(shard_dir, "stored.json"), + mode="w", + encoding="utf-8", + ) as data_file: + json.dump( + stored_slice, + data_file, + cls=JSONEncoder, + separators=(",", ":"), + ) + + +async def generate_category_data( + category: str, + repository_name: str | None = None, + shard: tuple[int, int] = (1, 1), +) -> None: + """Generate data. + + ``shard`` is a 1-based ``(number, total)`` pair. With a total of 1 the full + category output (summary, diff and validation) is produced, matching the + original behaviour. With a total > 1 only that shard's disjoint slice is + generated and written as a partial artifact for a later merge step. + """ + shard_number, shards = shard + index = shard_number - 1 async with ClientSession() as session: hacs = AdjustedHacs( session=session, token=os.getenv("DATA_GENERATOR_TOKEN")) - os.makedirs(os.path.join(OUTPUT_DIR, category), exist_ok=True) - os.makedirs(os.path.join(OUTPUT_DIR, "diff"), exist_ok=True) force = os.environ.get("FORCE_REPOSITORY_UPDATE") == "True" stored_data = await hacs.data_client.get_data(category, validate=False) current_data = ( @@ -480,125 +685,63 @@ async def generate_category_data(category: str, repository_name: str = None): repository_name, current_data, force=force, + index=index, + shards=shards, ) - summary = await hacs.summarize_data(current_data, updated_data) - with open( - os.path.join(OUTPUT_DIR, "summary.json"), - mode="w", - encoding="utf-8", - ) as data_file: - json.dump( - summary, - data_file, - cls=JSONEncoder, - sort_keys=True, - indent=2, - ) - - did_raise = False - if ( - not updated_data - or len(updated_data) == 0 - or not isinstance(updated_data, dict) - ): - print_error_and_exit("Updated data is empty", category) - did_raise = True - - try: - VALIDATE_GENERATED_V2_REPO_DATA[category](updated_data) - except vol.Invalid as error: - did_raise = True - errors = expand_and_humanize_error(updated_data, error) - if isinstance(errors, list): - for err in errors: - print(f"::error::{err}") - sys.exit(1) - - print_error_and_exit(f"Invalid data: {errors}", category) - - if did_raise: - print_error_and_exit( - "Validation did raise but did not exit!", category) - sys.exit(1) # Fallback, should not be reached - - with open( - os.path.join(OUTPUT_DIR, category, "stored.json"), - mode="w", - encoding="utf-8", - ) as data_file: - json.dump( - stored_data, - data_file, - cls=JSONEncoder, - separators=(",", ":"), - ) - with open( - os.path.join(OUTPUT_DIR, category, "data.json"), - mode="w", - encoding="utf-8", - ) as data_file: - json.dump( + if shards > 1: + write_shard_output( + category, + shard_number, updated_data, - data_file, - cls=JSONEncoder, - separators=(",", ":"), - ) - with open( - os.path.join(OUTPUT_DIR, category, "repositories.json"), - mode="w", - encoding="utf-8", - ) as repositories_file: - json.dump( - [v["full_name"] for v in updated_data.values()], - repositories_file, - separators=(",", ":"), - sort_keys=True, + _slice_by_shard(stored_data, index, shards), ) + return - with open( - os.path.join(OUTPUT_DIR, "diff", f"{category}_before.json"), - mode="w", - encoding="utf-8", - ) as data_file: - json.dump( - { - i: { - k: v - for k, v in d.items() if k not in COMPARE_IGNORE - } - for i, d in current_data.items() - }, - data_file, - cls=JSONEncoder, - sort_keys=True, - indent=2, - ) + await finalize_category_output( + hacs, + category, + stored_data, + current_data, + updated_data, + ) - with open( - os.path.join(OUTPUT_DIR, "diff", f"{category}_after.json"), - mode="w", - encoding="utf-8", - ) as data_file: - json.dump( - { - i: { - k: v - for k, v in d.items() if k not in COMPARE_IGNORE - } - for i, d in updated_data.items() - }, - data_file, - cls=JSONEncoder, - sort_keys=True, - indent=2, - ) + +def _parse_shard(value: str) -> tuple[int, int]: + """Parse a ``x/y`` (1-based) shard argument into an ``(x, y)`` tuple.""" + parts = value.split("/") + if len(parts) != 2: + raise argparse.ArgumentTypeError( + f"Invalid shard '{value}', expected format 'x/y' (e.g. 1/3)" + ) + try: + number, total = int(parts[0]), int(parts[1]) + except ValueError: + raise argparse.ArgumentTypeError( + f"Invalid shard '{value}', expected format 'x/y' (e.g. 1/3)" + ) from None + if total < 1 or number < 1 or number > total: + raise argparse.ArgumentTypeError( + f"Invalid shard '{value}', require 1 <= x <= y" + ) + return (number, total) if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Generate HACS compliant data.") + parser.add_argument("category") + parser.add_argument("repository_name", nargs="?", default=None) + parser.add_argument( + "--shard", + type=_parse_shard, + default=(1, 1), + help="Shard to generate as 'x/y' (1-based), e.g. --shard 1/3.", + ) + args = parser.parse_args() asyncio.run( generate_category_data( - sys.argv[1], # category - sys.argv[2] if len(sys.argv) > 2 else None, # repository_name + args.category, + args.repository_name, + shard=args.shard, ) ) diff --git a/scripts/data/merge_category_data.py b/scripts/data/merge_category_data.py new file mode 100644 index 00000000000..4039aadf7fe --- /dev/null +++ b/scripts/data/merge_category_data.py @@ -0,0 +1,63 @@ +"""Merge sharded HACS data into a single per-category output. + +Each shard of a category produces a partial ``data.json`` (its slice of the +generated data) and a partial ``stored.json`` (its slice of the currently +published data) under ``outputdata/_shards///``. This script +concatenates those partials back into the full dataset and runs the shared +``finalize_category_output`` step, which computes the summary and diff and +validates the complete category data. +""" + +from __future__ import annotations + +import argparse +import asyncio +import glob +import json +import os +from typing import Any + +from aiohttp import ClientSession + +from .common import print_error_and_exit +from .generate_category_data import ( + SHARDS_DIR, + AdjustedHacs, + finalize_category_output, +) + + +def _load_shard_files(category: str, name: str) -> dict[str, dict[str, Any]]: + """Merge a given partial file across all shards of a category.""" + merged: dict[str, dict[str, Any]] = {} + paths = sorted(glob.glob(os.path.join(SHARDS_DIR, category, "*", name))) + if not paths: + print_error_and_exit(f"No shard '{name}' files found", category) + for path in paths: + with open(path, encoding="utf-8") as shard_file: + merged.update(json.loads(shard_file.read())) + return merged + + +async def merge_category_data(category: str) -> None: + """Merge all shards for a category and write the final output.""" + updated_data = _load_shard_files(category, "data.json") + stored_data = _load_shard_files(category, "stored.json") + + async with ClientSession() as session: + hacs = AdjustedHacs( + session=session, token=os.getenv("DATA_GENERATOR_TOKEN")) + await finalize_category_output( + hacs, + category, + stored_data, + stored_data, + updated_data, + ) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Merge sharded HACS data.") + parser.add_argument("category") + args = parser.parse_args() + asyncio.run(merge_category_data(args.category)) diff --git a/tests/conftest.py b/tests/conftest.py index acc012c856d..52ebb4532f8 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -357,6 +357,8 @@ async def proxy_session(hass: HomeAssistant) -> Generator: with patch( "homeassistant.helpers.aiohttp_client.async_get_clientsession", return_value=mock_session, ), patch("scripts.data.generate_category_data.ClientSession", ProxyClientSession), patch( + "scripts.data.merge_category_data.ClientSession", ProxyClientSession, + ), patch( "aiohttp.ClientSession", ProxyClientSession, ): yield diff --git a/tests/scripts/data/test_generate_category_data.py b/tests/scripts/data/test_generate_category_data.py index 6b602d9feb8..f875c5a2fb4 100644 --- a/tests/scripts/data/test_generate_category_data.py +++ b/tests/scripts/data/test_generate_category_data.py @@ -1,14 +1,24 @@ """Test generate category data.""" +import argparse import asyncio import json import os +import shutil from typing import Any from homeassistant.core import HomeAssistant import pytest -from scripts.data.generate_category_data import OUTPUT_DIR, generate_category_data +from scripts.data.generate_category_data import ( + OUTPUT_DIR, + SHARDS_DIR, + _parse_shard, + _slice_by_shard, + generate_category_data, + shard_for, +) +from scripts.data.merge_category_data import merge_category_data from tests.common import ( FIXTURES_PATH, @@ -311,3 +321,109 @@ async def test_generate_category_data_with_30plus_prereleases( f"scripts/data/test_generate_category_data_with_30plus_prereleases/{ category_test_data['category']}.json", ) + + +@pytest.mark.parametrize( + ("full_name", "shards", "expected"), + [ + ("hacs/integration", 3, 2), + ("hacs/integration", 2, 1), + ("hacs-test-org/integration-basic", 3, 1), + ("octocat/Hello-World", 3, 1), + ("octocat/Hello-World", 2, 0), + ("anything", 1, 0), + ("anything", 0, 0), + ], +) +def test_shard_for(full_name: str, shards: int, expected: int): + """Shard assignment is stable and matches precomputed values.""" + assert shard_for(full_name, shards) == expected + + +def test_shard_for_is_case_insensitive(): + """Repository casing must not change the shard assignment.""" + assert shard_for("HACS/Integration", 3) == shard_for("hacs/integration", 3) + + +def test_shard_for_partitions_all_buckets(): + """Every shard is used and assignments stay within range.""" + names = [f"user{i}/repo{i}" for i in range(500)] + for shards in (2, 3): + assignments = [shard_for(name, shards) for name in names] + assert set(assignments) == set(range(shards)) + + +@pytest.mark.parametrize( + ("value", "expected"), + [("1/3", (1, 3)), ("3/3", (3, 3)), ("1/1", (1, 1))], +) +def test_parse_shard_valid(value: str, expected: tuple[int, int]): + assert _parse_shard(value) == expected + + +@pytest.mark.parametrize("value", ("0/3", "4/3", "3/0", "abc", "1/2/3", "1", "-1/3")) +def test_parse_shard_invalid(value: str): + with pytest.raises(argparse.ArgumentTypeError): + _parse_shard(value) + + +def test_slice_by_shard_is_disjoint_and_complete(): + """Slices are disjoint, cover everything, and match the shard function.""" + data = {str(i): {"full_name": f"user{i}/repo{i}"} for i in range(200)} + shards = 3 + slices = [_slice_by_shard(data, index, shards) for index in range(shards)] + + keys = [set(entry) for entry in slices] + assert set.union(*keys) == set(data) + assert sum(len(entry) for entry in slices) == len(data) + for first in range(shards): + for second in range(first + 1, shards): + assert keys[first].isdisjoint(keys[second]) + + for index, entries in enumerate(slices): + for value in entries.values(): + assert shard_for(value["full_name"], shards) == index + + +def test_slice_by_shard_single_shard_returns_all(): + data = {"1": {"full_name": "a/b"}} + assert _slice_by_shard(data, 0, 1) == data + + +@pytest.mark.parametrize("category_test_data", category_test_data_parametrized()) +@pytest.mark.parametrize("shards", (2, 3)) +async def test_sharded_generate_matches_unsharded( + hass: HomeAssistant, + response_mocker: ResponseMocker, + category_test_data: CategoryTestData, + shards: int, +): + """Generating a category in shards and merging equals the unsharded run.""" + category = category_test_data["category"] + response_mocker.add( + f"https://data-v2.hacs.xyz/{category}/data.json", + MockedResponse(content={}, keep=True), + ) + + # Reference: the original, unsharded output. + await generate_category_data(category) + with open(f"{OUTPUT_DIR}/{category}/data.json", encoding="utf-8") as file: + reference_data = recursive_remove_key( + json.loads(file.read()), ("last_fetched",)) + with open(f"{OUTPUT_DIR}/{category}/repositories.json", encoding="utf-8") as file: + reference_repositories = json.loads(file.read()) + + # Sharded: generate each shard, then merge. + shutil.rmtree(os.path.join(SHARDS_DIR, category), ignore_errors=True) + for shard in range(1, shards + 1): + await generate_category_data(category, shard=(shard, shards)) + await merge_category_data(category) + + with open(f"{OUTPUT_DIR}/{category}/data.json", encoding="utf-8") as file: + merged_data = recursive_remove_key( + json.loads(file.read()), ("last_fetched",)) + with open(f"{OUTPUT_DIR}/{category}/repositories.json", encoding="utf-8") as file: + merged_repositories = json.loads(file.read()) + + assert merged_data == reference_data + assert sorted(merged_repositories) == sorted(reference_repositories) From ba5fbb6eb9908ee54e312d2c67cf59ac03c3eeac Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 23 Jul 2026 06:05:13 +0000 Subject: [PATCH 2/3] Address review: fix 1-shard path and make output deterministic - generate_category_data: distinguish a sharded invocation from an unsharded one by whether a shard was passed, not by shards > 1. With --shard omitted the full category output is produced (unchanged); with any shard given -- including 1/1 -- a partial artifact is written, so the single-shard categories go through the merge step uniformly like the workflow expects (previously they took the finalize path and the shard artifact upload had nothing to upload). - finalize_category_output: sort repositories.json explicitly (sort_keys is a no-op on a list) and sort the keys of data.json/stored.json, so the published output is deterministic regardless of shard-assembly order. - tests: replace the network-driven sharded/unsharded equivalence test (which required API-usage snapshots) with a network-free roundtrip test of write_shard_output + _load_shard_files across 1/2/3 shards; drop the now-unused merge ClientSession patch. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01NL8ieP9vEVJLTZ51esR6VD --- scripts/data/generate_category_data.py | 34 +++++++---- tests/conftest.py | 2 - .../data/test_generate_category_data.py | 59 ++++++++----------- 3 files changed, 48 insertions(+), 47 deletions(-) diff --git a/scripts/data/generate_category_data.py b/scripts/data/generate_category_data.py index c9386f905a2..f9463a6d714 100644 --- a/scripts/data/generate_category_data.py +++ b/scripts/data/generate_category_data.py @@ -566,6 +566,11 @@ async def finalize_category_output( "Validation did raise but did not exit!", category) sys.exit(1) # Fallback, should not be reached + # All three files are published to R2, and under sharding the merged + # insertion order depends on shard-assembly order. Sort so the output is + # deterministic regardless of how the data was produced. (sort_keys sorts + # the repo-id keys of the dicts; repositories.json is an explicit list that + # sort_keys would not touch, so it is sorted directly.) with open( os.path.join(OUTPUT_DIR, category, "stored.json"), mode="w", @@ -576,6 +581,7 @@ async def finalize_category_output( data_file, cls=JSONEncoder, separators=(",", ":"), + sort_keys=True, ) with open( os.path.join(OUTPUT_DIR, category, "data.json"), @@ -587,6 +593,7 @@ async def finalize_category_output( data_file, cls=JSONEncoder, separators=(",", ":"), + sort_keys=True, ) with open( os.path.join(OUTPUT_DIR, category, "repositories.json"), @@ -594,10 +601,9 @@ async def finalize_category_output( encoding="utf-8", ) as repositories_file: json.dump( - [v["full_name"] for v in updated_data.values()], + sorted(v["full_name"] for v in updated_data.values()), repositories_file, separators=(",", ":"), - sort_keys=True, ) with open( @@ -651,16 +657,19 @@ def write_shard_output( async def generate_category_data( category: str, repository_name: str | None = None, - shard: tuple[int, int] = (1, 1), + shard: tuple[int, int] | None = None, ) -> None: """Generate data. - ``shard`` is a 1-based ``(number, total)`` pair. With a total of 1 the full - category output (summary, diff and validation) is produced, matching the - original behaviour. With a total > 1 only that shard's disjoint slice is - generated and written as a partial artifact for a later merge step. + ``shard`` is a 1-based ``(number, total)`` pair, or ``None`` for an + unsharded run. When ``shard`` is ``None`` the full category output (summary, + diff and validation) is produced, matching the original behaviour. When a + shard is given only that shard's disjoint slice is generated and written as + a partial artifact for a later merge step -- this holds even for ``1/1``, so + every sharded invocation goes through the merge step uniformly. """ - shard_number, shards = shard + sharded = shard is not None + shard_number, shards = shard if sharded else (1, 1) index = shard_number - 1 async with ClientSession() as session: hacs = AdjustedHacs( @@ -689,7 +698,7 @@ async def generate_category_data( shards=shards, ) - if shards > 1: + if sharded: write_shard_output( category, shard_number, @@ -734,8 +743,11 @@ def _parse_shard(value: str) -> tuple[int, int]: parser.add_argument( "--shard", type=_parse_shard, - default=(1, 1), - help="Shard to generate as 'x/y' (1-based), e.g. --shard 1/3.", + default=None, + help=( + "Shard to generate as 'x/y' (1-based), e.g. --shard 1/3. " + "Omit for a full, unsharded run." + ), ) args = parser.parse_args() asyncio.run( diff --git a/tests/conftest.py b/tests/conftest.py index 52ebb4532f8..acc012c856d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -357,8 +357,6 @@ async def proxy_session(hass: HomeAssistant) -> Generator: with patch( "homeassistant.helpers.aiohttp_client.async_get_clientsession", return_value=mock_session, ), patch("scripts.data.generate_category_data.ClientSession", ProxyClientSession), patch( - "scripts.data.merge_category_data.ClientSession", ProxyClientSession, - ), patch( "aiohttp.ClientSession", ProxyClientSession, ): yield diff --git a/tests/scripts/data/test_generate_category_data.py b/tests/scripts/data/test_generate_category_data.py index f875c5a2fb4..d7bfcd2498b 100644 --- a/tests/scripts/data/test_generate_category_data.py +++ b/tests/scripts/data/test_generate_category_data.py @@ -17,8 +17,9 @@ _slice_by_shard, generate_category_data, shard_for, + write_shard_output, ) -from scripts.data.merge_category_data import merge_category_data +from scripts.data.merge_category_data import _load_shard_files from tests.common import ( FIXTURES_PATH, @@ -390,40 +391,30 @@ def test_slice_by_shard_single_shard_returns_all(): assert _slice_by_shard(data, 0, 1) == data -@pytest.mark.parametrize("category_test_data", category_test_data_parametrized()) -@pytest.mark.parametrize("shards", (2, 3)) -async def test_sharded_generate_matches_unsharded( - hass: HomeAssistant, - response_mocker: ResponseMocker, - category_test_data: CategoryTestData, - shards: int, -): - """Generating a category in shards and merging equals the unsharded run.""" - category = category_test_data["category"] - response_mocker.add( - f"https://data-v2.hacs.xyz/{category}/data.json", - MockedResponse(content={}, keep=True), - ) +@pytest.mark.parametrize("shards", (1, 2, 3)) +def test_write_and_merge_shard_partials_roundtrip(shards: int): + """Writing disjoint shard partials and merging reassembles the full set. - # Reference: the original, unsharded output. - await generate_category_data(category) - with open(f"{OUTPUT_DIR}/{category}/data.json", encoding="utf-8") as file: - reference_data = recursive_remove_key( - json.loads(file.read()), ("last_fetched",)) - with open(f"{OUTPUT_DIR}/{category}/repositories.json", encoding="utf-8") as file: - reference_repositories = json.loads(file.read()) + This mirrors the production shard -> merge data flow (``write_shard_output`` + followed by ``_load_shard_files``) without touching the network, so it needs + no API-usage snapshots. + """ + category = "integration" + full = {str(i): {"full_name": f"user{i}/repo{i}"} for i in range(30)} - # Sharded: generate each shard, then merge. - shutil.rmtree(os.path.join(SHARDS_DIR, category), ignore_errors=True) - for shard in range(1, shards + 1): - await generate_category_data(category, shard=(shard, shards)) - await merge_category_data(category) + shard_root = os.path.join(SHARDS_DIR, category) + shutil.rmtree(shard_root, ignore_errors=True) - with open(f"{OUTPUT_DIR}/{category}/data.json", encoding="utf-8") as file: - merged_data = recursive_remove_key( - json.loads(file.read()), ("last_fetched",)) - with open(f"{OUTPUT_DIR}/{category}/repositories.json", encoding="utf-8") as file: - merged_repositories = json.loads(file.read()) + written = 0 + for index in range(shards): + shard_slice = _slice_by_shard(full, index, shards) + written += len(shard_slice) + write_shard_output(category, index + 1, shard_slice, shard_slice) + + # The shards partition the input: disjoint and complete. + assert written == len(full) + + assert _load_shard_files(category, "data.json") == full + assert _load_shard_files(category, "stored.json") == full - assert merged_data == reference_data - assert sorted(merged_repositories) == sorted(reference_repositories) + shutil.rmtree(shard_root, ignore_errors=True) From 0ba849d73e8ecbc90c7b5856ff2a8c87439afb7a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20S=C3=B8rensen?= Date: Thu, 23 Jul 2026 18:31:25 +0200 Subject: [PATCH 3/3] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- scripts/data/merge_category_data.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/scripts/data/merge_category_data.py b/scripts/data/merge_category_data.py index 4039aadf7fe..44d6de615b3 100644 --- a/scripts/data/merge_category_data.py +++ b/scripts/data/merge_category_data.py @@ -35,7 +35,14 @@ def _load_shard_files(category: str, name: str) -> dict[str, dict[str, Any]]: print_error_and_exit(f"No shard '{name}' files found", category) for path in paths: with open(path, encoding="utf-8") as shard_file: - merged.update(json.loads(shard_file.read())) + shard_data = json.load(shard_file) + overlap = merged.keys() & shard_data.keys() + if overlap: + print_error_and_exit( + f"Duplicate keys across shards in '{name}': {sorted(overlap)[:5]}", + category, + ) + merged.update(shard_data) return merged