Skip to content

Do not throw if no JUMBF data is found #180

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
tmathern wants to merge 3 commits into from
Closed

Do not throw if no JUMBF data is found #180

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ae04e35
fix: None
tmathern Oct 6, 2025
79e5d00
fix: COntext manager
tmathern Oct 6, 2025
0c73597
fix: Refactor
tmathern Oct 6, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
172 changes: 106 additions & 66 deletions src/c2pa/c2pa.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1264,7 +1264,9 @@ class Reader:
'file_error': "Error cleaning up file: {}",
'reader_cleanup_error': "Error cleaning up reader: {}",
'encoding_error': "Invalid UTF-8 characters in input: {}",
'closed_error': "Reader is closed"
'closed_error': "Reader is closed",
'not_initialized_error': "Reader is not properly initialized",
'no_manifest_error': "No manifest store to read"
}

@classmethod
Expand Down Expand Up @@ -1350,6 +1352,7 @@ def __init__(self,

self._closed = False
self._initialized = False
self._no_manifest_to_read = False

self._reader = None
self._own_stream = None
Expand Down Expand Up @@ -1397,6 +1400,12 @@ def __init__(self,
error = _parse_operation_result_for_error(
_lib.c2pa_error())
if error:
# Check if this is a ManifestNotFound error
if error.startswith("ManifestNotFound"):
# Set reader to no manifest state
self._no_manifest_to_read = True
self._initialized = True
return
raise C2paError(error)
raise C2paError(
Reader._ERROR_MESSAGES['reader_error'].format(
Expand Down Expand Up @@ -1460,6 +1469,12 @@ def __init__(self,
error = _parse_operation_result_for_error(
_lib.c2pa_error())
if error:
# Check if this is a ManifestNotFound error
if error.startswith("ManifestNotFound"):
# Set reader to no manifest state
self._no_manifest_to_read = True
self._initialized = True
return
raise C2paError(error)
raise C2paError(
Reader._ERROR_MESSAGES['reader_error'].format(
Expand Down Expand Up @@ -1514,6 +1529,12 @@ def __init__(self,
error = _parse_operation_result_for_error(
_lib.c2pa_error())
if error:
# Check if this is a ManifestNotFound error
if error.startswith("ManifestNotFound"):
# Set reader to no manifest state
self._no_manifest_to_read = True
self._initialized = True
return
raise C2paError(error)
raise C2paError(
Reader._ERROR_MESSAGES['reader_error'].format(
Expand All @@ -1539,16 +1560,18 @@ def __del__(self):
self._cleanup_resources()

def _ensure_valid_state(self):
"""Ensure the reader is in a valid state for operations.
"""Ensure the Reader is in a valid state for operations.

Raises:
C2paError: If the reader is closed, not initialized, or invalid
C2paError: If the Reader is closed, not initialized, invalid, or
has no manifest to read.
"""
# self._no_manifest_to_read is a valid state, albeit an empty one
if self._closed:
raise C2paError(Reader._ERROR_MESSAGES['closed_error'])
if not self._initialized:
raise C2paError("Reader is not properly initialized")
if not self._reader:
raise C2paError(Reader._ERROR_MESSAGES['not_initialized_error'])
if not self._reader and not self._no_manifest_to_read:
raise C2paError(Reader._ERROR_MESSAGES['closed_error'])

def _cleanup_resources(self):
Expand Down Expand Up @@ -1609,14 +1632,22 @@ def _get_cached_manifest_data(self) -> Optional[dict]:

Returns:
A dictionary containing the parsed manifest data, or None if
JSON parsing fails
JSON parsing fails or reader has no manifest to read

Raises:
C2paError: If there was an error getting the JSON
"""
if self._no_manifest_to_read:
return None

self._ensure_valid_state()

if self._manifest_data_cache is None:
if self._manifest_json_str_cache is None:
self._manifest_json_str_cache = self.json()
manifest_json = self.json()
if manifest_json is None:
return None
self._manifest_json_str_cache = manifest_json

try:
self._manifest_data_cache = json.loads(
Expand Down Expand Up @@ -1656,15 +1687,20 @@ def close(self):
self._manifest_data_cache = None
self._closed = True

def json(self) -> str:
def json(self) -> Optional[str]:
"""Get the manifest store as a JSON string.

Returns:
The manifest store as a JSON string
The manifest store as a JSON string, or None if no manifest is
found

Raises:
C2paError: If there was an error getting the JSON
C2paError: If there was an error getting the JSON (other than
ManifestNotFound)
"""
# Check if reader is in no-manifest state first
if self._no_manifest_to_read:
return None

self._ensure_valid_state()

Expand All @@ -1677,7 +1713,11 @@ def json(self) -> str:
if result is None:
error = _parse_operation_result_for_error(_lib.c2pa_error())
if error:
raise C2paError(error)
# Check if this is a ManifestNotFound error
if error.startswith("ManifestNotFound"):
# Set reader to no manifest state and return None
self._no_manifest_to_read = True
return None
raise C2paError("Error during manifest parsing in Reader")

# Cache the result and return it
Expand All @@ -1698,30 +1738,27 @@ def get_active_manifest(self) -> Optional[dict]:
Raises:
KeyError: If the active_manifest key is missing from the JSON
"""
try:
# Get cached manifest data
manifest_data = self._get_cached_manifest_data()
if manifest_data is None:
# raise C2paError("Failed to parse manifest JSON")
return None
# Get cached manifest data
manifest_data = self._get_cached_manifest_data()
if manifest_data is None:
# raise C2paError("Failed to parse manifest JSON")
return None

# Get the active manfiest id/label
if "active_manifest" not in manifest_data:
raise KeyError("No 'active_manifest' key found")
# Get the active manfiest id/label
if "active_manifest" not in manifest_data:
raise KeyError("No 'active_manifest' key found")

active_manifest_id = manifest_data["active_manifest"]
active_manifest_id = manifest_data["active_manifest"]

# Retrieve the active manifest data using manifest id/label
if "manifests" not in manifest_data:
raise KeyError("No 'manifests' key found in manifest data")
# Retrieve the active manifest data using manifest id/label
if "manifests" not in manifest_data:
raise KeyError("No 'manifests' key found in manifest data")

manifests = manifest_data["manifests"]
if active_manifest_id not in manifests:
raise KeyError("Active manifest not found in manifest store")
manifests = manifest_data["manifests"]
if active_manifest_id not in manifests:
raise KeyError("Active manifest not found in manifest store")

return manifests[active_manifest_id]
except C2paError.ManifestNotFound:
return None
return manifests[active_manifest_id]

def get_manifest_from_label(self, label: str) -> Optional[dict]:
"""Get a specific manifest from the manifest store by its label.
Expand All @@ -1740,23 +1777,20 @@ def get_manifest_from_label(self, label: str) -> Optional[dict]:
Raises:
KeyError: If the manifests key is missing from the JSON
"""
try:
# Get cached manifest data
manifest_data = self._get_cached_manifest_data()
if manifest_data is None:
# raise C2paError("Failed to parse manifest JSON")
return None
# Get cached manifest data
manifest_data = self._get_cached_manifest_data()
if manifest_data is None:
# raise C2paError("Failed to parse manifest JSON")
return None

if "manifests" not in manifest_data:
raise KeyError("No 'manifests' key found in manifest data")
if "manifests" not in manifest_data:
raise KeyError("No 'manifests' key found in manifest data")

manifests = manifest_data["manifests"]
if label not in manifests:
raise KeyError(f"Manifest {label} not found in manifest store")
manifests = manifest_data["manifests"]
if label not in manifests:
raise KeyError(f"Manifest {label} not found in manifest store")

return manifests[label]
except C2paError.ManifestNotFound:
return None
return manifests[label]

def get_validation_state(self) -> Optional[str]:
"""Get the validation state of the manifest store.
Expand All @@ -1770,16 +1804,13 @@ def get_validation_state(self) -> Optional[str]:
or None if the validation_state field is not present or if no
manifest is found or if there was an error parsing the JSON.
"""
try:
# Get cached manifest data
manifest_data = self._get_cached_manifest_data()
if manifest_data is None:
return None

return manifest_data.get("validation_state")
except C2paError.ManifestNotFound:
# Get cached manifest data
manifest_data = self._get_cached_manifest_data()
if manifest_data is None:
return None

return manifest_data.get("validation_state")

def get_validation_results(self) -> Optional[dict]:
"""Get the validation results of the manifest store.

Expand All @@ -1793,29 +1824,29 @@ def get_validation_results(self) -> Optional[dict]:
field is not present or if no manifest is found or if
there was an error parsing the JSON.
"""
try:
# Get cached manifest data
manifest_data = self._get_cached_manifest_data()
if manifest_data is None:
return None

return manifest_data.get("validation_results")
except C2paError.ManifestNotFound:
# Get cached manifest data
manifest_data = self._get_cached_manifest_data()
if manifest_data is None:
return None

def resource_to_stream(self, uri: str, stream: Any) -> int:
return manifest_data.get("validation_results")

def resource_to_stream(self, uri: str, stream: Any) -> Optional[int]:
"""Write a resource to a stream.

Args:
uri: The URI of the resource to write
stream: The stream to write to (any Python stream-like object)

Returns:
The number of bytes written
The number of bytes written, or None if no manifest to read

Raises:
C2paError: If there was an error writing the resource to stream
"""
if self._no_manifest_to_read:
return None

self._ensure_valid_state()

uri_str = uri.encode('utf-8')
Expand All @@ -1833,16 +1864,20 @@ def resource_to_stream(self, uri: str, stream: Any) -> int:

return result

def is_embedded(self) -> bool:
def is_embedded(self) -> Optional[bool]:
"""Check if the reader was created from an embedded manifest.
This method determines whether the C2PA manifest is embedded directly
in the asset file or stored remotely.
Returns:
True if the reader was created from an embedded manifest,
False if it was created from a remote manifest
False if it was created from a remote manifest,
None if no manifest to read
Raises:
C2paError: If there was an error checking the embedded status
"""
if self._no_manifest_to_read:
return None

self._ensure_valid_state()

result = _lib.c2pa_reader_is_embedded(self._reader)
Expand All @@ -1856,10 +1891,14 @@ def get_remote_url(self) -> Optional[str]:
is embedded in the asset, this will return None.
Returns:
The remote URL as a string if the manifest was obtained remotely,
None if the manifest is embedded or no remote URL is available
None if the manifest is embedded, no remote URL is available,
or no manifest to read
Raises:
C2paError: If there was an error getting the remote URL
"""
if self._no_manifest_to_read:
return None

self._ensure_valid_state()

result = _lib.c2pa_reader_remote_url(self._reader)
Expand Down Expand Up @@ -2194,6 +2233,7 @@ class Builder:
'cleanup_error': "Error during cleanup: {}",
'builder_cleanup': "Error cleaning up builder: {}",
'closed_error': "Builder is closed",
'not_initialized_error': "Builder is not properly initialized",
'manifest_error': "Invalid manifest data: must be string or dict",
'url_error': "Error setting remote URL: {}",
'resource_error': "Error adding resource: {}",
Expand Down Expand Up @@ -2376,7 +2416,7 @@ def _ensure_valid_state(self):
if self._closed:
raise C2paError(Builder._ERROR_MESSAGES['closed_error'])
if not self._initialized:
raise C2paError("Builder is not properly initialized")
raise C2paError(Builder._ERROR_MESSAGES['not_initialized_error'])
if not self._builder:
raise C2paError(Builder._ERROR_MESSAGES['closed_error'])

Expand Down
Loading