Coverage for src / lexigram / ai / relay / media.py: 96%

1"""Pure data-URI decoding and URL media resolution for relay conversion. 

2 

3The engine is synchronous and side-effect free. It never performs 

4network I/O: data URIs are decoded locally, and URL media is resolved 

5through the host-supplied resolver on :class:`ConversionContext`. 

6""" 

7 

8from __future__ import annotations 

9 

10import base64 

11import binascii 

12 

13from lexigram.ai.relay.context import ConversionContext 

14from lexigram.ai.relay.errors import media_resolution_required, serialization_error 

15from lexigram.contracts.ai.exceptions import RelayError 

16from lexigram.contracts.ai.relay.types import RelayFormat, RelayLoss 

17from lexigram.contracts.core.result import Err, Ok, Result 

18 

19__all__ = ["decode_data_uri", "resolve_media"] 

20 

21_DATA_URI_PREFIX = "data:" 

22_BASE64_MARKER = ";base64," 

23 

24 

25def decode_data_uri(uri: str) -> Result[tuple[str, str], RelayError]: 

26 """Decode a base64 data URI into ``(media_type, data)``. 

27 

28 Supports ``data:<media_type>;base64,<data>`` payloads. The returned 

29 data is exactly the raw base64 text (no ``data:`` prefix), ready for 

30 protocols that consume inline base64 (Claude ``image_source.data``, 

31 Gemini ``inlineData.data``). 

32 

33 Args: 

34 uri: A base64 data URI. 

35 

36 Returns: 

37 ``Ok((media_type, data))`` on success, or ``Err(RelayError)`` 

38 with code ``serialization_error`` when the URI is malformed, 

39 empty, or carries invalid base64. 

40 """ 

41 if not uri.startswith(_DATA_URI_PREFIX): 

42 return Err(serialization_error("expected a data URI")) 

43 rest = uri[len(_DATA_URI_PREFIX) :] 

44 marker = rest.find(_BASE64_MARKER) 

45 if marker < 0: 

46 return Err(serialization_error("data URI must be base64 encoded")) 

47 media_type = rest[:marker] 

48 if not media_type: 

49 return Err(serialization_error("data URI media type is empty")) 

50 data = rest[marker + len(_BASE64_MARKER) :] 

51 if not data: 

52 return Err(serialization_error("data URI payload is empty")) 

53 try: 

54 base64.b64decode(data, validate=True) 

55 except (binascii.Error, ValueError): 

56 return Err(serialization_error("data URI payload is not valid base64")) 

57 return Ok((media_type, data)) 

58 

59 

60def resolve_media( 

61 uri: str, 

62 context: ConversionContext, 

63 *, 

64 field: str, 

65 target: RelayFormat, 

66 lossy: bool = False, 

67) -> Result[tuple[str, str] | None, RelayError]: 

68 """Convert media content into ``(media_type, base64_data)``. 

69 

70 Data URIs decode locally and never touch the resolver. URL content 

71 is delegated to the context resolver; when no resolver exists the 

72 call fails with ``media_resolution_required`` unless *lossy* is set, 

73 in which case a ``media_unresolved_dropped`` loss is recorded and 

74 ``Ok(None)`` is returned. 

75 

76 Args: 

77 uri: A data URI or URL that requires conversion. 

78 context: Per-conversion context holding the resolver and the loss 

79 sink. 

80 field: Source wire JSON field the content came from, carried into 

81 errors and losses. 

82 target: Target wire format, used when recording losses. 

83 lossy: When ``True``, unresolvable URL content degrades to a 

84 recorded loss instead of a hard error. 

85 

86 Returns: 

87 ``Ok((media_type, data))``, ``Ok(None)`` for a lossy drop, or 

88 ``Err(RelayError)`` with the source field preserved. 

89 """ 

90 if uri.startswith(_DATA_URI_PREFIX): 

91 decoded = decode_data_uri(uri) 

92 if decoded.is_err(): 

93 return Err(_with_field(decoded.unwrap_err(), field)) 

94 return Ok(decoded.unwrap()) 

95 resolver = context.media_resolver 

96 if resolver is None: 

97 if lossy: 

98 context.losses.append( 

99 RelayLoss( 

100 field=field, 

101 target=target, 

102 reason="media_unresolved_dropped", 

103 severity="warning", 

104 ) 

105 ) 

106 return Ok(None) 

107 return Err(media_resolution_required(f"{field}: {uri}")) 

108 result = resolver.resolve(uri) 

109 if result.is_err(): 

110 return Err(_with_field(result.unwrap_err(), field)) 

111 return Ok(result.unwrap()) 

112 

113 

114def _with_field(error: RelayError, field: str) -> RelayError: 

115 """Prefix a resolver error detail with the source JSON field.""" 

116 return RelayError(f"{field}: {error}", code=error.code)