File size: 2,918 Bytes
1e1b8a7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Map v2 API JSON to ViewData. This is the only module that reads API field names.

Contract (v2/playground/README.md "Response contract"): status, answer, claims[]
{text, quote, slug, verified, mm:ss, ...}, triage[] {slug, label, reason}, warnings[],
plus the public API's episodes {slug: {title, date, url}}. Unknown fields are ignored.
"""

from __future__ import annotations

from dataclasses import dataclass

STATUSES = ("answered", "partial", "not_in_podcast", "false_premise", "refused", "error")


@dataclass(frozen=True)
class EpisodeView:
    slug: str
    title: str
    date: str
    url: str


@dataclass(frozen=True)
class ClaimView:
    text: str
    quote: str
    verified: bool
    episode: EpisodeView
    timestamp: str


@dataclass(frozen=True)
class TriageView:
    episode: EpisodeView
    label: str
    reason: str


@dataclass(frozen=True)
class ViewData:
    status: str
    answer_markdown: str
    claims: tuple[ClaimView, ...]
    sources: tuple[EpisodeView, ...]  # cited episodes, first-cited order, unique
    triage: tuple[TriageView, ...]
    warnings: tuple[str, ...]

    @property
    def is_error(self) -> bool:
        return self.status == "error"


def _s(value: object) -> str:
    return value if isinstance(value, str) else ""


def map_api_response(data: object) -> ViewData:
    if not isinstance(data, dict):
        return ViewData("error", "", (), (), (), ())
    status = _s(data.get("status"))
    if status not in STATUSES:
        status = "error"
    raw_eps = data.get("episodes") if isinstance(data.get("episodes"), dict) else {}

    def episode(slug: str) -> EpisodeView:
        meta = raw_eps.get(slug) if isinstance(raw_eps.get(slug), dict) else {}
        return EpisodeView(slug, _s(meta.get("title")), _s(meta.get("date")), _s(meta.get("url")))

    claims: list[ClaimView] = []
    for item in data.get("claims") if isinstance(data.get("claims"), list) else []:
        if not isinstance(item, dict):
            continue
        claims.append(ClaimView(
            text=_s(item.get("text")), quote=_s(item.get("quote")),
            # Only an explicit true counts as verified.
            verified=item.get("verified") is True,
            episode=episode(_s(item.get("slug"))), timestamp=_s(item.get("mm:ss")),
        ))
    sources: dict[str, EpisodeView] = {}
    for claim in claims:
        if claim.episode.slug:
            sources.setdefault(claim.episode.slug, claim.episode)
    triage = tuple(
        TriageView(episode(_s(t.get("slug"))), _s(t.get("label")), _s(t.get("reason")))
        for t in (data.get("triage") if isinstance(data.get("triage"), list) else []) if isinstance(t, dict)
    )
    warnings = tuple(str(w) for w in (data.get("warnings") if isinstance(data.get("warnings"), list) else []) if str(w).strip())
    return ViewData(status, _s(data.get("answer")), tuple(claims), tuple(sources.values()), triage, warnings)