Bases: BaseService
A service class for AI operations.
This class provides methods for code review, commit summarization,
project analysis, and question answering using AI bots.
The constructor for the AIService class.
Parameters:
-
analyzer
(ProjectAnalysisBot | None, default:
None
)
–
The project analysis bot instance.
-
commit_summarizer
(CommitSummaryBot | None, default:
None
)
–
The commit summary bot instance.
-
questioner
(ProjectQuestionBot | None, default:
None
)
–
The project question bot instance.
-
reviewer
(CodeReviewBot | None, default:
None
)
–
The code review bot instance.
Source code in dev_tool/services/ai/service.py
| def __init__(
self,
analyzer: ProjectAnalysisBot | None = None,
commit_summarizer: CommitSummaryBot | None = None,
questioner: ProjectQuestionBot | None = None,
reviewer: CodeReviewBot | None = None
) -> None:
"""
The constructor for the AIService class.
:param analyzer: The project analysis bot instance.
:param commit_summarizer: The commit summary bot instance.
:param questioner: The project question bot instance.
:param reviewer: The code review bot instance.
"""
super().__init__()
self._analyzer = analyzer
self._cache: dict[str, Any] = {}
self._commit_summarizer = commit_summarizer
self._questioner = questioner
self._reviewer = reviewer
self.task_manager = TaskManager()
|
A method that asks a question about the project using AI.
Parameters:
-
question
(str)
–
-
context
(str)
–
The project context information.
-
task_id
(str | None, default:
None
)
–
The task ID for cancellation checking.
Returns:
Raises:
-
AIBotNotInitializedError
–
If the question bot is not initialized.
-
AIProcessingError
–
If the AI processing fails.
Source code in dev_tool/services/ai/service.py
| def ask_question(self, question: str, context: str, task_id: str | None = None) -> str:
"""
A method that asks a question about the project using AI.
:param question: The question to ask.
:param context: The project context information.
:param task_id: The task ID for cancellation checking.
:return: The AI-generated answer.
:raises AIBotNotInitializedError: If the question bot is not initialized.
:raises AIProcessingError: If the AI processing fails.
"""
if not self._questioner:
message = 'Question bot not initialized'
log.exception(message)
raise AIBotNotInitializedError(message)
if task_id and self.task_manager.is_cancelled(task_id):
return 'Task cancelled'
try:
intel = self._questioner.process(question, context)
except Exception as exception:
message = f'Failed to get response from bot: {exception}'
log.exception(message)
raise AIProcessingError(message) from None
else:
return intel.answer
|
A method that clears the project context cache.
Source code in dev_tool/services/ai/service.py
| def clear_cache(self) -> None:
"""A method that clears the project context cache."""
self._cache.clear()
message = 'Project context cache cleared'
self.notification.normal_text(message)
log.debug(message)
|
A method that creates a project map.
Returns:
Source code in dev_tool/services/ai/service.py
| def create_project_map(self) -> Any:
"""
A method that creates a project map.
:return: The project map or None.
"""
return None
|
A method that retrieves current git changes.
Parameters:
-
task_id
(str | None, default:
None
)
–
The task ID for cancellation checking.
Returns:
-
str
–
The git diff output or a message if no changes.
Raises:
-
GitNotFoundError
–
If git is not found in PATH.
-
GitOperationError
–
If the git operation fails.
Source code in dev_tool/services/ai/service.py
| def get_git_changes(self, task_id: str | None = None) -> str:
"""
A method that retrieves current git changes.
:param task_id: The task ID for cancellation checking.
:return: The git diff output or a message if no changes.
:raises GitNotFoundError: If git is not found in PATH.
:raises GitOperationError: If the git operation fails.
"""
if task_id and self.task_manager.is_cancelled(task_id):
return 'Task cancelled'
git = shutil.which('git')
if not git:
message = 'Git not found in PATH'
log.exception(message)
raise GitNotFoundError(message)
try:
result = subprocess.run(
[git, 'diff', '--cached'],
capture_output=True,
text=True,
cwd=BASE,
check=False
)
if task_id and self.task_manager.is_cancelled(task_id):
return 'Task cancelled'
if result.stdout:
return result.stdout
result = subprocess.run(
[git, 'diff'],
capture_output=True,
text=True,
cwd=BASE,
check=False
)
except Exception as exception:
message = f'Could not retrieve git changes: {exception}'
log.exception(message)
raise GitOperationError(message) from None
else:
return result.stdout or 'No changes detected'
|
A method that retrieves recent git commits.
Parameters:
-
count
(int, default:
10
)
–
The number of commits to retrieve.
-
task_id
(str | None, default:
None
)
–
The task ID for cancellation checking.
Returns:
-
str
–
A formatted string of recent commits.
Raises:
-
GitNotFoundError
–
If git is not found in PATH.
-
GitOperationError
–
If the git operation fails.
Source code in dev_tool/services/ai/service.py
| def get_recent_commits(self, count: int = 10, task_id: str | None = None) -> str:
"""
A method that retrieves recent git commits.
:param count: The number of commits to retrieve.
:param task_id: The task ID for cancellation checking.
:return: A formatted string of recent commits.
:raises GitNotFoundError: If git is not found in PATH.
:raises GitOperationError: If the git operation fails.
"""
if task_id and self.task_manager.is_cancelled(task_id):
return 'Task cancelled'
git = shutil.which('git')
if not git:
message = 'Git not found in PATH'
log.exception(message)
raise GitNotFoundError(message)
try:
result = subprocess.run(
[git, 'log', f'-{count}', '--oneline', '--decorate', '--graph'],
capture_output=True,
text=True,
cwd=BASE,
check=False
)
except Exception as exception:
message = f'Could not retrieve commit history: {exception}'
log.exception(message)
raise GitOperationError(message) from None
if result.returncode != 0:
message = 'Could not retrieve commit history'
log.exception(message)
raise GitOperationError(message)
if not result.stdout:
message = 'No commits found'
log.exception(message)
raise GitOperationError(message)
if task_id and self.task_manager.is_cancelled(task_id):
return 'Task cancelled'
try:
detailed = subprocess.run(
[git, 'log', f'-{count}', '--format=%H|%an|%ae|%ad|%s', '--date=short'],
capture_output=True,
text=True,
cwd=BASE,
check=False
)
if detailed.stdout:
commits: list[str] = []
for line in detailed.stdout.strip().split('\n'):
if task_id and self.task_manager.is_cancelled(task_id):
return 'Task cancelled'
parts = line.split('|')
if len(parts) >= 5:
_, author, _, date, message = parts[0], parts[1], parts[2], parts[3], '|'.join(parts[4:])
commit = f'{date} - {message} (by {author})'
commits.append(commit)
return '\n'.join(commits)
except Exception as exception:
message = f'Could not retrieve commit history: {exception}'
log.exception(message)
raise GitOperationError(message) from None
else:
return result.stdout
|
A method that checks if all AI bots are available.
Returns:
-
bool
–
True if all bots are initialized, False otherwise.
Source code in dev_tool/services/ai/service.py
| def is_available(self) -> bool:
"""
A method that checks if all AI bots are available.
:return: True if all bots are initialized, False otherwise.
"""
return all([self._reviewer, self._analyzer, self._questioner, self._commit_summarizer])
|
A method that packages project information into a context string.
Parameters:
-
task_id
(str | None, default:
None
)
–
The task ID for cancellation checking.
Returns:
-
str
–
A formatted string containing project context.
Source code in dev_tool/services/ai/service.py
| def package_project(self, task_id: str | None = None) -> str:
"""
A method that packages project information into a context string.
:param task_id: The task ID for cancellation checking.
:return: A formatted string containing project context.
"""
if 'project_context' in self._cache:
return self._cache['project_context']
if task_id and self.task_manager.is_cancelled(task_id):
return 'Task cancelled'
info = [f'Project: {BASE.name}', f'Location: {BASE}']
self._package_structure(info, task_id)
if task_id and self.task_manager.is_cancelled(task_id):
return 'Task cancelled'
self._package_key_files(info, task_id)
if task_id and self.task_manager.is_cancelled(task_id):
return 'Task cancelled'
self._package_python_files(info, task_id)
context = '\n'.join(info)
self._cache['project_context'] = context
return context
|
A method that reviews code changes using AI.
Parameters:
-
changes
(str)
–
The code changes to review.
-
task_id
(str | None, default:
None
)
–
The task ID for cancellation checking.
Returns:
Raises:
-
AIBotNotInitializedError
–
If the reviewer bot is not initialized.
-
AIProcessingError
–
If the AI processing fails.
Source code in dev_tool/services/ai/service.py
| def review_code(self, changes: str, task_id: str | None = None) -> str:
"""
A method that reviews code changes using AI.
:param changes: The code changes to review.
:param task_id: The task ID for cancellation checking.
:return: A formatted code review.
:raises AIBotNotInitializedError: If the reviewer bot is not initialized.
:raises AIProcessingError: If the AI processing fails.
"""
if not self._reviewer:
message = 'Code review bot not initialized'
log.exception(message)
raise AIBotNotInitializedError(message)
if task_id and self.task_manager.is_cancelled(task_id):
return 'Task cancelled'
try:
intel = self._reviewer.process(changes)
if task_id and self.task_manager.is_cancelled(task_id):
return 'Task cancelled'
result = [f'Code Quality: {intel.quality}']
if intel.bugs:
result.append('\nBugs/Issues:')
result.extend(f' - {bug}' for bug in intel.bugs)
if intel.improvements:
result.append('\nImprovements:')
result.extend(f' - {improvement}' for improvement in intel.improvements)
if intel.security:
result.append('\nSecurity Concerns:')
result.extend(f' - {concern}' for concern in intel.security)
except Exception as exception:
message = f'Failed to review code: {exception}'
log.exception(message)
raise AIProcessingError(message) from None
else:
return '\n'.join(result)
|
A method that summarizes git commits using AI.
Parameters:
-
commits
(str)
–
The commit history to summarize.
-
task_id
(str | None, default:
None
)
–
The task ID for cancellation checking.
Returns:
-
str
–
A formatted commit summary.
Raises:
-
AIBotNotInitializedError
–
If the summarizer bot is not initialized.
-
AIProcessingError
–
If the AI processing fails.
Source code in dev_tool/services/ai/service.py
| def summarize_commits(self, commits: str, task_id: str | None = None) -> str:
"""
A method that summarizes git commits using AI.
:param commits: The commit history to summarize.
:param task_id: The task ID for cancellation checking.
:return: A formatted commit summary.
:raises AIBotNotInitializedError: If the summarizer bot is not initialized.
:raises AIProcessingError: If the AI processing fails.
"""
if not self._commit_summarizer:
message = 'Commit summary bot not initialized'
log.exception(message)
raise AIBotNotInitializedError(message)
if task_id and self.task_manager.is_cancelled(task_id):
return 'Task cancelled'
try:
intel = self._commit_summarizer.process(commits)
if task_id and self.task_manager.is_cancelled(task_id):
return 'Task cancelled'
result = [f'Overview: {intel.overview}']
if intel.features:
result.append('\nNew Features:')
result.extend(f' - {feature}' for feature in intel.features)
if intel.improvements:
result.append('\nImprovements:')
result.extend(f' - {improvement}' for improvement in intel.improvements)
if intel.fixes:
result.append('\nBug Fixes:')
result.extend(f' - {fix}' for fix in intel.fixes)
except Exception as exception:
message = f'Failed to summarize commit(s): {exception}'
log.exception(message)
raise AIProcessingError(message) from None
else:
return '\n'.join(result)
|