#!/usl/bin/env python3
# Libervia plugin to handle Git repositories
# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org)
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
import re
from enum import StrEnum, auto
from pathlib import Path
from typing import Iterator, TYPE_CHECKING, Literal, cast
from pydantic import BaseModel, ConfigDict, Field, RootModel, field_validator
from twisted.internet import defer
from twisted.python.procutils import which
from libervia.backend.core import exceptions
from libervia.backend.core.constants import Const as C
from libervia.backend.core.core_types import SatXMPPEntity
from libervia.backend.core.i18n import _
from libervia.backend.core.log import getLogger
from libervia.backend.plugins.plugin_forge import Forge
from libervia.backend.tools import utils
from libervia.backend.tools.common import async_process
if TYPE_CHECKING:
from libervia.backend.core.main import LiberviaBackend
log = getLogger(__name__)
# Git formatting, cf.: https://git-scm.com/docs/git-log#_pretty_formats
GIT_LOG_SEPARATOR = "%x00"
GIT_LOG_FORMAT = GIT_LOG_SEPARATOR.join(
[
# Full commit hash
"%H",
# Short commit hash
"%h",
# Author name
"%an",
# Author email
"%ae",
# Committer date in UNIX timestamp
"%ct",
# Subject (first line of commit message)
"%s",
# Body (rest of commit message)
"%b",
]
)
GIT_LOG_FORMAT += "%x1E"
# Parse hunk header: "@@ -old_start,old_lines +new_start,new_lines @@"
HUNK_MATCH_RE = re.compile(r"@@\s+-(\d+)(?:,(\d+))?\s+\+(\d+)(?:,(\d+))?\s+@@")
class CommitRequest(BaseModel):
"""Model to specify the commits to retrieve."""
model_config = ConfigDict(extra="forbid")
after: str | None = Field(
default=None, description="Get commits after this reference."
)
before: str | None = Field(
default=None, description="Get commits before this reference."
)
limit: int = Field(default=20, description="Maximum number of commits to retrieve.")
class Commit(BaseModel):
"""Information about a Git commit for history display"""
id: str = Field(description="Commit hash (SHA)")
short_id: str = Field(description="Short commit hash (7 characters)")
author_name: str = Field(description="Name of the author")
author_email: str = Field(description="Email of the author")
commit_ts: int = Field(description="UNIX timestamp of the commit")
message_title: str = Field(description="First line of the commit message")
message: str = Field(description="Commit message")
class Commits(RootModel):
root: list[Commit]
def __iter__(self) -> Iterator[Commit]: # type: ignore
return iter(self.root)
def __getitem__(self, item) -> Commit:
return self.root[item]
def __len__(self) -> int:
return len(self.root)
def append(self, item: Commit) -> None:
self.root.append(item)
def sort(self, key=None, reverse=False) -> None:
self.root.sort(key=key, reverse=reverse) # type: ignore
class File(BaseModel):
"""Metadata for a file or directory in a Git repository"""
name: str
type: Literal["file", "directory"]
hash: str
mode: int
size: int | None = None
class Files(RootModel):
root: list[File]
def __iter__(self) -> Iterator[File]: # type: ignore
return iter(self.root)
def __getitem__(self, item) -> str:
return self.root[item]
def __len__(self) -> int:
return len(self.root)
def append(self, item: File) -> None:
self.root.append(item)
def sort(self, key=None, reverse=False) -> None:
self.root.sort(key=key, reverse=reverse) # type: ignore
class FileStatus(StrEnum):
"""Status of a file in a diff"""
ADDED = auto()
DELETED = auto()
MODIFIED = auto()
RENAMED = auto()
COPIED = auto()
TYPE_CHANGED = auto()
UNTRACKED = auto()
class DiffHunk(BaseModel):
"""A single hunk in a diff"""
model_config = ConfigDict(extra="forbid")
old_start: int = Field(description="Starting line number in old file")
old_lines: int = Field(description="Number of lines in old file")
new_start: int = Field(description="Starting line number in new file")
new_lines: int = Field(description="Number of lines in new file")
lines: list[str] = Field(description="Lines in the hunk (with +/- prefix)")
class DiffFile(BaseModel):
"""A file changed in a commit"""
model_config = ConfigDict(extra="forbid")
old_path: str = Field(description="Path of the file before changes")
new_path: str = Field(description="Path of the file after changes")
old_mode: str = Field(description="File mode before changes (e.g., 100644)")
new_mode: str = Field(description="File mode after changes (e.g., 100644)")
old_hash: str | None = Field(
default=None, description="Git hash before changes (None for new files)"
)
new_hash: str | None = Field(
default=None, description="Git hash after changes (None for deleted files)"
)
status: FileStatus = Field(description="File status")
hunks: list[DiffHunk] = Field(default_factory=list, description="Diff hunks")
class DiffFiles(RootModel):
root: list[DiffFile]
def __iter__(self) -> Iterator[DiffFile]: # type: ignore
return iter(self.root)
def __getitem__(self, item) -> DiffFile:
return self.root[item]
def __len__(self) -> int:
return len(self.root)
def append(self, item: DiffFile) -> None:
self.root.append(item)
class CommitDiff(BaseModel):
"""Diff information for a single commit"""
model_config = ConfigDict(extra="forbid")
commit: Commit = Field(description="Commit information")
files: DiffFiles = Field(description="Files changed in this commit")
class CommitDiffs(RootModel):
root: list[CommitDiff]
def __iter__(self) -> Iterator[CommitDiff]: # type: ignore
return iter(self.root)
def __getitem__(self, item) -> CommitDiff:
return self.root[item]
def __len__(self) -> int:
return len(self.root)
def append(self, item: CommitDiff) -> None:
self.root.append(item)
class CommitDiffRequest(BaseModel):
"""Model to specify the commits to retrieve diffs for.
Supports different use cases:
- Single commit: provide commit_hash only
- Commit range: provide after and before
- Merge request: provide merge_request ref to get all commits in the MR
- Branch history: provide before only to get commits up to a branch
"""
model_config = ConfigDict(extra="forbid")
commit_hash: str | None = Field(
default=None, description="Single commit hash to get diff for."
)
merge_request: str | None = Field(
default=None,
description='Merge request ref (e.g., "mr/42") to get all commits.',
)
after: str | None = Field(
default=None,
description="Get commits after this reference (e.g., base branch for MR).",
)
before: str | None = Field(
default=None,
description="Get commits before this reference (e.g., MR branch or commit).",
)
limit: int = Field(default=50, description="Maximum number of commits to retrieve.")
@field_validator("commit_hash")
@classmethod
def validate_commit_hash_exclusive(cls, v: str | None, info):
"""commit_hash cannot be used with other fields."""
if v is not None:
data = info.data
conflicting_fields = ["merge_request", "after", "before"]
if any(data.get(field) is not None for field in conflicting_fields):
raise ValueError(
"commit_hash cannot be used together with"
f" {', '.join(conflicting_fields)}"
)
return v
@field_validator("merge_request")
@classmethod
def validate_merge_request_exclusive(cls, v: str | None, info):
"""merge_request is a standalone option."""
if v is not None:
data = info.data
conflicting_fields = ["after", "before"]
if any(data.get(field) is not None for field in conflicting_fields):
raise ValueError(
"merge_request cannot be used together with after/before range"
)
return v
class InfoRequest(BaseModel):
"""Model to indicate which infos a requested in an info_get request."""
model_config = ConfigDict(extra="forbid")
branches: bool = True
last_commit: bool = False
class InfoResult(BaseModel):
branches: list[str] = Field(default_factory=list)
last_commit: Commit | None = None
PLUGIN_INFO = {
C.PI_NAME: "Git Repository Handler",
C.PI_IMPORT_NAME: "GIT_REPOS",
C.PI_TYPE: C.PLUG_TYPE_MISC,
C.PI_DEPENDENCIES: ["FORGE"],
C.PI_MAIN: "GitHandler",
C.PI_HANDLER: "no",
C.PI_DESCRIPTION: _("""Handle Git Repositories"""),
}
class GitProtocol(async_process.CommandProtocol):
"""handle git commands"""
name = "Git"
command = None
@classmethod
def run(cls, path: Path, command: str, *args, **kwargs) -> defer.Deferred[bytes]:
"""Create a new MercurialRegisterProtocol and execute the given mercurial command.
@param path: path to the repository
@param command: git command to run
@return: stdout of the command
"""
assert "path" not in kwargs
kwargs["path"] = path
d = super().run(command, *args, **kwargs)
d.addErrback(utils.logError)
return d
class GitHandler:
def __init__(self, host: "LiberviaBackend") -> None:
log.info(f"Plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization.")
try:
GitProtocol.command = which("git")[0]
except IndexError:
raise exceptions.NotFound(
_("Git executable not found, can't use Git handler.")
)
self.host = host
self._f = cast(Forge, host.plugins["FORGE"])
host.bridge.add_method(
"git_files_get",
".plugin",
in_sign="sssss",
out_sign="s",
method=self._files_get,
async_=True,
)
host.bridge.add_method(
"git_infos_get",
".plugin",
in_sign="ssss",
out_sign="s",
method=self._infos_get,
async_=True,
)
host.bridge.add_method(
"git_file_content_get",
".plugin",
in_sign="sssss",
out_sign="s",
method=self._git_file_content_get,
async_=True,
)
host.bridge.add_method(
"git_history_get",
".plugin",
in_sign="ssss",
out_sign="s",
method=self._git_history_get,
async_=True,
)
host.bridge.add_method(
"git_diff_get",
".plugin",
in_sign="ssss",
out_sign="s",
method=self._git_diff_get,
async_=True,
)
def get_path_from_repository(self, target_profile: str, repo: str) -> Path:
try:
profile_repositories = self._f.settings.repositories[target_profile]
repo_data = next(
r for r in profile_repositories if r.name.lower() == repo.lower()
)
except (KeyError, StopIteration):
raise exceptions.NotFound(f"Repository Unknown: {repo!r}")
else:
return repo_data.path
def _files_get(
self,
target_profile: str,
repository: str,
branch: str,
path: str,
profile_key: str,
) -> defer.Deferred[str]:
client = self.host.get_client(profile_key)
repository_path = self.get_path_from_repository(target_profile, repository)
d = defer.ensureDeferred(
self.files_get(
client,
repository_path,
branch,
path,
)
)
d.addCallback(lambda m: m.model_dump_json())
return d
async def files_get(
self,
client: SatXMPPEntity,
repository: Path,
branch: str = "main",
path: str = "",
) -> Files:
"""List files from a repository branch and path
@param client: Client instance.
@param repository: Repository to list files from.
@param branch: Branch of the repository to use.
@param path: Path to list.
If not set, list root files.
@return: files listing.
"""
output = await GitProtocol.run(repository, "ls-tree", "-l", f"{branch}:{path}")
files = Files([])
for line in output.splitlines():
mode, type_, hash_, size, name = line.split(maxsplit=4)
size_int = int(size) if size != b"-" else None
file = File(
name=name.decode(),
type="file" if type_ == b"blob" else "directory",
hash=hash_.decode(),
mode=int(mode),
size=size_int,
)
files.append(file)
files.sort(key=lambda f: (f.type != "directory", f.name.lower()))
return files
def _infos_get(
self, target_profile: str, repository: str, info_request_s: str, profile_key: str
) -> defer.Deferred[str]:
client = self.host.get_client(profile_key)
repository_path = self.get_path_from_repository(target_profile, repository)
info_request = InfoRequest.model_validate_json(info_request_s)
d = defer.ensureDeferred(self.infos_get(client, repository_path, info_request))
d.addCallback(lambda info_result: info_result.model_dump_json())
d = cast(defer.Deferred[str], d)
return d
async def infos_get(
self, client: SatXMPPEntity, repository: Path, info_request: InfoRequest
) -> InfoResult:
"""Get information about a Git repository.
@param client: Client instance.
@param repository: Path to the repository.
@param info_request: Which information to retrieve.
@return: Informations about the repository.
"""
branches = []
last_commit = None
if info_request.branches:
try:
output = await GitProtocol.run(repository, "branch")
output_str = output.decode()
lines = output_str.splitlines()
branches = [line.strip().lstrip("* ").strip() for line in lines]
except Exception as e:
log.warning(f"Could not retrieve branches for {repository}: {e}")
if info_request.last_commit:
try:
output = await GitProtocol.run(
repository, "log", "-1", f"--format={GIT_LOG_FORMAT}"
)
commits = self._parse_git_log_output(output)
if commits:
last_commit = commits[0]
except Exception as e:
log.warning(f"Could not retrieve last commit data for {repository}: {e}")
return InfoResult(branches=branches, last_commit=last_commit)
def _git_file_content_get(
self,
target_profile: str,
repository: str,
branch: str,
file_path: str,
profile_key: str,
) -> defer.Deferred[str]:
client = self.host.get_client(profile_key)
repository_path = self.get_path_from_repository(target_profile, repository)
d = defer.ensureDeferred(
self.git_file_content_get(client, repository_path, branch, file_path)
)
d.addCallback(lambda content: content)
return d
async def git_file_content_get(
self,
client: SatXMPPEntity,
repository: Path,
branch: str,
file_path: str,
) -> str:
"""Get the content of a file from a Git repository.
@param client: Client instance.
@param repository: Path to the repository.
@param branch: Branch of the repository to use.
@param file_path: Path to the file within the repository.
@return: Content of the file as a string.
@raise exceptions.NotFound: If the file does not exist.
"""
output = await GitProtocol.run(repository, "show", f"{branch}:{file_path}")
return output.decode()
def _git_history_get(
self,
target_profile: str,
repository: str,
commit_request_s: str,
profile_key: str,
) -> defer.Deferred[str]:
client = self.host.get_client(profile_key)
repository_path = self.get_path_from_repository(target_profile, repository)
commit_request = CommitRequest.model_validate_json(commit_request_s)
d = defer.ensureDeferred(
self.git_history_get(client, repository_path, commit_request)
)
d.addCallback(lambda commits: commits.model_dump_json())
d = cast(defer.Deferred[str], d)
return d
@staticmethod
def _parse_commit_data(commit_bytes: bytes) -> Commit:
"""Parse commit data from bytes into a Commit object.
@param commit_bytes: Raw commit data.
@return: Parsed Commit object.
@raise exceptions.DataError: If the data doesn't have exactly 7 fields.
"""
commit_data = commit_bytes.decode().split("\x00")
if len(commit_data) != 7:
raise exceptions.DataError(
"Failed to parse commit information (expected 7 fields, got"
f" {len(commit_data)})"
)
return Commit(
id=commit_data[0],
short_id=commit_data[1],
author_name=commit_data[2],
author_email=commit_data[3],
commit_ts=int(commit_data[4]),
message_title=commit_data[5],
message=commit_data[6],
)
@staticmethod
def _parse_git_log_output(output: bytes) -> Commits:
"""Parse git log output into Commit objects.
@param output: Raw output from git log command.
@return: Commits object containing parsed commit data.
"""
output_str = output.decode().strip()
if not output_str:
return Commits([])
commits = Commits([])
commit_blocks = output_str.split("\x1e")
for block in commit_blocks:
try:
commit = GitHandler._parse_commit_data(block.strip().encode())
commits.append(commit)
except exceptions.DataError:
log.warning(f"Error while parsing commit, ignoring:\n{block!r}")
continue
return commits
async def git_history_get(
self, client: SatXMPPEntity, repository: Path, commit_request: CommitRequest
) -> Commits:
"""Get the commit history of a repository.
@param client: Client instance.
@param repository: Path to the repository.
@param commit_request: Requested commit and limit.
@return: List of commits in reverse chronological order.
"""
extra = []
match (commit_request.after, commit_request.before):
case (after, before) if after is not None and before is not None:
extra.append(f"{after}..{before}^")
case (_, before) if before is not None:
extra.append(f"{before}^")
case (after, _) if after is not None:
# Git can't retrieve immediate children, so we first need to get commits
# IDs until the limit.
commit_ids = (
(await GitProtocol.run(repository, "rev-list", f"{after}.."))
.decode()
.split()
)
if not commit_ids:
return Commits([])
elif len(commit_ids) >= commit_request.limit:
upper_limit = commit_ids[-commit_request.limit]
else:
upper_limit = commit_ids[0]
extra.append(f"{after}..{upper_limit}")
output = await GitProtocol.run(
repository,
"log",
f"-{commit_request.limit}",
f"--format={GIT_LOG_FORMAT}",
*extra,
)
return self._parse_git_log_output(output)
@staticmethod
def _parse_git_diff_output(output: bytes) -> tuple[Commit, DiffFiles]:
"""Parse git show output into model instances.
@param output: Raw output from git show.
@return: Commit and DiffFiles model instances.
"""
try:
commit_part, diff_part = output.split(b"\x1e", maxsplit=1)
except ValueError:
raise exceptions.InternalError(
"Failed to parse commit information: missing separator"
)
commit = GitHandler._parse_commit_data(commit_part)
# Parse diff files (skip the \n\n after the terminator)
diff_str = diff_part.decode().lstrip("\n")
diff_lines = diff_str.splitlines()
diff_files = DiffFiles([])
current_file: DiffFile | None = None
current_hunk: DiffHunk | None = None
for line in diff_lines:
if line.startswith("diff --git"):
# Save previous file if exists
if current_file is not None:
if current_hunk is not None:
current_file.hunks.append(current_hunk)
diff_files.append(current_file)
# Parse new file header
# Format: diff --git a/old_path b/new_path
parts = line.split()
if len(parts) >= 4:
old_path = parts[2].removeprefix("a/").removeprefix("b/")
new_path = parts[3].removeprefix("a/").removeprefix("b/")
else:
old_path = ""
new_path = ""
current_file = DiffFile(
old_path=old_path,
new_path=new_path,
old_mode="",
new_mode="",
old_hash=None,
new_hash=None,
status=FileStatus.MODIFIED,
hunks=[],
)
current_hunk = None
elif current_file is None:
# Skip lines before any diff --git header
continue
elif line.startswith("old mode "):
current_file.old_mode = line.split()[1]
elif line.startswith("new mode "):
current_file.new_mode = line.split()[1]
elif line.startswith("deleted file mode "):
current_file.old_mode = line.split()[1]
current_file.status = FileStatus.DELETED
elif line.startswith("new file mode "):
current_file.new_mode = line.split()[1]
current_file.status = FileStatus.ADDED
elif line.startswith("index "):
# Parse hashes from index line: "index .. "
idx_parts = line.split()[1:]
if idx_parts:
hashes = idx_parts[0].split("..")
if len(hashes) == 2:
current_file.old_hash = (
hashes[0] if hashes[0] != "0000000" else None
)
current_file.new_hash = (
hashes[1] if hashes[1] != "0000000" else None
)
elif line.startswith("similarity index"):
# Renamed or copied file
current_file.status = (
FileStatus.RENAMED if "rename" in line else FileStatus.COPIED
)
elif line.startswith("rename from"):
current_file.status = FileStatus.RENAMED
current_file.old_path = line.split()[2]
elif line.startswith("rename to"):
current_file.new_path = line.split()[2]
elif line.startswith("copy from"):
current_file.old_path = line.split()[2]
elif line.startswith("copy to"):
current_file.new_path = line.split()[2]
elif line.startswith("--- "):
# Update paths from unified diff header if not already set
path = line[4:].strip()
if not path.startswith("dev/null"):
if not (
current_file.old_path and current_file.status == FileStatus.ADDED
):
if not current_file.old_path:
current_file.old_path = path.rstrip("\t...")
elif line.startswith("+++ "):
# Update paths from unified diff header if not already set
path = line[4:].strip()
if not path.startswith("dev/null"):
if not (
current_file.new_path
and current_file.status == FileStatus.DELETED
):
if not current_file.new_path:
current_file.new_path = path.rstrip("\t...")
elif line.startswith("@@"):
# New hunk
if current_hunk is not None:
current_file.hunks.append(current_hunk)
hunk_header = line.strip()
hunk_match = HUNK_MATCH_RE.match(hunk_header)
if hunk_match:
old_start = int(hunk_match.group(1))
old_lines = int(hunk_match.group(2)) if hunk_match.group(2) else 1
new_start = int(hunk_match.group(3))
new_lines = int(hunk_match.group(4)) if hunk_match.group(4) else 1
current_hunk = DiffHunk(
old_start=old_start,
old_lines=old_lines,
new_start=new_start,
new_lines=new_lines,
lines=[],
)
else:
log.warning(f"Failed to parse hunk header: {line!r}")
elif current_hunk is not None:
# Diff line (must be last case to catch +, -, or space prefixed lines)
if line.startswith(("+", "-", " ")):
current_hunk.lines.append(line)
elif line == "":
# Empty line in diff
if current_hunk.lines and not current_hunk.lines[-1].startswith(
("+", "-", " ")
):
current_hunk.lines.append(" ")
# Save last file and hunk
if current_file is not None:
if current_hunk is not None:
current_file.hunks.append(current_hunk)
diff_files.append(current_file)
return commit, diff_files
def _git_diff_get(
self,
target_profile: str,
repository: str,
commit_diff_request_s: str,
profile_key: str,
) -> defer.Deferred[str]:
client = self.host.get_client(profile_key)
repository_path = self.get_path_from_repository(target_profile, repository)
commit_diff_request = CommitDiffRequest.model_validate_json(commit_diff_request_s)
d = defer.ensureDeferred(
self.git_diff_get(client, repository_path, commit_diff_request)
)
d.addCallback(lambda diffs: diffs.model_dump_json())
d = cast(defer.Deferred[str], d)
return d
async def git_diff_get(
self,
client: SatXMPPEntity,
repository: Path,
commit_diff_request: CommitDiffRequest,
) -> CommitDiffs:
"""Get the diff for commits in a repository.
@param client: Client instance.
@param repository: Path to the repository.
@param commit_diff_request: Requested commits and limit.
@return: List of commit diffs in reverse chronological order.
"""
# Handle single commit hash
if commit_diff_request.commit_hash is not None:
try:
output = await GitProtocol.run(
repository,
"show",
"-m",
f"--format={GIT_LOG_FORMAT}",
commit_diff_request.commit_hash,
)
parsed_commit, diff_files = self._parse_git_diff_output(output)
return CommitDiffs([CommitDiff(commit=parsed_commit, files=diff_files)])
except Exception as e:
log.warning(
"Could not retrieve diff for commit"
f" {commit_diff_request.commit_hash}: {e}"
)
return CommitDiffs([])
# Handle merge request ref (e.g., "mr/42")
if commit_diff_request.merge_request is not None:
try:
# Get all commits in the merge request branch
output = await GitProtocol.run(
repository,
"log",
f"-{commit_diff_request.limit}",
f"--format={GIT_LOG_FORMAT}",
commit_diff_request.merge_request,
)
commits = self._parse_git_log_output(output)
except Exception as e:
log.warning(
"Could not retrieve commits for merge request"
f" {commit_diff_request.merge_request}: {e}"
)
return CommitDiffs([])
else:
# Handle commit range (after..before) or branch history (before only)
# Convert CommitDiffRequest to CommitRequest for git_history_get
commit_request = CommitRequest(
after=commit_diff_request.after,
before=commit_diff_request.before,
limit=commit_diff_request.limit,
)
commits = await self.git_history_get(client, repository, commit_request)
if not commits:
return CommitDiffs([])
commit_diffs = CommitDiffs([])
for commit in commits:
try:
output = await GitProtocol.run(
repository, "show", "-m", f"--format={GIT_LOG_FORMAT}", commit.id
)
parsed_commit, diff_files = self._parse_git_diff_output(output)
commit_diff = CommitDiff(commit=parsed_commit, files=diff_files)
commit_diffs.append(commit_diff)
except Exception as e:
log.warning(f"Could not retrieve diff for commit {commit.id}: {e}")
continue
return commit_diffs