tools/bitbucket/src/magpie_bitbucket/normalize.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

"""Normalize Bitbucket Cloud and Data Center responses."""

from __future__ import annotations

from datetime import UTC, datetime
from typing import Any

READ_ONLY_LABELS = ["bitbucket", "read-only", "partial-change-request"]


def repository(kind: str, raw: dict[str, Any]) -> dict[str, Any]:
    """Normalize repository metadata from Bitbucket Cloud or Data Center."""
    if kind == "cloud":
        return {
            "backend": "bitbucket-cloud",
            "id": _string(raw.get("uuid") or raw.get("full_name") or raw.get("slug")),
            "name": raw.get("name"),
            "slug": raw.get("slug"),
            "description": raw.get("description"),
            "is_private": raw.get("is_private"),
            "main_branch": _cloud_main_branch(raw),
            "links": _cloud_links(raw),
            "capabilities": {
                "repository_metadata": "read",
                "pull_requests": "read",
                "issues": "not_implemented",
                "writes": "not_implemented",
            },
            "raw": raw,
        }

    return {
        "backend": "bitbucket-datacenter",
        "id": _string(raw.get("id") or raw.get("slug") or raw.get("name")),
        "name": raw.get("name"),
        "slug": raw.get("slug"),
        "description": raw.get("description"),
        "is_private": _datacenter_private(raw),
        "main_branch": _datacenter_main_branch(raw),
        "links": _datacenter_links(raw),
        "capabilities": {
            "repository_metadata": "read",
            "pull_requests": "read",
            "issues": "not_implemented",
            "writes": "not_implemented",
        },
        "raw": raw,
    }


def pull_request_summary(kind: str, raw: dict[str, Any]) -> dict[str, Any]:
    """Normalize one pull request as a read-only change-request summary."""
    if kind == "cloud":
        return {
            "backend": "bitbucket-cloud",
            "id": _string(raw.get("id")),
            "title": raw.get("title"),
            "author": _cloud_user(raw.get("author")),
            "state": _normalize_state(raw.get("state")),
            "created": _cloud_timestamp(raw.get("created_on")),
            "updated": _cloud_timestamp(raw.get("updated_on")),
            "source": _cloud_branch(raw.get("source")),
            "target": _cloud_branch(raw.get("destination")),
            "permalink": _cloud_link(raw, "html"),
            "labels": READ_ONLY_LABELS,
        }

    return {
        "backend": "bitbucket-datacenter",
        "id": _string(raw.get("id")),
        "title": raw.get("title"),
        "author": _datacenter_user(raw.get("author")),
        "state": _normalize_state(raw.get("state")),
        "created": _epoch_millis_to_iso(raw.get("createdDate")),
        "updated": _epoch_millis_to_iso(raw.get("updatedDate")),
        "source": _datacenter_branch(raw.get("fromRef")),
        "target": _datacenter_branch(raw.get("toRef")),
        "permalink": _datacenter_link(raw),
        "labels": READ_ONLY_LABELS,
    }


def pull_request(kind: str, raw: dict[str, Any]) -> dict[str, Any]:
    """Normalize one pull request as a read-only change-request proposal."""
    summary = pull_request_summary(kind, raw)
    summary["description"] = raw.get("description")
    summary["mergeable"] = "unknown"
    summary["checks"] = "none"
    summary["diff"] = None
    summary["commits"] = None
    summary["raw"] = raw
    return summary


def pull_request_list(kind: str, raw: dict[str, Any]) -> dict[str, Any]:
    """Normalize a Bitbucket pull-request list response."""
    values = raw.get("values")
    if not isinstance(values, list):
        values = []

    return {
        "backend": "bitbucket-cloud" if kind == "cloud" else "bitbucket-datacenter",
        "coverage": "read-only-partial-change-request",
        "pull_requests": [pull_request_summary(kind, item) for item in values if isinstance(item, dict)],
        "raw": raw,
    }


def pull_request_discussion(kind: str, raw: dict[str, Any]) -> dict[str, Any]:
    """Normalize pull request discussion/comments from Bitbucket."""
    values = raw.get("values")
    if not isinstance(values, list):
        values = []

    comments: list[dict[str, Any]] = []
    for item in values:
        if not isinstance(item, dict):
            continue
        if kind == "cloud":
            comments.append(_cloud_comment(item))
        else:
            comments.extend(_datacenter_comment_activity(item))

    return {
        "backend": "bitbucket-cloud" if kind == "cloud" else "bitbucket-datacenter",
        "coverage": "partial-read-only",
        "pull_request_id": _string(raw.get("pull_request_id")),
        "comments": comments,
        "participants": _participants(comments),
        "unresolved_count": None,
        "raw": raw,
    }


def pull_request_status(kind: str, raw: dict[str, Any]) -> dict[str, Any]:
    """Normalize pull request build/status checks from Bitbucket."""
    values = raw.get("values")
    if not isinstance(values, list):
        values = []

    check_details = [
        _cloud_status_check(item) if kind == "cloud" else _datacenter_status_check(item)
        for item in values
        if isinstance(item, dict)
    ]

    return {
        "backend": "bitbucket-cloud" if kind == "cloud" else "bitbucket-datacenter",
        "coverage": "partial-read-only",
        "pull_request_id": _string(raw.get("pull_request_id")),
        "commit": _string(raw.get("commit")),
        "state": _pull_request_state(kind, raw.get("pull_request")),
        "checks": _aggregate_checks(check_details),
        "mergeable": "unknown",
        "check_details": check_details,
        "raw": raw,
    }


def pull_request_commits(kind: str, raw: dict[str, Any]) -> dict[str, Any]:
    """Normalize pull request commits from Bitbucket."""
    values = raw.get("values")
    if not isinstance(values, list):
        values = []

    commits = [
        _cloud_commit(item) if kind == "cloud" else _datacenter_commit(item)
        for item in values
        if isinstance(item, dict)
    ]

    return {
        "backend": "bitbucket-cloud" if kind == "cloud" else "bitbucket-datacenter",
        "coverage": "partial-read-only",
        "pull_request_id": _string(raw.get("pull_request_id")),
        "commits": commits,
        "raw": raw,
    }


def pull_request_diff(kind: str, raw: dict[str, Any]) -> dict[str, Any]:
    """Normalize pull request diff text from Bitbucket."""
    return {
        "backend": "bitbucket-cloud" if kind == "cloud" else "bitbucket-datacenter",
        "coverage": "partial-read-only",
        "pull_request_id": _string(raw.get("pull_request_id")),
        "diff": _string(raw.get("body")) or "",
        "content_type": _string(raw.get("content_type")),
        "url": _string(raw.get("url")),
        "raw": raw,
    }


def pull_request_reviews(kind: str, raw: dict[str, Any]) -> dict[str, Any]:
    """Normalize pull request review-state activity from Bitbucket."""
    pull_request_raw = raw.get("pull_request")
    pull_request_data = pull_request_raw if isinstance(pull_request_raw, dict) else {}

    values = raw.get("values")
    if not isinstance(values, list):
        values = []

    review_events: list[dict[str, Any]] = []
    for item in values:
        if not isinstance(item, dict):
            continue

        event = _cloud_review_event(item) if kind == "cloud" else _datacenter_review_event(item)
        if event is not None:
            review_events.append(event)

    reviewers = (
        _cloud_reviewers(pull_request_data) if kind == "cloud" else _datacenter_reviewers(pull_request_data)
    )

    current_approvals = _current_review_signals(reviewers, "approved")
    current_changes_requested = _current_review_signals(reviewers, "changes_requested")

    latest_events = _latest_review_events(review_events)
    event_approvals = [event for event in latest_events if event.get("kind") == "approval"]
    event_changes_requested = [event for event in latest_events if event.get("kind") == "changes_requested"]

    current_approval_authors = {item.get("author") for item in current_approvals}
    current_change_authors = {item.get("author") for item in current_changes_requested}

    approvals = current_approvals or [
        item for item in event_approvals if item.get("author") not in current_change_authors
    ]
    changes_requested = current_changes_requested or [
        item for item in event_changes_requested if item.get("author") not in current_approval_authors
    ]

    return {
        "backend": "bitbucket-cloud" if kind == "cloud" else "bitbucket-datacenter",
        "coverage": "partial-read-only",
        "pull_request_id": _string(raw.get("pull_request_id")),
        "review_decision": _review_decision_from_signals(
            reviewers, latest_events, approvals, changes_requested
        ),
        "reviewers": reviewers,
        "approvals": approvals,
        "changes_requested": changes_requested,
        "review_requests": _review_requests(reviewers),
        "review_events": review_events,
        "raw": raw,
    }


def _cloud_reviewers(raw: dict[str, Any]) -> list[dict[str, Any]]:
    reviewers = raw.get("reviewers")
    participants = raw.get("participants")
    normalized: dict[str, dict[str, Any]] = {}

    if isinstance(reviewers, list):
        for item in reviewers:
            if isinstance(item, dict):
                reviewer = _cloud_reviewer(item, requested=True)
                key = reviewer.get("user")
                if isinstance(key, str):
                    normalized[key] = reviewer

    if isinstance(participants, list):
        for item in participants:
            if isinstance(item, dict):
                participant = _cloud_reviewer(item, requested=False)
                key = participant.get("user")
                if not isinstance(key, str):
                    continue
                existing = normalized.get(key)
                if existing is None:
                    normalized[key] = participant
                elif (
                    existing.get("review_state") == "pending" and participant.get("review_state") != "unknown"
                ):
                    # A participants[] entry refines a requested reviewer's state,
                    # but only when it carries a definite signal (approved /
                    # changes_requested). Never downgrade a pending request to
                    # "unknown" — that would drop the reviewer from review_requests
                    # and collapse review_decision to "unknown".
                    normalized[key] = participant

    return list(normalized.values())


def _cloud_reviewer(raw: dict[str, Any], *, requested: bool) -> dict[str, Any]:
    state = _cloud_reviewer_state(raw, requested=requested)
    return {
        "user": _cloud_user(raw.get("user") or raw),
        "approved": state == "approved",
        "review_state": state,
        "role": _string(raw.get("role")),
        "raw": raw,
    }


def _cloud_reviewer_state(raw: dict[str, Any], *, requested: bool) -> str:
    if raw.get("approved") is True:
        return "approved"

    state = _string(raw.get("state") or raw.get("status"))
    normalized = state.upper() if state is not None else ""
    if normalized in {"CHANGES_REQUESTED", "NEEDS_WORK"}:
        return "changes_requested"
    if normalized in {"APPROVED"}:
        return "approved"
    if normalized in {"UNAPPROVED", "PENDING", "REVIEW_REQUESTED"}:
        return "pending"

    return "pending" if requested else "unknown"


def _datacenter_reviewers(raw: dict[str, Any]) -> list[dict[str, Any]]:
    reviewers = raw.get("reviewers")
    if not isinstance(reviewers, list):
        return []

    return [_datacenter_reviewer(item) for item in reviewers if isinstance(item, dict)]


def _datacenter_reviewer(raw: dict[str, Any]) -> dict[str, Any]:
    review_state = _datacenter_reviewer_state(raw)

    return {
        "user": _datacenter_user(raw.get("user") or raw),
        "approved": review_state == "approved",
        "status": _string(raw.get("status")),
        "review_state": review_state,
        "role": _string(raw.get("role")),
        "raw": raw,
    }


def _datacenter_reviewer_state(raw: dict[str, Any]) -> str:
    approved = raw.get("approved")
    if approved is True:
        return "approved"

    status = _string(raw.get("status"))
    normalized = status.upper() if status is not None else ""
    if normalized == "APPROVED":
        return "approved"
    if normalized in {"NEEDS_WORK", "CHANGES_REQUESTED"}:
        return "changes_requested"
    if normalized in {"UNAPPROVED", "PENDING", "NOT_APPROVED"}:
        return "pending"

    return "unknown"


def _cloud_review_event(raw: dict[str, Any]) -> dict[str, Any] | None:
    approval = raw.get("approval")
    if isinstance(approval, dict):
        return {
            "kind": "approval",
            "author": _cloud_user(approval.get("user")),
            "date": _cloud_timestamp(approval.get("date")),
            "raw": raw,
        }

    changes_requested = raw.get("changes_requested")
    if isinstance(changes_requested, dict):
        return {
            "kind": "changes_requested",
            "author": _cloud_user(changes_requested.get("user")),
            "date": _cloud_timestamp(changes_requested.get("date")),
            "raw": raw,
        }

    approval_removed = raw.get("approval_removed") or raw.get("unapproval")
    if isinstance(approval_removed, dict):
        return {
            "kind": "approval_removed",
            "author": _cloud_user(approval_removed.get("user")),
            "date": _cloud_timestamp(approval_removed.get("date")),
            "raw": raw,
        }

    update = raw.get("update")
    if isinstance(update, dict):
        return {
            "kind": "updated",
            "author": _cloud_user(update.get("author")),
            "date": _cloud_timestamp(update.get("date")),
            "raw": raw,
        }

    return None


def _datacenter_review_event(raw: dict[str, Any]) -> dict[str, Any] | None:
    action = _string(raw.get("action") or raw.get("type"))
    normalized_action = action.upper() if action is not None else ""

    if normalized_action == "APPROVED":
        return _datacenter_activity_event("approval", raw)
    if normalized_action in {"UNAPPROVED", "APPROVAL_REMOVED"}:
        return _datacenter_activity_event("approval_removed", raw)
    if normalized_action in {"NEEDS_WORK", "CHANGES_REQUESTED"}:
        return _datacenter_activity_event("changes_requested", raw)
    if normalized_action in {"REVIEWED", "UPDATED", "RESCOPED"}:
        return _datacenter_activity_event(normalized_action.lower(), raw)

    return None


def _datacenter_activity_event(kind: str, raw: dict[str, Any]) -> dict[str, Any]:
    return {
        "kind": kind,
        "author": _datacenter_user(raw.get("user")),
        "date": _epoch_millis_to_iso(raw.get("createdDate")),
        "raw": raw,
    }


def _current_review_signals(
    reviewers: list[dict[str, Any]],
    review_state: str,
) -> list[dict[str, Any]]:
    return [
        {
            "kind": review_state,
            "author": reviewer.get("user"),
            "status": reviewer.get("status"),
            "raw": reviewer.get("raw"),
        }
        for reviewer in reviewers
        if reviewer.get("review_state") == review_state
    ]


def _review_decision_from_signals(
    reviewers: list[dict[str, Any]],
    latest_events: Any,
    approvals: list[dict[str, Any]],
    changes_requested: list[dict[str, Any]],
) -> str:
    if changes_requested:
        return "changes_requested"
    if approvals:
        return "approved"
    return _review_decision(reviewers, latest_events)


def _review_decision(
    reviewers: list[dict[str, Any]],
    latest_events: list[dict[str, Any]],
) -> str:
    reviewer_states = {reviewer.get("review_state") for reviewer in reviewers}
    if "changes_requested" in reviewer_states:
        return "changes_requested"
    if "approved" in reviewer_states:
        return "approved"
    if "pending" in reviewer_states:
        return "review_required"

    event_states = {event.get("kind") for event in latest_events}
    if "changes_requested" in event_states:
        return "changes_requested"
    if "approval" in event_states:
        return "approved"
    if "approval_removed" in event_states:
        return "review_required"

    return "unknown"


def _latest_review_events(review_events: list[dict[str, Any]]) -> list[dict[str, Any]]:
    latest: dict[str, tuple[str, int, dict[str, Any]]] = {}
    for index, event in enumerate(review_events):
        kind = event.get("kind")
        if kind not in {"approval", "approval_removed", "changes_requested"}:
            continue

        author = event.get("author")
        if not isinstance(author, str):
            continue

        date = _string(event.get("date")) or ""
        previous = latest.get(author)
        if previous is None or (date, index) >= (previous[0], previous[1]):
            latest[author] = (date, index, event)

    return [item[2] for item in latest.values()]


def _review_requests(reviewers: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return [reviewer for reviewer in reviewers if reviewer.get("review_state") == "pending"]


def _cloud_commit(raw: dict[str, Any]) -> dict[str, Any]:
    return {
        "hash": _string(raw.get("hash")),
        "message": _string(raw.get("message")),
        "author": _cloud_commit_author(raw.get("author")),
        "date": _cloud_timestamp(raw.get("date")),
        "links": _cloud_links(raw),
        "raw": raw,
    }


def _datacenter_commit(raw: dict[str, Any]) -> dict[str, Any]:
    return {
        "hash": _string(raw.get("id") or raw.get("displayId")),
        "display_hash": _string(raw.get("displayId")),
        "message": _string(raw.get("message")),
        "author": _datacenter_commit_author(raw.get("author")),
        "date": _epoch_millis_to_iso(raw.get("authorTimestamp") or raw.get("committerTimestamp")),
        "links": _datacenter_links(raw),
        "raw": raw,
    }


def _cloud_commit_author(raw: object) -> str | None:
    if not isinstance(raw, dict):
        return None

    user = raw.get("user")
    if isinstance(user, dict):
        display_name = _cloud_user(user)
        if display_name:
            return display_name

    raw_author = raw.get("raw")
    if isinstance(raw_author, str):
        return raw_author

    return None


def _datacenter_commit_author(raw: object) -> str | None:
    if isinstance(raw, dict):
        return _datacenter_user(raw)
    if isinstance(raw, str):
        return raw
    return None


def _cloud_status_check(raw: dict[str, Any]) -> dict[str, Any]:
    return {
        "key": _string(raw.get("key")),
        "name": _string(raw.get("name") or raw.get("key")),
        "state": _normalize_check_state(raw.get("state")),
        "url": _string(raw.get("url")),
        "description": _string(raw.get("description")),
        "created": _cloud_timestamp(raw.get("created_on")),
        "updated": _cloud_timestamp(raw.get("updated_on")),
        "raw": raw,
    }


def _datacenter_status_check(raw: dict[str, Any]) -> dict[str, Any]:
    return {
        "key": _string(raw.get("key")),
        "name": _string(raw.get("name") or raw.get("key")),
        "state": _normalize_check_state(raw.get("state")),
        "url": _string(raw.get("url")),
        "description": _string(raw.get("description")),
        "created": _epoch_millis_to_iso(raw.get("dateAdded")),
        "updated": _epoch_millis_to_iso(raw.get("dateUpdated")),
        "raw": raw,
    }


def _aggregate_checks(check_details: list[dict[str, Any]]) -> str:
    states = {check.get("state") for check in check_details}
    if not states:
        return "none"
    if "failure" in states:
        return "failing"
    if "pending" in states:
        return "pending"
    if states == {"success"}:
        return "passing"
    return "pending"


def _pull_request_state(kind: str, raw: object) -> str:
    if not isinstance(raw, dict):
        return "unknown"
    if kind == "cloud":
        return _normalize_state(raw.get("state"))
    return _normalize_state(raw.get("state"))


def _normalize_check_state(value: object) -> str:
    raw_state = _string(value)
    state = raw_state.upper() if raw_state is not None else ""
    if state in {"SUCCESS", "SUCCESSFUL", "PASSED"}:
        return "success"
    if state in {"FAILED", "FAILURE", "ERROR"}:
        return "failure"
    if state in {"INPROGRESS", "IN_PROGRESS", "PENDING"}:
        return "pending"
    if state in {"STOPPED", "CANCELLED", "CANCELED"}:
        return "cancelled"
    return "unknown"


def _cloud_comment(raw: dict[str, Any]) -> dict[str, Any]:
    """Normalize one Bitbucket Cloud pull request comment."""
    body = _content_text(raw.get("content"))
    author = _cloud_user(raw.get("user"))
    created = _cloud_timestamp(raw.get("created_on"))
    updated = _cloud_timestamp(raw.get("updated_on"))

    return {
        "id": _string(raw.get("id")),
        "author": author,
        "date": created,
        "created": created,
        "updated": updated,
        "body": body,
        "kind": "comment",
        "deleted": _bool_or_none(raw.get("deleted")),
        "inline": _cloud_inline(raw.get("inline")),
        "raw": raw,
    }


def _datacenter_comment_activity(raw: dict[str, Any]) -> list[dict[str, Any]]:
    """Normalize comment-bearing Bitbucket Data Center activity, including replies."""
    action = str(raw.get("action") or raw.get("type") or "").upper()
    if action and action != "COMMENTED":
        return []

    comment = raw.get("comment")
    if not isinstance(comment, dict):
        return []

    return _datacenter_comment_tree(comment, raw)


def _datacenter_comment_tree(
    raw: dict[str, Any],
    activity: dict[str, Any],
    parent_id: str | None = None,
) -> list[dict[str, Any]]:
    normalized = _datacenter_comment(raw, activity, parent_id)

    replies: list[dict[str, Any]] = []
    for reply in raw.get("comments") or []:
        if isinstance(reply, dict):
            replies.extend(_datacenter_comment_tree(reply, activity, normalized["id"]))

    return [normalized, *replies]


def _datacenter_comment(
    raw: dict[str, Any],
    activity: dict[str, Any],
    parent_id: str | None,
) -> dict[str, Any]:
    created = _epoch_millis_to_iso(raw.get("createdDate") or activity.get("createdDate"))
    updated = _epoch_millis_to_iso(raw.get("updatedDate") or activity.get("updatedDate"))

    return {
        "id": _string(raw.get("id") or activity.get("id")),
        "parent_id": parent_id,
        "author": _datacenter_user(raw.get("author") or activity.get("user")),
        "date": created,
        "created": created,
        "updated": updated,
        "body": _string(raw.get("text")),
        "kind": "comment",
        "deleted": _bool_or_none(raw.get("deleted")),
        "inline": _datacenter_inline(raw.get("anchor")),
        "raw": raw,
    }


def _participants(comments: list[dict[str, Any]]) -> list[str]:
    """Return sorted unique discussion participants derived from comments."""
    names: set[str] = set()
    for comment in comments:
        author = comment.get("author")
        if isinstance(author, str):
            names.add(author)
    return sorted(names)


def _content_text(raw: object) -> str | None:
    """Extract Bitbucket Cloud raw comment text without truthiness fallback."""
    if not isinstance(raw, dict):
        return None
    for key in ("raw", "markup", "html"):
        if key in raw:
            return _string(raw.get(key))
    return None


def _bool_or_none(value: object) -> bool | None:
    """Normalize optional booleans."""
    return value if isinstance(value, bool) else None


def _cloud_inline(raw: object) -> dict[str, Any] | None:
    """Normalize Bitbucket Cloud inline comment location."""
    if not isinstance(raw, dict):
        return None

    inline: dict[str, Any] = {}
    if isinstance(raw.get("path"), str):
        inline["path"] = raw["path"]
    if isinstance(raw.get("from"), int):
        inline["from_line"] = raw["from"]
    if isinstance(raw.get("to"), int):
        inline["to_line"] = raw["to"]

    return inline or None


def _datacenter_inline(raw: object) -> dict[str, Any] | None:
    """Normalize Bitbucket Data Center inline comment location."""
    if not isinstance(raw, dict):
        return None

    inline: dict[str, Any] = {}
    if isinstance(raw.get("path"), str):
        inline["path"] = raw["path"]
    if isinstance(raw.get("from"), int):
        inline["from_line"] = raw["from"]
    if isinstance(raw.get("to"), int):
        inline["to_line"] = raw["to"]
    if "to_line" not in inline and isinstance(raw.get("line"), int):
        inline["to_line"] = raw["line"]

    return inline or None


def _string(value: object) -> str | None:
    """Convert a value to string while preserving missing values as None."""
    if value is None:
        return None
    return str(value)


def _normalize_state(value: object) -> str:
    """Normalize backend-specific PR states to change-request lifecycle words."""
    state = str(value or "").lower()
    if state in {"open", "opened"}:
        return "open"
    if state in {"merged", "fulfilled"}:
        return "merged"
    if state in {"declined", "superseded"}:
        return "declined"
    return state or "unknown"


def _cloud_timestamp(value: object) -> str | None:
    """Return a Cloud timestamp string when present."""
    return _string(value)


def _epoch_millis_to_iso(value: object) -> str | None:
    """Convert Bitbucket Data Center epoch milliseconds to UTC ISO-8601."""
    if isinstance(value, int | float):
        return datetime.fromtimestamp(value / 1000, tz=UTC).isoformat().replace("+00:00", "Z")
    return _string(value)


def _cloud_main_branch(raw: dict[str, Any]) -> str | None:
    mainbranch = raw.get("mainbranch")
    if isinstance(mainbranch, dict):
        value = mainbranch.get("name")
        return _string(value)
    return _string(mainbranch)


def _cloud_links(raw: dict[str, Any]) -> dict[str, str]:
    links = raw.get("links")
    if not isinstance(links, dict):
        return {}
    normalized: dict[str, str] = {}
    for name, value in links.items():
        if isinstance(value, dict) and isinstance(value.get("href"), str):
            normalized[name] = value["href"]
    return normalized


def _cloud_link(raw: dict[str, Any], name: str) -> str | None:
    links = _cloud_links(raw)
    return links.get(name)


def _cloud_user(raw: object) -> str | None:
    if not isinstance(raw, dict):
        return None
    return _string(raw.get("display_name") or raw.get("nickname") or raw.get("username") or raw.get("uuid"))


def _cloud_branch(raw: object) -> str | None:
    if not isinstance(raw, dict):
        return None
    branch = raw.get("branch")
    if isinstance(branch, dict):
        return _string(branch.get("name"))
    return None


def _datacenter_private(raw: dict[str, Any]) -> bool | None:
    public = raw.get("public")
    if isinstance(public, bool):
        return not public
    return None


def _datacenter_main_branch(raw: dict[str, Any]) -> str | None:
    branch = raw.get("defaultBranch")
    if isinstance(branch, dict):
        return _string(branch.get("displayId") or branch.get("id"))
    return _string(branch)


def _datacenter_links(raw: dict[str, Any]) -> dict[str, str]:
    links = raw.get("links")
    if not isinstance(links, dict):
        return {}

    normalized: dict[str, str] = {}
    for name, value in links.items():
        if isinstance(value, list) and value:
            first = value[0]
            if isinstance(first, dict) and isinstance(first.get("href"), str):
                normalized[name] = first["href"]
    return normalized


def _datacenter_link(raw: dict[str, Any]) -> str | None:
    return _datacenter_links(raw).get("self")


def _datacenter_user(raw: object) -> str | None:
    if not isinstance(raw, dict):
        return None
    user = raw.get("user")
    if isinstance(user, dict):
        return _string(user.get("displayName") or user.get("name") or user.get("emailAddress"))
    return _string(raw.get("displayName") or raw.get("name") or raw.get("emailAddress"))


def _datacenter_branch(raw: object) -> str | None:
    if not isinstance(raw, dict):
        return None
    return _string(raw.get("displayId") or raw.get("id"))