Functions¶

def fetch_list_mmcif(
    entry_ids: list[str],
    provider: str,
    output_dir: Path,
    source_uri: str | None = None,
    fetch_options: FetchOptions | None = None,
) -> list[IngestionProvenance]:
    """Fetch a list of raw mmCIF files from a provider URL, write them to
    output_dir, return provenance for each.

    Calls `fetch_mmcif()` once per entry ID. If
    `fetch_options.allow_partial` is True, entries that fail to fetch
    (`RuntimeError` or `ValueError`) are skipped instead of aborting
    the whole batch.

    Args:
        entry_ids: The PDB/PDBe entry identifiers to fetch.
        provider: One of "pdbe" or "pdb", used to build the download
            URL when `source_uri` is not given.
        source_uri: A URL/path template overriding `provider`'s default
            URL template, applied to every entry. Must contain an
            `{id}` placeholder (same convention as `_PROVIDER_URLS`),
            formatted per entry_id — e.g.
            `"https://example.org/mirror/{id}.cif"`.
        output_dir: Directory to write the downloaded (or cached)
            files into; created if missing.
        fetch_options: Caching/decompression options, including
            `allow_partial`. Defaults to `FetchOptions()` when not
            given.

    Returns:
        A list of `IngestionProvenance`, one per successfully fetched
        entry, in the same order as `entry_ids` (with failed entries
        omitted when `allow_partial` is True).

    Raises:
        ValueError: If `provider` is invalid and no `source_uri` was
            given, and `allow_partial` is False.
        RuntimeError: If a fetch fails and `allow_partial` is False.
    """

    fetch_options = fetch_options or FetchOptions()

    provenance_list = []
    for entry_id in entry_ids:
        try:
            provenance = fetch_mmcif(
                entry_id=entry_id,
                provider=provider,
                source_uri=(
                    source_uri.format(id=entry_id) if source_uri else None
                ),
                output_dir=output_dir,
                fetch_options=fetch_options,
            )
        except (RuntimeError, ValueError):
            if fetch_options.allow_partial:
                continue
            raise
        provenance_list.append(provenance)

    return provenance_list

def fetch_mmcif(
    entry_id: str,
    provider: str,
    source_uri: str | None,
    output_dir: Path,
    fetch_options: FetchOptions | None = None,
) -> IngestionProvenance:
    """Fetch a raw mmCIF file from a provider URL, write it to
    output_dir, return provenance.

    Resolves the download URL from `source_uri` or from `provider`
    ("pdbe"/"pdb"), reuses a cached file when one is fresh (per
    `fetch_options`), otherwise downloads it, optionally decompresses
    it, and writes it to `output_dir`.

    Args:
        entry_id: The PDB/PDBe entry identifier (e.g. "1abc").
        provider: One of "pdbe" or "pdb", used to build the download
            URL when `source_uri` is not given.
        source_uri: An explicit URL to fetch from, overriding
            `provider`'s default URL template.
        output_dir: Directory to write the downloaded (or cached) file
            into; created if missing.
        fetch_options: Caching/decompression options. Defaults
            to `FetchOptions()` when not given.

    Returns:
        `IngestionProvenance` describing the provider, source URL,
        retrieval timestamp, whether the result came from cache, and
        the entry's own revision date where the provider exposes one.

    Raises:
        ValueError: If `provider` is not "pdbe"/"pdb" and no
            `source_uri` was given.
        RuntimeError: If the HTTP request fails, or the response
            cannot be decompressed/decoded.
    """

    fetch_options = fetch_options or FetchOptions()

    if source_uri:
        url = source_uri
    elif provider in _PROVIDER_URLS:
        fmt_id = entry_id.lower() if provider == "pdbe" else entry_id.upper()
        url = _PROVIDER_URLS[provider].format(id=fmt_id)
    else:
        raise ValueError(
            f"provider={provider!r} requires an explicit source_uri "
            f"or one of 'pdbe' or 'pdb'"
        )

    if fetch_options.use_cache:
        cached = resolve_cache_hit(
            entry_id=entry_id,
            output_dir=output_dir,
            max_age_seconds=fetch_options.max_age_seconds,
            stale_behavior=fetch_options.stale_behavior,
        )
        if cached is not None:
            return IngestionProvenance(
                provider=provider,
                source_uri=url,
                retrieved_at=mtime_iso(cached),
                from_cache=True,
                revision_date=(
                    _revision_date_from_cif_file(cached)
                    if provider == "pdb"
                    else None
                ),
            )

    httpx = require("httpx", "ingestion")
    try:
        resp = httpx.get(url, follow_redirects=True, timeout=60.0)
        resp.raise_for_status()
    except httpx.HTTPStatusError as exc:
        raise RuntimeError(
            f"HTTP {exc.response.status_code} fetching {url}"
        ) from exc
    except httpx.RequestError as exc:
        raise RuntimeError(f"Network error fetching {url}: {exc}") from exc

    raw_bytes = resp.content
    is_gzipped = url.endswith(".gz") or raw_bytes[:2] == b"\x1f\x8b"

    if is_gzipped and fetch_options.decompress:
        try:
            raw_bytes = gzip.decompress(raw_bytes)
        except gzip.BadGzipFile as exc:
            raise RuntimeError(
                f"Failed to decompress response from {url}"
            ) from exc
        is_gzipped = False

    output_dir.mkdir(parents=True, exist_ok=True)

    if is_gzipped:
        out_path = output_dir / f"{entry_id.lower()}.cif.gz"
        out_path.write_bytes(raw_bytes)
    else:
        try:
            content = raw_bytes.decode("utf-8")
        except UnicodeDecodeError as exc:
            decompressed = "decompressed " if fetch_options.decompress else ""
            raise RuntimeError(
                f"Failed to decode {decompressed}response from {url} as UTF-8"
            ) from exc
        out_path = output_dir / f"{entry_id.lower()}.cif"
        out_path.write_text(content, encoding="utf-8")

    if provider == "pdb":
        revision_date = _revision_date_from_cif_file(out_path)
    elif provider == "pdbe":
        revision_date = _pdbe_revision_date(entry_id)
    else:
        revision_date = None

    return IngestionProvenance(
        provider=provider,
        source_uri=url,
        retrieved_at=now_iso(),
        from_cache=False,
        revision_date=revision_date,
    )

def ingest_local_mmcif(
    path: Path,
    source_uri: str | None = None,
) -> IngestionProvenance:
    """Describe an already-downloaded mmCIF file as provenance, without
    fetching or copying anything.

    For a file that arrived via some external bulk source — a local
    rsync mirror of PDB, an archived dated snapshot, a lab file server —
    rather than `fetch_mmcif()`. Pandora has no bulk-download step of its
    own; this just lets such a file's origin be recorded the same way a
    live fetch's is, so it can still be attached to a `ProvenanceBundle`.

    Args:
        path: Path to the existing mmCIF file on disk. Not modified or
            copied.
        source_uri: A label for where this file came from (e.g.
            "pdb_snapshot_2024-01-29"), stored verbatim. Defaults to
            `str(path)` when not given.

    Returns:
        `IngestionProvenance` with `provider="local"`, `from_cache=False`
        (this isn't Pandora's own fetch cache), the file's own mtime as
        `retrieved_at`, and `revision_date` read from the file's
        `_pdbx_audit_revision_history` category when present.

    Raises:
        ValueError: If `path` doesn't exist.
    """

    if not path.exists():
        raise ValueError(f"path={path} does not exist")

    return IngestionProvenance(
        provider="local",
        source_uri=source_uri or str(path),
        retrieved_at=mtime_iso(path),
        from_cache=False,
        revision_date=_revision_date_from_cif_file(path),
    )

def load_policy(path: str) -> canonicalisationPolicy:
    """Load a `canonicalisationPolicy` from a YAML file.

    Args:
        path: Path to the policy YAML file.

    Returns:
        The parsed, validated `canonicalisationPolicy`.
    """

    with open(path) as stream:
        data = yaml.safe_load(stream)

    return canonicalisationPolicy.model_validate(data)

def search_pdbe(
    query: str,
    rows: int = 100,
    start: int = 0,
) -> list[str]:
    """Search PDBe's Solr-based search API for matching entry ids.

    `query` is a raw Solr/Lucene query string, passed through verbatim as
    the `q` parameter — Pandora doesn't validate or translate it. See
    https://www.ebi.ac.uk/pdbe/api/doc/search.html for the query syntax
    and the full list of searchable fields. The returned ids feed
    directly into `fetch_mmcif`/`fetch_list_mmcif` with
    `provider="pdbe"`.

    Unlike RCSB's search API, PDBe's Solr endpoint doesn't reject a
    malformed query — an unparseable clause is silently dropped rather
    than erroring, which can make a typo'd query match far more than
    intended. Sanity-check the returned count against what you expect.

    Args:
        query: A Solr query string, e.g. `'experimental_method:"X-ray
            diffraction" AND resolution:[0 TO 1.5]'`.
        rows: Maximum number of ids to return in this page.
        start: Offset into the full result set, for paging through more
            than `rows` matches.

    Returns:
        Matching PDB ids, lowercase (PDBe's own convention), deduplicated
        — PDBe's Solr index can list the same id more than once (e.g. one
        row per assembly/entity). Empty list if nothing matches.

    Raises:
        RuntimeError: If the request fails.
    """

    params = {
        "q": query,
        "fl": "pdb_id",
        "rows": rows,
        "start": start,
        "wt": "json",
    }
    httpx = require("httpx", "ingestion")
    try:
        resp = httpx.get(_PDBE_SEARCH_URL, params=params, timeout=60.0)
        resp.raise_for_status()
    except httpx.HTTPStatusError as exc:
        raise RuntimeError(
            f"HTTP {exc.response.status_code} querying PDBe search API: "
            f"{exc.response.text}"
        ) from exc
    except httpx.RequestError as exc:
        raise RuntimeError(
            f"Network error querying PDBe search API: {exc}"
        ) from exc

    docs = resp.json().get("response", {}).get("docs", [])
    # PDBe's Solr index can return the same pdb_id more than once (e.g.
    # once per assembly/entity row); dedupe, preserving result order.
    return list(dict.fromkeys(doc["pdb_id"] for doc in docs if "pdb_id" in doc))

def search_rcsb(
    query: dict,
    return_type: str = "entry",
    rows: int = 100,
    start: int = 0,
) -> list[str]:
    """Search RCSB's Search API for matching entry ids.

    `query` is RCSB's own query tree, passed through verbatim — Pandora
    doesn't validate or translate it. See
    https://search.rcsb.org/#search-api for the query language
    (terminal/group nodes, services, attributes, operators) and
    https://search.rcsb.org/rcsbsearch/v2/metadata/schema for the full
    list of searchable attributes. The returned ids feed directly into
    `fetch_mmcif`/`fetch_list_mmcif` with `provider="pdb"`.

    Args:
        query: An RCSB query tree, e.g. `{"type": "terminal", "service":
            "text", "parameters": {"attribute": "exptl.method",
            "operator": "exact_match", "value": "X-RAY DIFFRACTION"}}`.
        return_type: What kind of id to return — "entry" for PDB entry
            ids (the default), or one of RCSB's other result types
            ("polymer_entity", "assembly", ...).
        rows: Maximum number of ids to return in this page.
        start: Offset into the full result set, for paging through more
            than `rows` matches.

    Returns:
        Matching identifiers, in RCSB's own casing (e.g. "104M"). Empty
        list if nothing matches.

    Raises:
        RuntimeError: If the request fails, including a malformed query
            — RCSB validates the query tree server-side and returns its
            own error message on a bad attribute/operator.
    """

    body = {
        "query": query,
        "return_type": return_type,
        "request_options": {"paginate": {"start": start, "rows": rows}},
    }
    httpx = require("httpx", "ingestion")
    try:
        resp = httpx.post(_RCSB_SEARCH_URL, json=body, timeout=60.0)
        resp.raise_for_status()
    except httpx.HTTPStatusError as exc:
        raise RuntimeError(
            f"HTTP {exc.response.status_code} querying RCSB search API: "
            f"{exc.response.text}"
        ) from exc
    except httpx.RequestError as exc:
        raise RuntimeError(
            f"Network error querying RCSB search API: {exc}"
        ) from exc

    if resp.status_code == 204:
        return []
    return [hit["identifier"] for hit in resp.json().get("result_set", [])]

def mmcif_to_structure(
    path_to_mmcif: str,
    # entry_id: str = "",
    model_num: int = 1,
) -> tuple[Structure | None, DiagnosticBundle, ResultStatus]:
    """Convert raw mmCIF text to a Structure (mmCIF data model).

    Parses the mmCIF file with gemmi, selects the requested model
    (falling back to the first model with a warning if not found),
    and extracts entities, atoms, asym units, connections, assemblies,
    and secondary structure into Pandora's typed `Structure` schema.
    Any other mmCIF categories are preserved verbatim in `Structure.raw`.

    Args:
        path_to_mmcif: Path to the mmCIF file on disk.
        model_num: Which model number to select from a multi-model
            entry.

    Returns:
        A tuple of:
            - The parsed `Structure`, or `None` if parsing failed.
            - A `DiagnosticBundle` of warnings/errors encountered.
            - A `ResultStatus`: "success", "warning", or "failed".
    """
    diag = DiagnosticBundle()

    if not path_to_mmcif.strip():
        diag.errors.append(
            Diagnostic(
                code="EMPTY_CONTENT",
                severity="error",
                message="mmCIF content is empty",
                # entry_id=entry_id or None,
            )
        )
        return None, diag, "failed"

    try:
        st = gemmi.read_structure(path_to_mmcif, format=gemmi.CoorFormat.Mmcif)
    except (OSError, ValueError, RuntimeError) as exc:
        diag.errors.append(
            Diagnostic(
                code="PARSE_ERROR",
                severity="error",
                message=str(exc),
                # entry_id=entry_id or None,
            )
        )
        return None, diag, "failed"

    try:
        doc = gemmi.cif.read(path_to_mmcif)
        block = doc.sole_block()
    except (OSError, ValueError, RuntimeError) as exc:
        diag.errors.append(
            Diagnostic(
                code="CIF_PARSE_ERROR",
                severity="error",
                message=str(exc),
                # entry_id=entry_id or None,
            )
        )
        return None, diag, "failed"

    if len(st) == 0:
        diag.errors.append(
            Diagnostic(
                code="NO_MODEL",
                severity="error",
                message="Structure contains no models",
                # entry_id=entry_id or None,
            )
        )
        return None, diag, "failed"

    eid = st.name

    # Select model --------------------------
    model = next((m for m in st if m.num == model_num), None)
    if model is None:
        model = st[0]
        diag.warnings.append(
            Diagnostic(
                code="MODEL_NOT_FOUND",
                severity="warning",
                message=(
                    f"Model {model_num} not found; falling back "
                    f"to model {model.num}"
                ),
                entry_id=eid,
            )
        )

    # Entry ------------------------------------------
    raw_title = block.find_value("_struct.title")
    entry = EntryRecord(
        id=eid,
        title=_cs(raw_title) if raw_title else None,
    )

    # Subchain to entity lookup -----------------------
    subchain_to_entity: dict[str, str] = {
        sc: ent.name for ent in st.entities for sc in ent.subchains
    }

    # Entity metadata from cif ------------------------------
    entity_desc_map: dict[str, tuple[str | None, float | None, str | None]] = {}
    try:
        for row in block.find(
            "_entity.",
            ["id", "pdbx_description", "formula_weight", "src_method"],
        ):
            entity_desc_map[_cs(row[0]) or ""] = (
                _cs(row[1]),
                _cf(row[2]),
                _cs(row[3]),
            )
    except _MALFORMED_CIF_ERRORS as exc:
        diag.warnings.append(
            Diagnostic(
                code="ENTITY_METADATA_PARSE_ERROR",
                severity="warning",
                message=str(exc),
                entry_id=eid,
            )
        )

    entity_poly_map: dict[str, EntityPolyRecord] = {}
    try:
        for row in block.find(
            "_entity_poly.",
            [
                "entity_id",
                "type",
                "pdbx_seq_one_letter_code",
                "pdbx_seq_one_letter_code_can",
                "pdbx_strand_id",
            ],
        ):
            entity_poly_map[_cs(row[0]) or ""] = EntityPolyRecord(
                type=_cs(row[1]),
                pdbx_seq_one_letter_code=_cseq(row[2]),
                pdbx_seq_one_letter_code_can=_cseq(row[3]),
                pdbx_strand_id=_cs(row[4]),
            )
    except _MALFORMED_CIF_ERRORS as exc:
        diag.warnings.append(
            Diagnostic(
                code="ENTITY_POLY_PARSE_ERROR",
                severity="warning",
                message=str(exc),
                entry_id=eid,
            )
        )

    entities_out: list[EntityRecord] = []
    for ent in st.entities:
        desc, fw, src = entity_desc_map.get(ent.name, (None, None, None))
        entities_out.append(
            EntityRecord(
                id=ent.name,
                type=_ENTITY_TYPE_MAP.get(ent.entity_type, "non-polymer"),
                pdbx_description=desc,
                formula_weight=fw,
                src_method=src,
                poly=entity_poly_map.get(ent.name),
            )
        )

    # Atoms + asym tracking ----------------------------------------------
    asym_auth_map: dict[str, str] = {}  # label_asym_id → auth_asym_id
    atoms_out: list[AtomSiteRecord] = []
    _serial = 0

    for chain in model:
        for res in chain:
            sc = res.subchain
            if sc and sc not in asym_auth_map:
                asym_auth_map[sc] = chain.name

            is_polymer = res.entity_type == gemmi.EntityType.Polymer
            group = "ATOM" if is_polymer else "HETATM"

            try:
                seq_num: int | None = res.label_seq
                ins = res.seqid.icode
                ins_code: str | None = (
                    ins if ins not in (" ", "\x00", "") else None
                )
            except _MALFORMED_CIF_ERRORS:
                seq_num, ins_code = None, None

            for atom in res:
                _serial += 1
                serial = atom.serial if atom.serial > 0 else _serial
                altloc = (
                    atom.altloc
                    if atom.altloc not in ("\x00", " ", "")
                    else None
                )
                atoms_out.append(
                    AtomSiteRecord(
                        group_PDB=group,
                        id=serial,
                        type_symbol=atom.element.name,
                        label_atom_id=atom.name,
                        label_alt_id=altloc,
                        label_comp_id=res.name,
                        label_asym_id=sc or "",
                        label_entity_id=subchain_to_entity.get(sc or "", ""),
                        label_seq_id=seq_num if is_polymer else None,
                        pdbx_PDB_ins_code=ins_code,
                        Cartn_x=atom.pos.x,
                        Cartn_y=atom.pos.y,
                        Cartn_z=atom.pos.z,
                        occupancy=atom.occ,
                        B_iso_or_equiv=atom.b_iso,
                        auth_seq_id=str(res.seqid.num),
                        auth_comp_id=res.name,
                        auth_asym_id=chain.name,
                        auth_atom_id=atom.name,
                        pdbx_PDB_model_num=model.num,
                    )
                )

    if not atoms_out:
        diag.warnings.append(
            Diagnostic(
                code="MISSING_ATOM_SITE",
                severity="warning",
                message="No atoms found in structure",
                entry_id=eid,
            )
        )

    # Asym units ------------------------------------------------
    seen_asyms: set[str] = set()
    asym_units: list[AsymRecord] = []
    for ent in st.entities:
        for sc in ent.subchains:
            if sc not in seen_asyms:
                seen_asyms.add(sc)
                asym_units.append(
                    AsymRecord(
                        id=sc,
                        entity_id=ent.name,
                        auth_id=asym_auth_map.get(sc),
                    )
                )

    # Connections -------------------------------------------
    connections_out: list[ConnRecord] = []
    try:
        for row in block.find(
            "_struct_conn.",
            [
                "id",
                "conn_type_id",
                "ptnr1_label_asym_id",
                "ptnr1_label_comp_id",
                "ptnr1_label_seq_id",
                "ptnr1_label_atom_id",
                "ptnr1_auth_asym_id",
                "ptnr1_auth_seq_id",
                "ptnr1_PDB_ins_code",
                "ptnr1_symmetry",
                "ptnr2_label_asym_id",
                "ptnr2_label_comp_id",
                "ptnr2_label_seq_id",
                "ptnr2_label_atom_id",
                "ptnr2_auth_asym_id",
                "ptnr2_auth_seq_id",
                "ptnr2_PDB_ins_code",
                "ptnr2_symmetry",
                "pdbx_dist_value",
                "details",
            ],
        ):
            connections_out.append(
                ConnRecord(
                    id=_cs(row[0]) or "",
                    conn_type_id=_cs(row[1]) or "covale",
                    ptnr1=ConnPartner(
                        label_asym_id=_cs(row[2]) or "",
                        label_comp_id=_cs(row[3]) or "",
                        label_seq_id=_ci(row[4]),
                        label_atom_id=_cs(row[5]),
                        auth_asym_id=_cs(row[6]),
                        auth_seq_id=_cs(row[7]),
                        pdbx_PDB_ins_code=_cs(row[8]),
                        symmetry=_cs(row[9]),
                    ),
                    ptnr2=ConnPartner(
                        label_asym_id=_cs(row[10]) or "",
                        label_comp_id=_cs(row[11]) or "",
                        label_seq_id=_ci(row[12]),
                        label_atom_id=_cs(row[13]),
                        auth_asym_id=_cs(row[14]),
                        auth_seq_id=_cs(row[15]),
                        pdbx_PDB_ins_code=_cs(row[16]),
                        symmetry=_cs(row[17]),
                    ),
                    pdbx_dist_value=_cf(row[18]),
                    details=_cs(row[19]),
                )
            )
    except _MALFORMED_CIF_ERRORS as exc:
        diag.warnings.append(
            Diagnostic(
                code="CONN_PARSE_ERROR",
                severity="warning",
                message=str(exc),
                entry_id=eid,
            )
        )

    # Assemblies ------------------------------------
    asm_meta: dict[str, dict] = {}
    try:
        for row in block.find(
            "_pdbx_struct_assembly.",
            [
                "id",
                "details",
                "method_details",
                "oligomeric_details",
                "oligomeric_count",
            ],
        ):
            asm_meta[_cs(row[0]) or ""] = {
                "details": _cs(row[1]),
                "method_details": _cs(row[2]),
                "oligomeric_details": _cs(row[3]),
                "oligomeric_count": _ci(row[4]),
            }
    except _MALFORMED_CIF_ERRORS as exc:
        diag.warnings.append(
            Diagnostic(
                code="ASSEMBLY_METADATA_PARSE_ERROR",
                severity="warning",
                message=str(exc),
                entry_id=eid,
            )
        )

    assemblies_out: list[AssemblyRecord] = []
    for asm in st.assemblies:
        meta = asm_meta.get(asm.name, {})
        gens: list[AssemblyGenRecord] = []
        operators: list[AssemblyOperRecord] = []
        seen_ops: set[str] = set()

        for gen in asm.generators:
            gens.append(
                AssemblyGenRecord(
                    assembly_id=asm.name,
                    oper_expression=",".join(op.name for op in gen.operators),
                    asym_id_list=list(gen.subchains),
                )
            )
            for op in gen.operators:
                if op.name not in seen_ops:
                    seen_ops.add(op.name)
                    mat = op.transform.mat
                    vec = op.transform.vec
                    operators.append(
                        AssemblyOperRecord(
                            id=op.name,
                            type=getattr(op, "type", None) or None,
                            matrix=mat.tolist(),
                            vector=[vec.x, vec.y, vec.z],
                        )
                    )

        assemblies_out.append(
            AssemblyRecord(
                id=asm.name,
                details=meta.get("details"),
                method_details=meta.get("method_details"),
                oligomeric_details=meta.get("oligomeric_details"),
                oligomeric_count=meta.get("oligomeric_count"),
                generators=gens,
                operators=operators,
                author_determined=asm.author_determined,
                software_determined=asm.software_determined,
            )
        )

    # Secondary structure --------------------------------
    conf_records: list[ConfRecord] = []
    try:
        for row in block.find(
            "_struct_conf.",
            [
                "id",
                "conf_type_id",
                "beg_label_asym_id",
                "beg_label_seq_id",
                "end_label_asym_id",
                "end_label_seq_id",
                "beg_auth_asym_id",
                "beg_auth_seq_id",
                "end_auth_asym_id",
                "end_auth_seq_id",
            ],
        ):
            conf_records.append(
                ConfRecord(
                    id=_cs(row[0]) or "",
                    conf_type_id=_cs(row[1]) or "",
                    beg_label_asym_id=_cs(row[2]) or "",
                    beg_label_seq_id=_ci(row[3]),
                    end_label_asym_id=_cs(row[4]) or "",
                    end_label_seq_id=_ci(row[5]),
                    beg_auth_asym_id=_cs(row[6]),
                    beg_auth_seq_id=_cs(row[7]),
                    end_auth_asym_id=_cs(row[8]),
                    end_auth_seq_id=_cs(row[9]),
                )
            )
    except _MALFORMED_CIF_ERRORS as exc:
        diag.warnings.append(
            Diagnostic(
                code="STRUCT_CONF_PARSE_ERROR",
                severity="warning",
                message=str(exc),
                entry_id=eid,
            )
        )

    sheet_strands: list[SheetStrandRecord] = []
    try:
        for row in block.find(
            "_struct_sheet_range.",
            [
                "sheet_id",
                "id",
                "beg_label_asym_id",
                "beg_label_seq_id",
                "end_label_asym_id",
                "end_label_seq_id",
                "beg_auth_asym_id",
                "beg_auth_seq_id",
                "end_auth_asym_id",
                "end_auth_seq_id",
            ],
        ):
            sheet_strands.append(
                SheetStrandRecord(
                    sheet_id=_cs(row[0]) or "",
                    id=_cs(row[1]) or "",
                    beg_label_asym_id=_cs(row[2]) or "",
                    beg_label_seq_id=_ci(row[3]),
                    end_label_asym_id=_cs(row[4]) or "",
                    end_label_seq_id=_ci(row[5]),
                    beg_auth_asym_id=_cs(row[6]),
                    beg_auth_seq_id=_cs(row[7]),
                    end_auth_asym_id=_cs(row[8]),
                    end_auth_seq_id=_cs(row[9]),
                )
            )
    except _MALFORMED_CIF_ERRORS as exc:
        diag.warnings.append(
            Diagnostic(
                code="SHEET_RANGE_PARSE_ERROR",
                severity="warning",
                message=str(exc),
                entry_id=eid,
            )
        )

    return (
        Structure(
            entry_id=eid,
            entry=entry,
            entities=entities_out,
            asym_units=asym_units,
            atoms=atoms_out,
            connections=connections_out,
            assemblies=assemblies_out,
            secondary_structure=SSRecord(
                conf_records=conf_records, sheet_strands=sheet_strands
            ),
            raw=_extract_raw(block),
        ),
        diag,
        "warning" if diag.warnings else "success",
    )