Create app.py

app.py

@@ -0,0 +1,122 @@
+import json
+import os
+from functools import lru_cache
+from typing import List, Tuple
+
+import gradio as gr
+from datasets import Dataset, load_dataset
+
+DATASET_ID = os.environ.get(
+    "CIRCLECI_RESULTS_DATASET_ID", "transformers-community/circleci-test-results"
+)
+MAX_ROWS = 200
+
+
+@lru_cache(maxsize=1)
+def _load_dataset() -> Dataset | None:
+    try:
+        return load_dataset(DATASET_ID, split="train")
+    except Exception as error:
+        print(f"Failed to load dataset {DATASET_ID}: {error}")
+        return None
+
+
+def _filter_records(dataset: Dataset, repo: str, pr: str, sha: str) -> List[dict]:
+    repo = repo.strip().lower()
+    pr = pr.strip()
+    sha = sha.strip().lower()
+
+    def _matches(example: dict) -> bool:
+        if repo and repo not in (example.get("repository") or "").lower():
+            return False
+        if pr and pr != (example.get("pr_number") or ""):
+            return False
+        if sha and sha not in (example.get("commit_sha") or "").lower():
+            return False
+        return True
+
+    items = [ex for ex in dataset if _matches(ex)]
+    items.sort(key=lambda ex: ex.get("collected_at") or "", reverse=True)
+    return items[:MAX_ROWS]
+
+
+def query(repo: str, pr: str, sha: str) -> Tuple[List[List[str]], str]:
+    dataset = _load_dataset()
+    if dataset is None:
+        return [], json.dumps({"error": "Dataset not available"}, indent=2)
+
+    records = _filter_records(dataset, repo, pr, sha)
+    table_rows = []
+    for record in records:
+        table_rows.append(
+            [
+                record.get("collected_at", ""),
+                record.get("repository", ""),
+                record.get("pr_number", ""),
+                record.get("commit_sha", "")[:12],
+                record.get("workflow_id", ""),
+                str(record.get("failure_count", 0)),
+                str(record.get("job_count", 0)),
+                str(record.get("test_count", 0)),
+            ]
+        )
+
+    details = json.dumps({}, indent=2)
+    if records:
+        payload = records[0].get("payload_json", "{}")
+        try:
+            details = json.dumps(json.loads(payload), indent=2)
+        except json.JSONDecodeError:
+            details = json.dumps({"error": "Unable to parse payload"}, indent=2)
+
+    return table_rows, details
+
+
+def refresh_dataset() -> str:
+    _load_dataset.cache_clear()
+    dataset = _load_dataset()
+    if dataset is None:
+        return "Failed to refresh dataset."
+    return f"Dataset refreshed with {len(dataset)} rows."
+
+
+with gr.Blocks() as demo:
+    gr.Markdown(
+        """
+        # CircleCI Test Collection Helper
+
+        Use the filters below to inspect CircleCI test aggregation records for the Transformers repository (or any
+        repository that uploads data to the `transformers-community/circleci-test-results` dataset).
+        """
+    )
+
+    with gr.Row():
+        repo_box = gr.Textbox(label="Repository", placeholder="huggingface/transformers")
+        pr_box = gr.Textbox(label="PR number")
+        sha_box = gr.Textbox(label="Commit SHA (prefix accepted)")
+
+    with gr.Row():
+        search_btn = gr.Button("Search")
+        refresh_btn = gr.Button("Refresh dataset cache")
+
+    table = gr.Dataframe(
+        headers=[
+            "Collected at",
+            "Repository",
+            "PR",
+            "Commit",
+            "Workflow ID",
+            "Failures",
+            "Jobs",
+            "Tests",
+        ],
+        wrap=True,
+    )
+    json_view = gr.Code(label="Latest entry details", language="json")
+    status = gr.Markdown("")
+
+    search_btn.click(query, inputs=[repo_box, pr_box, sha_box], outputs=[table, json_view])
+    refresh_btn.click(refresh_dataset, outputs=status)
+
+if __name__ == "__main__":
+    demo.queue(max_size=20).launch()