Index

class BuildEnvironment(BaseModel):
    """
    Structured metadata representing the current system and build execution context.

    Captures environmental data such as OS details, hardware specs, and CI presence.
    Utilized to record the provenance of a build artifact for auditing and debugging.

    Example:
        >>> env = BuildEnvironment()
        >>> print(env.os_system)
        'Darwin'
    """

    model_config = ConfigDict(frozen=True)

    # --- System & OS ---
    hostname: str = Field(
        default_factory=socket.gethostname,
        description="The network hostname of the build machine.",
    )
    user: str = Field(
        default_factory=get_user,
        description="The system username executing the build process.",
    )
    os_system: str = Field(
        default_factory=platform.system,
        description="The operating system name (e.g., 'Linux', 'Darwin', 'Windows').",
    )
    os_release: str = Field(
        default_factory=platform.release,
        description="The operating system release version.",
    )
    os_version: str = Field(
        default_factory=platform.version,
        description="The operating system build or release date string.",
    )

    # --- Hardware ---
    cpu_arch: str = Field(
        default_factory=platform.machine,
        description="Hardware architecture of the build machine (e.g., 'x86_64').",
    )
    cpu_cores: int = Field(
        default_factory=lambda: os.cpu_count() or 0,
        description="The number of logical CPU cores available.",
    )
    total_ram_gb: float = Field(
        default_factory=get_ram_gb,
        description="The total available system RAM in gigabytes.",
    )

    # --- Runtime ---
    python_version: str = Field(
        default_factory=platform.python_version,
        description="The version of the Python runtime executing the build.",
    )
    python_implementation: str = Field(
        default_factory=platform.python_implementation,
        description="The specific Python implementation (e.g., 'CPython', 'PyPy').",
    )
    python_compiler: str = Field(
        default_factory=platform.python_compiler,
        description="The compiler string used to build the Python runtime.",
    )
    timestamp: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc),
        description="UTC timestamp when this context was captured.",
    )

    # --- CI Context ---
    is_ci: bool = Field(
        default_factory=lambda: get_ci_info()[0],
        description="True if executing within a recognized CI environment.",
    )
    ci_provider: str | None = Field(
        default_factory=lambda: get_ci_info()[1],
        description="The name of the detected CI provider, or None if undetermined.",
    )

    # --- Path Context ---
    project_root: Path = Field(
        default_factory=Path.cwd,
        description="The root directory of the project where the build was initiated.",
    )

    build_id: str = Field(
        default_factory=lambda: str(uuid.uuid4()),
        description="A unique identifier generated for this specific build execution.",
    )

    def __str__(self) -> str:
        """Return a concise string representation."""

        ci_str = f" [CI: {self.ci_provider}]" if self.is_ci else " [Local]"
        return (
            f"BuildEnvironment({self.os_system} {self.os_release} {self.cpu_arch}, "
            f"Python {self.python_version}, project={self.project_root.name}, "
            f"id={self.build_id}){ci_str}"
        )

    def __repr__(self) -> str:
        """Return a detailed string representation."""
        return (
            f"BuildEnvironment("
            f"hostname={self.hostname!r}, user={self.user!r}, "
            f"os_system={self.os_system!r}, os_release={self.os_release!r}, "
            f"os_version={self.os_version!r}, "
            f"cpu_arch={self.cpu_arch!r}, cpu_cores={self.cpu_cores!r}, "
            f"total_ram_gb={self.total_ram_gb!r}, "
            f"python_version={self.python_version!r}, "
            f"python_implementation={self.python_implementation!r}, "
            f"python_compiler={self.python_compiler!r}, timestamp={self.timestamp!r}, "
            f"is_ci={self.is_ci!r}, ci_provider={self.ci_provider!r}, "
            f"project_root={self.project_root!r}, build_id={self.build_id!r}"
            f")"
        )

class EnsureList(list[TypeVarT], Generic[TypeVarT]):
    """
    A list subclass that tightly integrates with Pydantic Core Schema.

    It preprocesses inputs (splitting strings, normalizing bools) before
    delegating the final strict validation to Pydantic's native schema.

    .. code-block:: python

        from pydantic import BaseModel

        class MyModel(BaseModel):
            items: EnsureList[int]

        model = MyModel(items="1, 2, 3")
        print(model.items)  # [1, 2, 3]
    """

    @classmethod
    def __get_pydantic_core_schema__(
        cls, source_type: Any, handler: GetCoreSchemaHandler
    ) -> CoreSchema:
        """
        Create a schema that hooks a pre-validator into the Pydantic pipeline.

        :param source_type: The original type annotation.
        :param handler: The Pydantic core schema handler.
        :return: The constructed Pydantic core schema.
        """
        # Extract the inner type (e.g., int from EnsureList[int])
        type_arguments = get_args(source_type)
        inner_type: Any = type_arguments[0] if type_arguments else Any

        # Identify if we need 'special' help for types Pydantic is strict about.
        # Standard types like int, float, str are handled natively by Pydantic
        # once the string is split into a list.
        pre_coercer_map: dict[Any, Callable[[Any], Any]] = {
            bool: coerce_bool,
            Path: coerce_path,
        }
        target_pre_coercer = pre_coercer_map.get(inner_type)

        # This schema represents what we WANT the data to look like at the end.
        # By calling handler.generate_schema, we get Pydantic's native list logic.
        final_list_schema = handler.generate_schema(list[inner_type])

        def before_validator_logic(
            value: Any, _info: core_schema.ValidationInfo
        ) -> Any:
            """Preprocessing wrapper before handing off to Pydantic's core."""
            return coerce_list(value, item_pre_coercer=target_pre_coercer)

        # We wrap the native schema in a 'before' validator.
        # Pipeline: Raw Input -> before_validator_logic -> Native Pydantic list[T]
        return core_schema.with_info_before_validator_function(
            before_validator_logic,
            final_list_schema,
            serialization=core_schema.plain_serializer_function_ser_schema(
                list, when_used="json-unless-none"
            ),
        )

class GitReference(BaseModel):
    """
    Pydantic model representing a Git reference (commit, tag, or branch).

    Provides the foundational metadata fields for Git objects and includes
    optional fields for specific types like author information for commits,
    or names for tags and branches.

    .. code-block:: python

        def print_metadata(metadata: GitReference):
            print(f"SHA: {metadata.short_sha}, HEAD: {metadata.is_head_commit}")
    """

    commit_sha: str = Field(
        description="The full, un-abbreviated SHA hash of the commit.",
        default="",
    )
    short_sha: str = Field(
        description="The abbreviated SHA hash of the commit for display purposes.",
        default="",
    )
    timestamp: datetime = Field(
        description="The creation or commit timestamp of the Git object.",
        default=datetime.min,
    )
    distance_from_head: int = Field(
        description="The number of commits between this object and the current HEAD.",
        default=sys.maxsize,
    )
    is_head_commit: bool = Field(
        description="Indicates whether this object represents the current HEAD commit.",
        default=False,
    )
    total_commits: int = Field(
        description="Total number of commits in the repository.",
        default=0,
    )
    author_name: str = Field(
        description="The name of the author who created the commit.", default=""
    )
    author_email: str = Field(
        description="The email address of the author who created the commit.",
        default="",
    )
    commit_message: str = Field(
        description="The full message associated with the commit.", default=""
    )
    tag_name: str = Field(description="The name of the Git tag.", default="")
    branch_name: str = Field(description="The name of the Git branch.", default="")
    is_current_branch: bool = Field(
        description="Indicates whether this branch is currently checked out.",
        default=False,
    )

    def __str__(self) -> str:
        if self.tag_name:
            return f"{self.tag_name} -> {self.short_sha} ({self.timestamp.isoformat()})"
        if self.branch_name:
            marker = "*" if self.is_current_branch else " "
            return (
                f"{marker} {self.branch_name} -> {self.short_sha} "
                f"({self.timestamp.isoformat()})"
            )
        if self.commit_message:
            return (
                f"{self.short_sha} {self.commit_message} - {self.author_name} "
                f"({self.timestamp.isoformat()})"
            )
        return f"{self.short_sha} ({self.timestamp.isoformat()})"

    @model_validator(mode="before")
    @classmethod
    def parse_git_references(cls, data: Any) -> Any:
        """
        Extracts branch and tag metadata from the 'refs' input string.

        Logic identifies the current branch via 'HEAD ->' and extracts
        the most recent tags.
        """
        if not isinstance(data, dict):
            return data

        if "ref_names" in data:
            data["refs"] = data["ref_names"]

        if "refs" not in data:
            return data

        reference_string = data["refs"]
        reference_parts = [part.strip() for part in reference_string.split(",")]
        found_tags = []

        for part in reference_parts:
            # Detect current branch from 'HEAD -> branch_name'
            if "HEAD ->" in part:
                data["branch_name"] = part.replace("HEAD ->", "").strip()
                data["is_current_branch"] = True

            # Detect tags
            elif "tag:" in part:
                tag_content = part.replace("tag:", "").strip()
                found_tags.append(tag_content)

            # Fallback for plain branch names if HEAD was not explicitly indicated
            elif not data.get("branch_name") and not part.startswith("tag:"):
                data["branch_name"] = part

        # The first tag in the ref list is considered the closest/most recent
        if found_tags and not data.get("tag_name"):
            data["tag_name"] = found_tags[0]

        return data

class GitRepository:
    """
    Refined interface for Git operations using Pydantic and safe execution.

    Provides properties and methods to query a Git repository's status, branches, tags,
    and commits. It encapsulates subprocess calls to Git and returns typed
    Pydantic models.

    .. code-block:: python

        repo = GitRepository()
        if repo.is_available:
            print(repo.last_tag.tag_name if repo.last_tag else "No tags found")
    """

    def __init__(
        self,
        repository_path: Path | str | None = None,
    ) -> None:
        """
        Initializes the GitRepository instance.

        :param repository_path: The base path to the Git repository.
            Defaults to the current working directory.
        """
        self.base_path = Path(repository_path or Path.cwd()).resolve()

    def __str__(self) -> str:
        """Return a concise string representation."""
        if not self.is_available:
            return f"GitRepository({self.base_path}) - Unavailable"

        status = "*" if self.is_dirty else ""
        head = self.head_name or "detached"

        current = self.current_commit
        tag = self.last_tag
        branch = self.current_branch

        return (
            f"GitRepository(path={self.base_path!r}, is_available={self.is_available}, "
            f"commit_count={self.commit_count}, is_dirty={self.is_dirty}, "
            f"dirty_files={self.dirty_files}, "
            f"current_commit={current.short_sha if current else None}, "
            f"last_tag={tag.tag_name if tag else None}, "
            f"current_branch={branch.branch_name if branch else None}"
            f") - {head}{status}"
        )

    def __repr__(self) -> str:
        """Return a detailed string representation."""
        return f"GitRepository(base_path={self.base_path!r})"

    @property
    def is_available(self) -> bool:
        """
        Checks if the path is inside a valid git work tree.

        :return: True if the base path is a valid Git repository work tree,
            False otherwise.
        """
        return self._execute_command(["rev-parse", "--is-inside-work-tree"]) == "true"

    @property
    def root_directory(self) -> Path:
        """
        Gets the root directory of the Git repository.

        :return: The absolute path to the root directory of the Git repository.
        :raises NotAGitRepositoryError: If the repository is not valid.
        """
        self._ensure_valid_repository()
        return Path(self._execute_command(["rev-parse", "--show-toplevel"]))

    @property
    def repository_name(self) -> str:
        """
        Gets the name of the Git repository.

        Extracts the repository name from the remote origin URL if available; otherwise,
        falls back to the name of the root directory.

        :return: The string name of the Git repository.
        """
        if remote_url := self.remote_origin_url:
            name = remote_url.split("/")[-1]
            return name[:-4] if name.endswith(".git") else name
        return self.root_directory.name

    @property
    def remote_origin_url(self) -> str:
        """
        Gets the remote origin URL.

        :return: The URL of the remote origin, or an empty string if not configured.
        """
        return self._execute_command(["config", "--get", "remote.origin.url"])

    @property
    def commit_count(self) -> int:
        """
        Gets the total number of commits in the repository.

        :return: The total number of commits.
        """
        if not self.is_available:
            return 0
        try:
            return int(self._execute_command(["rev-list", "--count", "HEAD"]) or 0)
        except ValueError:
            return 0

    @property
    def is_dirty(self) -> bool:
        """
        Checks if the repository has uncommitted changes.

        :return: True if there are uncommitted changes, False otherwise.
        """
        return bool(self.dirty_files)

    @property
    def dirty_files(self) -> list[str]:
        """
        Gets a list of modified files.

        :return: A list of file paths that have uncommitted changes.
        """
        output = self._execute_command(["status", "--porcelain"])
        return [line[3:] for line in output.splitlines() if line]

    @property
    def current_commit(self) -> GitReference | None:
        """
        Gets the most recent commit.

        :return: The most recent GitReference object, or None if no commits exist.
        """
        return next(self.commits, None)

    @property
    def last_tag(self) -> GitReference | None:
        """
        Gets the most recent tag.

        :return: The most recent GitReference object, or None if no tags exist.
        """
        return next(self.tags, None)

    @property
    def current_branch(self) -> GitReference | None:
        """
        Gets the currently checked-out branch.

        :return: The GitReference object representing the current branch,
            or None if detached.
        """
        return next(
            (branch for branch in self.branches if branch.is_current_branch),
            None,
        )

    @property
    def head_name(self) -> str:
        """
        Gets the branch name or short sha of HEAD.

        :return: The current branch name, or the short SHA if in a detached HEAD state.
        """
        if branch := self.current_branch:
            return branch.branch_name
        if current := self.current_commit:
            return current.short_sha
        return ""

    @property
    def commits(self) -> Iterator[GitReference]:
        """
        Yields all commits in the repository.

        :return: An iterator of GitReference objects.
        :raises NotAGitRepositoryError: If the repository is not valid.
        """
        self._ensure_valid_repository()
        total_commits = self.commit_count
        format_string = "%H|%h|%cI|%an|%ae|%s|%D"
        lines = self._stream_command(["log", f"--format={format_string}"])

        for index, line in enumerate(lines):
            parts = line.split("|", 6)
            if len(parts) == 7:  # noqa: PLR2004
                tag_name = ""
                branch_name = ""
                is_current_branch = False

                refs = parts[6].split(", ") if parts[6] else []
                for ref in refs:
                    if ref.startswith("tag: "):
                        tag_name = ref[5:]
                    elif "->" in ref:
                        branch_name = ref.split(" -> ")[1]
                        is_current_branch = True
                    elif (
                        ref
                        and not ref.startswith("origin/")
                        and ref != "HEAD"
                        and not branch_name
                    ):
                        branch_name = ref

                yield GitReference(
                    commit_sha=parts[0],
                    short_sha=parts[1],
                    timestamp=datetime.fromisoformat(parts[2].replace("Z", "+00:00")),
                    author_name=parts[3],
                    author_email=parts[4],
                    commit_message=parts[5],
                    tag_name=tag_name,
                    branch_name=branch_name,
                    is_current_branch=is_current_branch,
                    distance_from_head=index,
                    is_head_commit=(index == 0),
                    total_commits=total_commits,
                )

    @property
    def tags(self) -> Iterator[GitReference]:
        """
        Yields all tags in the repository.

        :return: An iterator of GitReference objects.
        :raises NotAGitRepositoryError: If the repository is not valid.
        """
        self._ensure_valid_repository()
        current = self.current_commit
        head_sha = current.commit_sha if current else ""
        total_commits = self.commit_count
        format_string = "%(refname:short)|%(creatordate:iso-strict)|%(objectname)"

        lines = self._stream_command(
            [
                "for-each-ref",
                "--sort=-creatordate",
                f"--format={format_string}",
                "refs/tags/",
            ]
        )

        for line in lines:
            name, date_str, sha = line.split("|")
            distance_str = self._execute_command(
                ["rev-list", "--count", f"{sha}..HEAD"]
            )
            yield GitReference(
                tag_name=name,
                commit_sha=sha,
                short_sha=sha[:7],
                timestamp=datetime.fromisoformat(date_str.replace("Z", "+00:00")),
                distance_from_head=int(distance_str or 0),
                is_head_commit=(sha == head_sha),
                total_commits=total_commits,
            )

    @property
    def branches(self) -> Iterator[GitReference]:
        """
        Yields all branches in the repository.

        :return: An iterator of GitReference objects.
        :raises NotAGitRepositoryError: If the repository is not valid.
        """
        self._ensure_valid_repository()
        current = self.current_commit
        head_sha = current.commit_sha if current else ""
        total_commits = self.commit_count
        format_string = (
            "%(refname:short)|%(objectname)|%(HEAD)|%(committerdate:iso-strict)"
        )

        lines = self._stream_command(
            [
                "for-each-ref",
                f"--format={format_string}",
                "refs/heads/",
                "refs/remotes/",
            ]
        )

        for line in lines:
            name, sha, current_marker, date_str = line.split("|")
            yield GitReference(
                branch_name=name,
                commit_sha=sha,
                short_sha=sha[:7],
                timestamp=datetime.fromisoformat(date_str.replace("Z", "+00:00")),
                distance_from_head=0,
                is_head_commit=(sha == head_sha),
                is_current_branch=(current_marker == "*"),
                total_commits=total_commits,
            )

    def _stream_command(self, arguments: list[str]) -> Iterator[str]:
        """Executes a git command and streams output line by line."""
        full_command = ["git", *arguments]
        try:
            with subprocess.Popen(  # noqa: S603
                full_command, cwd=self.base_path, stdout=subprocess.PIPE, text=True
            ) as process:
                if process.stdout:
                    for line in process.stdout:
                        if clean_line := line.strip():
                            yield clean_line
        except (subprocess.CalledProcessError, FileNotFoundError, OSError) as error:
            logger.debug(f"Command '{shlex.join(full_command)}' failed: {error}")

    def _execute_command(self, arguments: list[str]) -> str:
        """Executes a git command with standardized error handling."""
        full_command = ["git", *arguments]
        try:
            return subprocess.run(  # noqa: S603
                full_command,
                cwd=self.base_path,
                capture_output=True,
                text=True,
                check=True,
            ).stdout.strip()
        except (subprocess.CalledProcessError, FileNotFoundError, OSError) as error:
            logger.debug(f"Command '{shlex.join(full_command)}' failed: {error}")
            return ""

    def _ensure_valid_repository(self) -> None:
        """Raises an error if the directory is not a Git repository."""
        if not self.is_available:
            raise NotAGitRepositoryError(
                f"Path '{self.base_path}' is not a Git repository."
            )

class NotAGitRepositoryError(Exception):
    """
    Raised when the directory is not a valid Git repository.

    This exception is raised when Git operations are attempted on a directory that
    is not part of a valid Git work tree.
    """