Coverage for src/epl/adapters/outbound/pulselive_parsers.py: 53%

1"""Pure functions that map Pulselive JSON wire format into domain models.""" 

2 

3from dataclasses import dataclass 

4from typing import Any 

5 

6from ...domain.models import PulselivePlayer 

7 

8 

9def _int_or_none(value: object) -> int | None: 

10 """Coerce a numeric-looking JSON value to an int, or None if absent/malformed.""" 

11 if value is None: 

12 return None 

13 try: 

14 return int(value) 

15 except (TypeError, ValueError): 

16 return None 

17 

18 

19def _str_or_none(value: object) -> str | None: 

20 """Return ``str(value)`` for non-None input, else None.""" 

21 return None if value is None else str(value) 

22 

23 

24def _sub(raw: dict[str, Any], key: str) -> dict[str, Any]: 

25 """Return ``raw[key]`` when set to a dict, else an empty dict. 

26 

27 Wire entries occasionally send ``null`` for optional sub-objects; treating 

28 those the same as missing keeps the downstream ``get`` calls uniform. 

29 """ 

30 return raw.get(key) or {} 

31 

32 

33@dataclass(frozen=True) 

34class _PlayerParts: 

35 """Sub-dicts extracted from a Pulselive player row.""" 

36 

37 info: dict[str, Any] 

38 current_team: dict[str, Any] 

39 name: dict[str, Any] 

40 national_team: dict[str, Any] 

41 alt_ids: dict[str, Any] 

42 birth: dict[str, Any] 

43 

44 

45def _split_parts(raw: dict[str, Any]) -> _PlayerParts: 

46 """Bundle the six nested sub-dicts a player row can carry.""" 

47 return _PlayerParts( 

48 info=_sub(raw, "info"), 

49 current_team=_sub(raw, "currentTeam"), 

50 name=_sub(raw, "name"), 

51 national_team=_sub(raw, "nationalTeam"), 

52 alt_ids=_sub(raw, "altIds"), 

53 birth=_sub(raw, "birth"), 

54 ) 

55 

56 

57def _extract_display_name(name: dict[str, Any]) -> str: 

58 """Prefer ``name.display``; fall back to ``first + last`` for older entries.""" 

59 display = name.get("display") 

60 if display: 

61 return str(display) 

62 return f"{name.get('first', '')} {name.get('last', '')}".strip() 

63 

64 

65def _extract_team_name(current_team: dict[str, Any]) -> str | None: 

66 """Return the current-team display name, walking to the nested club as needed.""" 

67 return current_team.get("name") or _sub(current_team, "club").get("name") 

68 

69 

70def _extract_nationality(national_team: dict[str, Any]) -> str | None: 

71 """Prefer the full country name; fall back to the demonym.""" 

72 return national_team.get("country") or national_team.get("demonym") 

73 

74 

75def _extract_pulselive_id(raw: dict[str, Any]) -> str: 

76 """Return the primary id field, tolerating the two names Pulselive uses.""" 

77 return str(raw.get("playerId") or raw.get("id") or "") 

78 

79 

80def _parse_pulselive_player(raw: dict[str, Any]) -> PulselivePlayer: 

81 """Map a Pulselive ``/players`` entry to a domain ``PulselivePlayer``. 

82 

83 The wire shape carries ``info`` (position + shirt + loan flag), 

84 ``currentTeam`` (club affiliation), ``name`` (display + first/last), 

85 ``nationalTeam`` (nationality), and ``birth`` (DoB). The Opta id is 

86 nested at ``altIds.opta``. 

87 """ 

88 parts = _split_parts(raw) 

89 return PulselivePlayer( 

90 pulselive_id=_extract_pulselive_id(raw), 

91 opta_id=_str_or_none(parts.alt_ids.get("opta")), 

92 full_name=_extract_display_name(parts.name), 

93 position=parts.info.get("position"), 

94 position_detail=parts.info.get("positionInfo"), 

95 shirt_number=_int_or_none(parts.info.get("shirtNum")), 

96 current_team_name=_extract_team_name(parts.current_team), 

97 current_team_id=_str_or_none(parts.current_team.get("id")), 

98 on_loan=bool(parts.info.get("loan", False)), 

99 nationality=_extract_nationality(parts.national_team), 

100 date_of_birth=parts.birth.get("date"), 

101 )