Spaces:
Running
Running
File size: 4,433 Bytes
8d42c78 b6bd2ee 8d42c78 b6bd2ee 8d42c78 b6bd2ee 8d42c78 b6bd2ee 8d42c78 b6bd2ee 8d42c78 b6bd2ee 8d42c78 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 | #!/usr/bin/env python3
"""Export BFSI-Bench preview rows to data/samples.json for the static Space.
The Space is public, so this writes at most PER_CATEGORY items per
category_code (sorted by id). The full test set stays on the gated Hub
dataset.
Prefers local hf_dataset/bfsi-bench/data/test.csv when present; otherwise
downloads the Hub dataset ground-truth/bfsi-bench.
"""
from __future__ import annotations
import csv
import io
import json
from pathlib import Path
_SPACE = Path(__file__).resolve().parents[1]
_REPO = Path(__file__).resolve().parents[3]
_LOCAL_CSV = _REPO / "hf_dataset" / "bfsi-bench" / "data" / "test.csv"
_OUT = _SPACE / "data" / "samples.json"
HUB_ID = "ground-truth/bfsi-bench"
JSON_COLS = ("sources", "evidence", "gold_nuggets")
PER_CATEGORY = 5
def _parse_json_col(raw: str, *, field: str, row_id: str):
text = (raw or "").strip()
if not text:
return []
try:
val = json.loads(text)
except json.JSONDecodeError as exc:
raise SystemExit(f"{row_id}: invalid JSON in {field}: {exc}") from exc
if val is None:
return []
if not isinstance(val, list):
raise SystemExit(f"{row_id}: {field} must be a JSON list")
return val
def _rows_from_csv(text: str) -> list[dict]:
reader = csv.DictReader(io.StringIO(text))
items = []
for raw in reader:
rid = (raw.get("id") or "").strip()
if not rid:
continue
item = {
"id": rid,
"category": (raw.get("category") or "").strip(),
"category_code": (raw.get("category_code") or "").strip(),
"query": raw.get("query") or "",
"gold_answer": raw.get("gold_answer") or "",
}
for col in JSON_COLS:
item[col] = _parse_json_col(raw.get(col) or "", field=col, row_id=rid)
items.append(item)
items.sort(key=lambda r: r["id"])
return items
def _counts_by_category(items: list[dict]) -> dict[str, int]:
counts: dict[str, int] = {}
for it in items:
code = it["category_code"] or "unknown"
counts[code] = counts.get(code, 0) + 1
return counts
def take_per_category(
items: list[dict], n: int = PER_CATEGORY
) -> list[dict]:
"""Keep the first n items in each category_code. Items must already be sorted by id."""
by_cat: dict[str, list[dict]] = {}
for it in items:
code = it["category_code"] or "unknown"
bucket = by_cat.setdefault(code, [])
if len(bucket) < n:
bucket.append(it)
sampled: list[dict] = []
for code in sorted(by_cat):
sampled.extend(by_cat[code])
sampled.sort(key=lambda r: r["id"])
return sampled
def _load_csv_text() -> tuple[str, str]:
if _LOCAL_CSV.is_file():
return _LOCAL_CSV.read_text(encoding="utf-8-sig"), str(_LOCAL_CSV)
try:
from huggingface_hub import hf_hub_download
except ImportError:
raise SystemExit(
"Local test.csv missing and huggingface_hub is not installed. "
"pip install huggingface_hub or place hf_dataset/bfsi-bench/data/test.csv"
) from None
path = hf_hub_download(
repo_id=HUB_ID,
repo_type="dataset",
filename="data/test.csv",
)
return Path(path).read_text(encoding="utf-8-sig"), f"hub:{HUB_ID}/data/test.csv"
def main() -> None:
csv_text, source = _load_csv_text()
items = _rows_from_csv(csv_text)
dataset_counts = _counts_by_category(items)
preview = take_per_category(items, PER_CATEGORY)
preview_counts = _counts_by_category(preview)
payload = {
"dataset": HUB_ID,
"split": "test",
"preview_per_category": PER_CATEGORY,
"dataset_total": len(items),
"dataset_counts": dataset_counts,
"total": len(preview),
"counts": preview_counts,
"items": preview,
}
_OUT.parent.mkdir(parents=True, exist_ok=True)
_OUT.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(
json.dumps(
{
"wrote": str(_OUT.relative_to(_REPO)),
"source": source,
"preview_per_category": PER_CATEGORY,
"dataset_total": len(items),
"total": len(preview),
"counts": preview_counts,
},
indent=2,
)
)
if __name__ == "__main__":
main()
|