"""Shared catalog data models."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Mapping
def _empty_to_none(value: str | None) -> str | None:
if value is None:
return None
value = value.strip()
return value or None
def _to_int(value: str | None) -> int | None:
value = _empty_to_none(value)
if value is None:
return None
return int(value)
def _to_bool(value: object) -> bool:
if isinstance(value, bool):
return value
if value is None:
return False
return str(value).strip().lower() in {"1", "true", "yes", "y"}
[docs]
@dataclass(frozen=True)
class DatasetRecord:
"""One recovered HectaresBC catalog record."""
dataset_id: str
source_family: str
source_zip_path: str
source_filename: str
source_stem: str
title_candidate: str
verification_status: str
manifest_size_bytes: int | None
known_gaps: str
uncertainty_notes: str
fields: Mapping[str, str]
[docs]
@classmethod
def from_row(cls, row: Mapping[str, str]) -> "DatasetRecord":
return cls(
dataset_id=row["dataset_id"],
source_family=row["source_family"],
source_zip_path=row["source_zip_path"],
source_filename=row["source_filename"],
source_stem=row["source_stem"],
title_candidate=row.get("title_candidate", ""),
verification_status=row.get("verification_status", ""),
manifest_size_bytes=_to_int(row.get("manifest_size_bytes")),
known_gaps=row.get("known_gaps", ""),
uncertainty_notes=row.get("uncertainty_notes", ""),
fields=dict(row),
)
[docs]
def field(self, name: str, default: str = "") -> str:
"""Return a raw recovered field without changing source names."""
return self.fields.get(name, default)
[docs]
def bool_field(self, name: str) -> bool:
"""Return a recovered flag-like field as a boolean."""
return _to_bool(self.fields.get(name))
[docs]
def int_field(self, name: str) -> int | None:
"""Return a recovered integer field when present."""
return _to_int(self.fields.get(name))
[docs]
def to_dict(self) -> dict[str, object]:
"""Serialize the record while preserving all recovered source fields."""
data: dict[str, object] = dict(self.fields)
data["manifest_size_bytes"] = self.manifest_size_bytes
return data
[docs]
@dataclass(frozen=True)
class ResolvedDatasetPath:
"""Filesystem resolution for a catalog record's raw ZIP payload."""
dataset_id: str
source_zip_path: str
data_repo_path: Path
raw_relative_path: Path
absolute_path: Path
submodule_initialized: bool
path_metadata_exists: bool
content_present: bool
[docs]
@dataclass(frozen=True)
class ContentStatus:
"""Local content status for a resolved raw ZIP payload."""
dataset_id: str
status: str
local_path: Path
submodule_initialized: bool
path_metadata_exists: bool
content_present: bool
message: str
[docs]
@dataclass(frozen=True)
class BackendDiagnostic:
"""One backend readiness diagnostic check."""
backend: str
check: str
status: str
message: str
command_summary: str | None = None
remediation: str | None = None
secret_safe: bool = True
[docs]
@dataclass(frozen=True)
class FetchResult:
"""Structured result for a dataset retrieval request."""
dataset_id: str
status: str
backend: str
local_path: Path
message: str
diagnostics: tuple[BackendDiagnostic, ...] = ()
command_summary: str | None = None
verification_performed: bool = False
secret_safe: bool = True