File size: 4,277 Bytes
7e2d640
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Persist eval scenarios in a Google Sheet (the source of truth on HF Spaces,
whose filesystem is ephemeral).

One row per scenario; the ``expect`` block is flattened into typed columns. The
app passes the same service-account creds dict it already uses for Google Docs
(needs the spreadsheets scope) plus SCENARIOS_SHEET_ID.
"""
from __future__ import annotations

from typing import Any

# Column order written to row 1. Loading maps by header name, so reordering
# columns in the sheet is safe; renaming them is not.
COLUMNS = [
    "id", "turns", "use_tools",
    "escalate", "format_ok", "no_prompt_leak", "tools_any", "tools_none", "judge",
]

SCOPES = ["https://www.googleapis.com/auth/spreadsheets"]


def _service(creds_dict: dict):
    from google.oauth2.service_account import Credentials
    from googleapiclient.discovery import build

    creds = Credentials.from_service_account_info(creds_dict, scopes=SCOPES)
    return build("sheets", "v4", credentials=creds, cache_discovery=False)


def _first_tab(svc, sheet_id: str) -> str:
    meta = svc.spreadsheets().get(spreadsheetId=sheet_id).execute()
    sheets = meta.get("sheets", [])
    if not sheets:
        return "Sheet1"
    return sheets[0]["properties"]["title"]


def _to_bool(s: str):
    v = (s or "").strip().lower()
    if v in ("true", "1", "yes", "да", "y"):
        return True
    if v in ("false", "0", "no", "нет", "n"):
        return False
    return None


def _b(v) -> str:
    if v is True:
        return "TRUE"
    if v is False:
        return "FALSE"
    return ""


def _row_to_scenario(row: list[str], header: list[str]) -> dict:
    d = {h: (row[i] if i < len(row) else "") for i, h in enumerate(header)}
    sc: dict[str, Any] = {
        "id": (d.get("id", "") or "").strip(),
    }
    sc["turns"] = [t for t in (d.get("turns", "") or "").splitlines() if t.strip()]

    ut = _to_bool(d.get("use_tools", ""))
    if ut is not None:
        sc["use_tools"] = ut

    expect: dict[str, Any] = {}
    for key in ("escalate", "format_ok"):
        b = _to_bool(d.get(key, ""))
        if b is not None:
            expect[key] = b
    for key in ("no_prompt_leak", "tools_none"):
        if _to_bool(d.get(key, "")) is True:
            expect[key] = True
    tools_any = [t.strip() for t in (d.get("tools_any", "") or "").split(",") if t.strip()]
    if tools_any:
        expect["tools_any"] = tools_any
    judge = (d.get("judge", "") or "").strip()
    if judge:
        expect["judge"] = judge
    sc["expect"] = expect
    return sc


def _scenario_to_row(sc: dict) -> list[str]:
    expect = sc.get("expect", {}) or {}
    turns = sc.get("turns") or ([sc["message"]] if sc.get("message") else [])
    return [
        sc.get("id", "") or "",
        "\n".join(turns),
        _b(sc.get("use_tools")) if "use_tools" in sc else "",
        _b(expect.get("escalate")) if "escalate" in expect else "",
        _b(expect.get("format_ok")) if "format_ok" in expect else "",
        "TRUE" if expect.get("no_prompt_leak") else "",
        ", ".join(expect.get("tools_any", []) or []),
        "TRUE" if expect.get("tools_none") else "",
        expect.get("judge", "") or "",
    ]


def load_scenarios_from_sheet(creds_dict: dict, sheet_id: str) -> list[dict]:
    svc = _service(creds_dict)
    tab = _first_tab(svc, sheet_id)
    resp = (
        svc.spreadsheets()
        .values()
        .get(spreadsheetId=sheet_id, range=tab)
        .execute()
    )
    values = resp.get("values", [])
    if not values:
        return []
    header = [h.strip() for h in values[0]]
    out = []
    for row in values[1:]:
        sc = _row_to_scenario(row, header)
        if sc["id"]:
            out.append(sc)
    return out


def save_scenarios_to_sheet(creds_dict: dict, sheet_id: str, scenarios: list[dict]) -> None:
    svc = _service(creds_dict)
    tab = _first_tab(svc, sheet_id)
    rows = [COLUMNS] + [_scenario_to_row(sc) for sc in scenarios]
    svc.spreadsheets().values().clear(spreadsheetId=sheet_id, range=tab).execute()
    (
        svc.spreadsheets()
        .values()
        .update(
            spreadsheetId=sheet_id,
            range=f"{tab}!A1",
            valueInputOption="RAW",
            body={"values": rows},
        )
        .execute()
    )