159 lines
4.7 KiB
Python
159 lines
4.7 KiB
Python
import json
|
|
import re
|
|
import ssl
|
|
import urllib.request as request
|
|
from typing import Dict, List
|
|
|
|
import git
|
|
from config.config import parse_config
|
|
from crewai_tools import tool
|
|
|
|
config = parse_config()
|
|
|
|
|
|
@tool("Get Categorized Git Commits Since Last Tag")
|
|
def get_categorized_commits_since_last_tag(
|
|
repo_path: str = config["GIT_REPO_PATH"],
|
|
) -> str:
|
|
"""
|
|
Retrieves and categorizes commits since the last tag in the Git repository using git shortlog.
|
|
|
|
Args:
|
|
repo_path (str): Path to the Git repository.
|
|
|
|
Returns:
|
|
str: JSON string containing categorized commit information.
|
|
"""
|
|
repo = git.Repo(repo_path)
|
|
tags = sorted(repo.tags, key=lambda t: t.commit.committed_datetime)
|
|
last_tag = tags[-1] if tags else None
|
|
|
|
if not last_tag:
|
|
range_spec = "HEAD"
|
|
else:
|
|
range_spec = f"{last_tag.name}..HEAD"
|
|
|
|
shortlog = repo.git.shortlog("-sne", range_spec)
|
|
|
|
author_commits = parse_shortlog(shortlog)
|
|
|
|
commits = list(repo.iter_commits(range_spec))
|
|
|
|
categorized_commits = categorize_commits(commits)
|
|
|
|
result = {
|
|
"last_tag": last_tag.name if last_tag else "No tags",
|
|
"author_commits": author_commits,
|
|
"categorized_commits": categorized_commits,
|
|
}
|
|
|
|
return json.dumps(result, indent=2)
|
|
|
|
|
|
def parse_shortlog(shortlog: str) -> List[Dict[str, str]]:
|
|
"""Parse git shortlog output."""
|
|
author_commits = []
|
|
for line in shortlog.split("\n"):
|
|
if line.strip():
|
|
count, author = line.strip().split("\t")
|
|
author_commits.append(
|
|
{"commit_count": int(count.strip()), "author": author.strip()}
|
|
)
|
|
return author_commits
|
|
|
|
|
|
def categorize_commits(commits: List[git.Commit]) -> Dict[str, List[Dict[str, str]]]:
|
|
"""Categorize commits based on their message."""
|
|
categories = {"features": [], "fixes": [], "documentation": [], "others": []}
|
|
|
|
for commit in commits:
|
|
commit_info = {
|
|
"hash": commit.hexsha[:7],
|
|
"message": commit.summary,
|
|
"date": commit.committed_datetime.isoformat(),
|
|
}
|
|
|
|
if commit.summary.startswith("feat:"):
|
|
categories["features"].append(commit_info)
|
|
elif commit.summary.startswith("fix:"):
|
|
categories["fixes"].append(commit_info)
|
|
elif commit.summary.startswith("docs:"):
|
|
categories["documentation"].append(commit_info)
|
|
else:
|
|
categories["others"].append(commit_info)
|
|
|
|
return categories
|
|
|
|
|
|
@tool("Get Git Commits Since Last Tag")
|
|
def get_commits_since_last_tag(repo_path: str = config["GIT_REPO_PATH"]) -> str:
|
|
"""
|
|
Retrieves all commits since the last tag in the Git repository.
|
|
|
|
Args:
|
|
repo_path (str): Path to the Git repository.
|
|
|
|
Returns:
|
|
str: JSON string containing commit information.
|
|
"""
|
|
repo = git.Repo(repo_path)
|
|
tags = sorted(repo.tags, key=lambda t: t.commit.committed_datetime)
|
|
last_tag = tags[-1] if tags else None
|
|
|
|
if not last_tag:
|
|
commits = list(repo.iter_commits())
|
|
else:
|
|
commits = list(repo.iter_commits(f"{last_tag.name}..HEAD"))
|
|
|
|
commit_info = [
|
|
{
|
|
"hash": commit.hexsha,
|
|
"date": commit.committed_datetime.isoformat(),
|
|
"message": commit.message.strip(),
|
|
}
|
|
for commit in commits
|
|
]
|
|
return json.dumps(commit_info)
|
|
|
|
|
|
@tool("Get Redmine Issues")
|
|
def get_redmine_issues(project_id: int, version_id: int, changelog_type: str) -> str:
|
|
"""
|
|
Retrieves issues for a specific Redmine project and version.
|
|
|
|
Args:
|
|
project_id (int): ID of the Redmine project.
|
|
version_id (int): ID of the project version.
|
|
changelog_type (str): Type of changelog ("Changelog" or "Release_notes").
|
|
|
|
Returns:
|
|
str: JSON string containing issue information.
|
|
"""
|
|
if changelog_type == "Changelog":
|
|
custom_field = "&cf_21=Changelog"
|
|
elif changelog_type == "ReleaseNotes":
|
|
custom_field = "&cf_21=Release_notes"
|
|
else:
|
|
custom_field = ""
|
|
url = f"{config['REDMINE_HOST']}/issues.json?project_id={project_id}&fixed_version_id={version_id}&status_id=*{custom_field}"
|
|
data = make_redmine_request(url)
|
|
issues = [
|
|
{
|
|
"id": issue["id"],
|
|
"subject": issue["subject"],
|
|
"tracker": issue["tracker"]["name"],
|
|
}
|
|
for issue in data["issues"]
|
|
]
|
|
return json.dumps(issues)
|
|
|
|
|
|
def make_redmine_request(url):
|
|
context = ssl.create_default_context()
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
|
|
req = request.Request(url, method="GET")
|
|
req.add_header("Content-Type", "application/json")
|
|
req.add_header("X-Redmine-API-Key", config["REDMINE_API_KEY"])
|
|
return json.load(request.urlopen(req, context=context))
|