834 lines
35 KiB
Python
834 lines
35 KiB
Python
"""Git synchronization implementation for git repositories"""
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
from typing import List, Dict, Any, Set
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from loguru import logger
|
|
from config import BaseDataSourceConfig
|
|
from sync.base_sync import BaseSync
|
|
from rag.file_parser import FileParser
|
|
from llama_index.core import Document
|
|
import paramiko
|
|
|
|
|
|
class GitSync(BaseSync):
|
|
"""Handle synchronization between git repositories and ChromaDB"""
|
|
|
|
def __init__(self, config: BaseDataSourceConfig, vector_store_manager=None):
|
|
"""
|
|
Initialize git sync with configuration
|
|
|
|
Args:
|
|
config: Git configuration
|
|
"""
|
|
super().__init__(config)
|
|
self.file_parser = FileParser()
|
|
self.vector_store_manager = vector_store_manager
|
|
|
|
def test_connection(self) -> tuple[bool, list]:
|
|
"""
|
|
Test git connection
|
|
|
|
Returns:
|
|
tuple[bool, list]: (True if connection is successful, list of repositories)
|
|
"""
|
|
try:
|
|
if self.config.git_mode == "single":
|
|
# Test single repo connection
|
|
success = self._test_single_repo_connection()
|
|
return success, []
|
|
else:
|
|
# Test server connection
|
|
return self._test_server_connection()
|
|
except Exception as e:
|
|
logger.error(f"Error testing git connection: {e}")
|
|
return False, []
|
|
|
|
def _test_single_repo_connection(self) -> bool:
|
|
"""
|
|
Test connection to a single git repository
|
|
|
|
Returns:
|
|
bool: True if connection is successful
|
|
"""
|
|
try:
|
|
# For SSH protocol, test SSH connection first
|
|
if self.config.git_protocol == "ssh" and self.config.git_ssh_host:
|
|
# Test SSH connection to the git server
|
|
ssh_client = paramiko.SSHClient()
|
|
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
|
|
# Connect to the server
|
|
ssh_client.connect(
|
|
hostname=self.config.git_ssh_host,
|
|
port=self.config.git_ssh_port,
|
|
username=self.config.git_ssh_username,
|
|
password=self.config.git_ssh_password or self.config.git_token,
|
|
timeout=10
|
|
)
|
|
|
|
# Connection successful
|
|
logger.info(f"Successfully connected to git server via SSH: {self.config.git_ssh_host}")
|
|
ssh_client.close()
|
|
return True
|
|
|
|
# For HTTPS or if no SSH config, use git clone
|
|
# Create a temporary directory for testing
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Build git clone command
|
|
cmd = ["git", "clone", self.config.git_repo_url, temp_dir]
|
|
|
|
# Add authentication if provided
|
|
env = os.environ.copy()
|
|
if self.config.git_token:
|
|
# For HTTPS, we can use the URL with token
|
|
if self.config.git_protocol == "https":
|
|
repo_url = self.config.git_repo_url
|
|
if "https://" in repo_url:
|
|
repo_url = repo_url.replace("https://", f"https://{self.config.git_token}@")
|
|
cmd = ["git", "clone", repo_url, temp_dir]
|
|
|
|
# Run the command
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
timeout=30
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
logger.info(f"Successfully connected to git repository: {self.config.git_repo_url}")
|
|
return True
|
|
else:
|
|
logger.error(f"Failed to connect to git repository: {result.stderr}")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"Error testing single repo connection: {e}")
|
|
return False
|
|
|
|
def _test_server_connection(self) -> tuple[bool, list]:
|
|
"""
|
|
Test connection to a git server
|
|
|
|
Returns:
|
|
tuple[bool, list]: (True if connection is successful, list of repositories with default branches)
|
|
"""
|
|
git_repos = []
|
|
try:
|
|
# Check if it's a local test server
|
|
if self.config.git_server_url in ['localhost', '127.0.0.1']:
|
|
# For local testing, directly check the directory
|
|
if os.path.exists(self.config.git_server_path):
|
|
# Check if there are git repositories in the path
|
|
repo_dirs = [d for d in os.listdir(self.config.git_server_path)
|
|
if os.path.isdir(os.path.join(self.config.git_server_path, d))
|
|
and (d.endswith('.git') or os.path.exists(os.path.join(self.config.git_server_path, d, 'HEAD')))]
|
|
|
|
# Get default branch for each repository
|
|
for repo_dir in repo_dirs:
|
|
repo_path = os.path.join(self.config.git_server_path, repo_dir)
|
|
try:
|
|
default_branch = self._get_default_branch(repo_path)
|
|
git_repos.append({
|
|
"name": repo_dir,
|
|
"default_branch": default_branch
|
|
})
|
|
except Exception as e:
|
|
logger.error(f"Error getting default branch for {repo_dir}: {e}")
|
|
git_repos.append({
|
|
"name": repo_dir,
|
|
"default_branch": "unknown"
|
|
})
|
|
|
|
if git_repos:
|
|
logger.info(f"Successfully connected to local git server and found repositories: {[repo['name'] for repo in git_repos]}")
|
|
return True, git_repos
|
|
else:
|
|
logger.warning(f"Connected to local git server but no repositories found in {self.config.git_server_path}")
|
|
return True, git_repos # Connection successful, just no repos found
|
|
else:
|
|
logger.error(f"Local git server path does not exist: {self.config.git_server_path}")
|
|
return False, git_repos
|
|
else:
|
|
# Test SSH connection to the server
|
|
ssh_client = paramiko.SSHClient()
|
|
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
|
|
# Connect to the server
|
|
ssh_client.connect(
|
|
hostname=self.config.git_server_url,
|
|
port=self.config.git_server_port,
|
|
username=self.config.git_server_username,
|
|
password=self.config.git_server_password or self.config.git_token,
|
|
timeout=10
|
|
)
|
|
|
|
# Try to list the git server path
|
|
stdin, stdout, stderr = ssh_client.exec_command(f"ls -la {self.config.git_server_path}")
|
|
output = stdout.read().decode('utf-8')
|
|
error = stderr.read().decode('utf-8')
|
|
|
|
if error:
|
|
logger.error(f"Error listing git server path: {error}")
|
|
return False, git_repos
|
|
|
|
# Check if there are git repositories in the path
|
|
repo_lines = [line for line in output.split('\n') if '.git' in line]
|
|
for line in repo_lines:
|
|
# Extract repo name from the line
|
|
parts = line.split()
|
|
if parts:
|
|
repo_name = parts[-1]
|
|
# Try to get default branch via SSH
|
|
try:
|
|
# For remote servers, we'll just return the repo name without branch info
|
|
# as getting branch info would require more complex SSH commands
|
|
git_repos.append({
|
|
"name": repo_name,
|
|
"default_branch": "unknown"
|
|
})
|
|
except Exception as e:
|
|
logger.error(f"Error processing repository {repo_name}: {e}")
|
|
git_repos.append({
|
|
"name": repo_name,
|
|
"default_branch": "unknown"
|
|
})
|
|
|
|
if git_repos:
|
|
logger.info(f"Successfully connected to git server and found repositories")
|
|
return True, git_repos
|
|
else:
|
|
logger.warning(f"Connected to git server but no repositories found in {self.config.git_server_path}")
|
|
return True, git_repos # Connection successful, just no repos found
|
|
except Exception as e:
|
|
logger.error(f"Error testing server connection: {e}")
|
|
return False, git_repos
|
|
finally:
|
|
if 'ssh_client' in locals():
|
|
ssh_client.close()
|
|
|
|
def fetch_all_documents(self, last_sync_time=None) -> List[Dict[str, Any]]:
|
|
"""
|
|
Fetch documents from git repositories
|
|
|
|
Args:
|
|
last_sync_time: Last synchronization time (for incremental sync)
|
|
|
|
Returns:
|
|
List of documents
|
|
"""
|
|
documents = []
|
|
|
|
if self.config.git_mode == "single":
|
|
# Fetch from single repository
|
|
repo_docs = self._fetch_from_single_repo(last_sync_time)
|
|
documents.extend(repo_docs)
|
|
else:
|
|
# Fetch from multiple repositories on server
|
|
server_docs = self._fetch_from_server(last_sync_time)
|
|
documents.extend(server_docs)
|
|
|
|
return documents
|
|
|
|
def _fetch_from_single_repo(self, last_sync_time=None) -> List[Dict[str, Any]]:
|
|
"""
|
|
Fetch documents from a single git repository
|
|
|
|
Args:
|
|
last_sync_time: Last synchronization time
|
|
|
|
Returns:
|
|
List of documents
|
|
"""
|
|
documents = []
|
|
|
|
try:
|
|
# Create a temporary directory for the repository
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Clone the repository
|
|
self._clone_repository(self.config.git_repo_url, temp_dir)
|
|
|
|
# Checkout the specified branch
|
|
self._checkout_branch(temp_dir, self.config.git_branch)
|
|
|
|
# Get all files in the repository
|
|
files = self._get_all_files(temp_dir)
|
|
|
|
# Process each file
|
|
for file_path in files:
|
|
# Check if file should be processed
|
|
if not self._should_process_file(file_path):
|
|
continue
|
|
|
|
# Generate document ID
|
|
doc_id = self._generate_doc_id(self.config.git_repo_url, file_path)
|
|
|
|
# Check if document has already been synced
|
|
if last_sync_time is not None:
|
|
if self.vector_store_manager and self.vector_store_manager.document_exists(doc_id + "_chunk_0"):
|
|
# Skip if not modified since last sync
|
|
continue
|
|
|
|
# Read and process the file
|
|
try:
|
|
with open(file_path, 'rb') as f:
|
|
file_bytes = f.read()
|
|
|
|
# Parse file content
|
|
parsed_docs = self.file_parser.parse_file_content(file_bytes, file_path, doc_id=doc_id, host=self.config.git_repo_url)
|
|
if parsed_docs:
|
|
content = '\n\n'.join(doc.text for doc in parsed_docs if doc.text)
|
|
else:
|
|
logger.warning(f"No content extracted from {file_path}")
|
|
content = f"[无法读取文件：{Path(file_path).name}]"
|
|
|
|
# Build document
|
|
document = {
|
|
'id': doc_id,
|
|
'content': content,
|
|
'metadata': {
|
|
'file_path': str(file_path),
|
|
'repository': self.config.git_repo_url,
|
|
'branch': self.config.git_branch,
|
|
'update_time': datetime.now()
|
|
}
|
|
}
|
|
documents.append(document)
|
|
except Exception as e:
|
|
logger.error(f"Error processing file {file_path}: {e}")
|
|
except Exception as e:
|
|
logger.error(f"Error fetching from single repo: {e}")
|
|
|
|
return documents
|
|
|
|
def _fetch_from_server(self, last_sync_time=None) -> List[Dict[str, Any]]:
|
|
"""
|
|
Fetch documents from multiple git repositories on a server
|
|
|
|
Args:
|
|
last_sync_time: Last synchronization time
|
|
|
|
Returns:
|
|
List of documents
|
|
"""
|
|
documents = []
|
|
|
|
try:
|
|
# Get list of repositories to process
|
|
if hasattr(self.config, 'git_repositories') and self.config.git_repositories:
|
|
# Use user-selected repositories
|
|
selected_repos = self.config.git_repositories
|
|
|
|
# Get all repositories on the server
|
|
all_repos = dict(self._get_server_repositories())
|
|
|
|
# Process only selected repositories
|
|
for repo_info in selected_repos:
|
|
repo_name = repo_info['name']
|
|
branch = repo_info.get('branch', 'main')
|
|
|
|
if repo_name in all_repos:
|
|
repo_path = all_repos[repo_name]
|
|
# Create a temporary directory for the repository
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Clone the repository
|
|
repo_url = f"ssh://{self.config.git_server_username}@{self.config.git_server_url}:{repo_path}"
|
|
self._clone_repository(repo_url, temp_dir)
|
|
|
|
# Checkout the specified branch
|
|
self._checkout_branch(temp_dir, branch)
|
|
|
|
# Get all files in the repository
|
|
files = self._get_all_files(temp_dir)
|
|
|
|
# Process each file
|
|
for file_path in files:
|
|
# Check if file should be processed
|
|
if not self._should_process_file(file_path):
|
|
continue
|
|
|
|
# Generate document ID
|
|
doc_id = self._generate_doc_id(repo_url, file_path)
|
|
|
|
# Check if document has already been synced
|
|
if last_sync_time is not None:
|
|
if self.vector_store_manager and self.vector_store_manager.document_exists(doc_id + "_chunk_0"):
|
|
# Skip if not modified since last sync
|
|
continue
|
|
|
|
# Read and process the file
|
|
try:
|
|
with open(file_path, 'rb') as f:
|
|
file_bytes = f.read()
|
|
|
|
# Parse file content
|
|
parsed_docs = self.file_parser.parse_file_content(file_bytes, file_path, doc_id=doc_id, host=repo_url)
|
|
if parsed_docs:
|
|
content = '\n\n'.join(doc.text for doc in parsed_docs if doc.text)
|
|
else:
|
|
logger.warning(f"No content extracted from {file_path}")
|
|
content = f"[无法读取文件：{Path(file_path).name}]"
|
|
|
|
# Build document
|
|
document = {
|
|
'id': doc_id,
|
|
'content': content,
|
|
'metadata': {
|
|
'file_path': str(file_path),
|
|
'repository': repo_url,
|
|
'branch': branch,
|
|
'update_time': datetime.now()
|
|
}
|
|
}
|
|
documents.append(document)
|
|
except Exception as e:
|
|
logger.error(f"Error processing file {file_path}: {e}")
|
|
else:
|
|
# Get list of repositories on the server
|
|
repos = self._get_server_repositories()
|
|
|
|
# Process each repository
|
|
for repo_name, repo_path in repos:
|
|
# Create a temporary directory for the repository
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Clone the repository
|
|
repo_url = f"ssh://{self.config.git_server_username}@{self.config.git_server_url}:{repo_path}"
|
|
self._clone_repository(repo_url, temp_dir)
|
|
|
|
# Get the default branch
|
|
branch = self._get_default_branch(temp_dir)
|
|
|
|
# Checkout the default branch
|
|
self._checkout_branch(temp_dir, branch)
|
|
|
|
# Get all files in the repository
|
|
files = self._get_all_files(temp_dir)
|
|
|
|
# Process each file
|
|
for file_path in files:
|
|
# Check if file should be processed
|
|
if not self._should_process_file(file_path):
|
|
continue
|
|
|
|
# Generate document ID
|
|
doc_id = self._generate_doc_id(repo_url, file_path)
|
|
|
|
# Check if document has already been synced
|
|
if last_sync_time is not None:
|
|
if self.vector_store_manager and self.vector_store_manager.document_exists(doc_id + "_chunk_0"):
|
|
# Skip if not modified since last sync
|
|
continue
|
|
|
|
# Read and process the file
|
|
try:
|
|
with open(file_path, 'rb') as f:
|
|
file_bytes = f.read()
|
|
|
|
# Parse file content
|
|
parsed_docs = self.file_parser.parse_file_content(file_bytes, file_path, doc_id=doc_id, host=repo_url)
|
|
if parsed_docs:
|
|
content = '\n\n'.join(doc.text for doc in parsed_docs if doc.text)
|
|
else:
|
|
logger.warning(f"No content extracted from {file_path}")
|
|
content = f"[无法读取文件：{Path(file_path).name}]"
|
|
|
|
# Build document
|
|
document = {
|
|
'id': doc_id,
|
|
'content': content,
|
|
'metadata': {
|
|
'file_path': str(file_path),
|
|
'repository': repo_url,
|
|
'branch': branch,
|
|
'update_time': datetime.now()
|
|
}
|
|
}
|
|
documents.append(document)
|
|
except Exception as e:
|
|
logger.error(f"Error processing file {file_path}: {e}")
|
|
except Exception as e:
|
|
logger.error(f"Error fetching from server: {e}")
|
|
|
|
return documents
|
|
|
|
def _get_server_repositories(self) -> List[tuple]:
|
|
"""
|
|
Get list of git repositories on the server
|
|
|
|
Returns:
|
|
List of (repo_name, repo_path) tuples
|
|
"""
|
|
repos = []
|
|
|
|
try:
|
|
# Connect to the server via SSH
|
|
ssh_client = paramiko.SSHClient()
|
|
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
|
|
ssh_client.connect(
|
|
hostname=self.config.git_server_url,
|
|
port=self.config.git_server_port,
|
|
username=self.config.git_server_username,
|
|
password=self.config.git_server_password or self.config.git_token,
|
|
timeout=10
|
|
)
|
|
|
|
# List directories in the server path
|
|
stdin, stdout, stderr = ssh_client.exec_command(f"ls -la {self.config.git_server_path}")
|
|
output = stdout.read().decode('utf-8')
|
|
error = stderr.read().decode('utf-8')
|
|
|
|
if error:
|
|
logger.error(f"Error listing server path: {error}")
|
|
return repos
|
|
|
|
# Parse the output to find git repositories
|
|
for line in output.split('\n'):
|
|
if 'drwxr' in line:
|
|
parts = line.split()
|
|
if len(parts) >= 9:
|
|
repo_name = parts[8]
|
|
repo_path = os.path.join(self.config.git_server_path, repo_name)
|
|
|
|
# Check if this is a git repository (bare or regular)
|
|
check_cmd = f"if [ -d '{repo_path}/.git' ] || [ -f '{repo_path}/HEAD' ]; then echo 'git'; else echo 'notgit'; fi"
|
|
stdin, stdout, stderr = ssh_client.exec_command(check_cmd)
|
|
is_git = stdout.read().decode('utf-8').strip() == 'git'
|
|
|
|
if is_git:
|
|
repos.append((repo_name, repo_path))
|
|
except Exception as e:
|
|
logger.error(f"Error getting server repositories: {e}")
|
|
finally:
|
|
if 'ssh_client' in locals():
|
|
ssh_client.close()
|
|
|
|
return repos
|
|
|
|
def _clone_repository(self, repo_url, dest_path):
|
|
"""
|
|
Clone a git repository
|
|
|
|
Args:
|
|
repo_url: Repository URL
|
|
dest_path: Destination path
|
|
"""
|
|
try:
|
|
# Build git clone command
|
|
cmd = ["git", "clone", repo_url, dest_path]
|
|
|
|
# Add authentication if provided
|
|
env = os.environ.copy()
|
|
if self.config.git_token:
|
|
# For HTTPS, we can use the URL with token
|
|
if self.config.git_protocol == "https":
|
|
if "https://" in repo_url:
|
|
repo_url = repo_url.replace("https://", f"https://{self.config.git_token}@")
|
|
cmd = ["git", "clone", repo_url, dest_path]
|
|
|
|
# Run the command
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
timeout=60
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
raise Exception(f"Failed to clone repository: {result.stderr}")
|
|
except Exception as e:
|
|
logger.error(f"Error cloning repository {repo_url}: {e}")
|
|
raise
|
|
|
|
def _checkout_branch(self, repo_path, branch):
|
|
"""
|
|
Checkout a branch in a git repository
|
|
|
|
Args:
|
|
repo_path: Repository path
|
|
branch: Branch name
|
|
"""
|
|
try:
|
|
# Run git checkout command
|
|
result = subprocess.run(
|
|
["git", "checkout", branch],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
raise Exception(f"Failed to checkout branch {branch}: {result.stderr}")
|
|
except Exception as e:
|
|
logger.error(f"Error checking out branch {branch}: {e}")
|
|
raise
|
|
|
|
def _get_default_branch(self, repo_path):
|
|
"""
|
|
Get the default branch of a git repository
|
|
|
|
Args:
|
|
repo_path: Repository path
|
|
|
|
Returns:
|
|
Default branch name
|
|
"""
|
|
try:
|
|
# Check if it's a bare repository
|
|
is_bare = os.path.exists(os.path.join(repo_path, 'HEAD')) and not os.path.exists(os.path.join(repo_path, '.git'))
|
|
|
|
if is_bare:
|
|
# For bare repositories, check HEAD file or refs/heads
|
|
head_path = os.path.join(repo_path, 'HEAD')
|
|
if os.path.exists(head_path):
|
|
with open(head_path, 'r') as f:
|
|
head_content = f.read().strip()
|
|
if head_content.startswith('ref: '):
|
|
# Extract branch from ref: refs/heads/branch
|
|
ref_path = head_content.split('ref: ')[1]
|
|
if ref_path.startswith('refs/heads/'):
|
|
return ref_path.split('refs/heads/')[1]
|
|
|
|
# Fallback: list branches and return the first one
|
|
result = subprocess.run(
|
|
["git", "branch", "-a"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30
|
|
)
|
|
if result.returncode == 0:
|
|
branches = [line.strip() for line in result.stdout.split('\n') if line.strip() and not line.strip().startswith('*')]
|
|
if branches:
|
|
# Extract branch name from remotes or local branches
|
|
for branch in branches:
|
|
if '->' in branch:
|
|
continue
|
|
if '/' in branch:
|
|
return branch.split('/')[-1]
|
|
return branch
|
|
else:
|
|
# For regular repositories
|
|
result = subprocess.run(
|
|
["git", "symbolic-ref", "--short", "HEAD"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
return result.stdout.strip()
|
|
else:
|
|
# Fallback to main or master
|
|
for branch in ["main", "master"]:
|
|
result = subprocess.run(
|
|
["git", "checkout", branch],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30
|
|
)
|
|
if result.returncode == 0:
|
|
return branch
|
|
|
|
# Final fallback
|
|
raise Exception("Could not determine default branch")
|
|
except Exception as e:
|
|
logger.error(f"Error getting default branch: {e}")
|
|
return "main"
|
|
|
|
def _get_all_files(self, repo_path):
|
|
"""
|
|
Get all files in a git repository
|
|
|
|
Args:
|
|
repo_path: Repository path
|
|
|
|
Returns:
|
|
List of file paths
|
|
"""
|
|
files = []
|
|
|
|
try:
|
|
# Run git ls-files to get all tracked files
|
|
result = subprocess.run(
|
|
["git", "ls-files"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
for file_path in result.stdout.strip().split('\n'):
|
|
if file_path:
|
|
full_path = os.path.join(repo_path, file_path)
|
|
if os.path.isfile(full_path):
|
|
files.append(full_path)
|
|
except Exception as e:
|
|
logger.error(f"Error getting all files: {e}")
|
|
|
|
return files
|
|
|
|
def _should_process_file(self, file_path):
|
|
"""
|
|
Check if a file should be processed
|
|
|
|
Args:
|
|
file_path: File path
|
|
|
|
Returns:
|
|
True if file should be processed, False otherwise
|
|
"""
|
|
# Check if file extension is supported
|
|
if Path(file_path).suffix.lower() not in FileParser.SUPPORTED_EXTENSIONS:
|
|
return False
|
|
|
|
# Check if file is in .git directory
|
|
if '.git' in file_path:
|
|
return False
|
|
|
|
return True
|
|
|
|
def _generate_doc_id(self, repo_url, file_path):
|
|
"""
|
|
Generate a unique document ID for git files
|
|
|
|
Args:
|
|
repo_url: Repository URL
|
|
file_path: File path
|
|
|
|
Returns:
|
|
Unique document ID
|
|
"""
|
|
# Extract repository name from URL
|
|
repo_name = repo_url.split('/')[-1].replace('.git', '')
|
|
|
|
# Get relative path from repository root
|
|
relative_path = os.path.relpath(file_path, os.path.dirname(file_path.split('.git')[0]))
|
|
|
|
# Replace special characters
|
|
sanitized_path = relative_path.replace('/', '_').replace('\\', '_').replace(':', '_').replace(' ', '_')
|
|
|
|
return f"git_{repo_name}_{sanitized_path}"
|
|
|
|
def generate_doc_id(self, identifier: str) -> str:
|
|
"""
|
|
Generate a unique document ID for git files
|
|
|
|
Args:
|
|
identifier: Unique identifier for the document (file path, record ID, etc.)
|
|
|
|
Returns:
|
|
Unique document ID
|
|
"""
|
|
# For git, identifier is typically in the format "repo_url:file_path"
|
|
if ':' in identifier:
|
|
repo_url, file_path = identifier.split(':', 1)
|
|
return self._generate_doc_id(repo_url, file_path)
|
|
else:
|
|
# Fallback: use the identifier as is
|
|
sanitized_identifier = identifier.replace('/', '_').replace('\\', '_')
|
|
return f"git_{sanitized_identifier}"
|
|
|
|
def fetch_new_documents(self, last_sync_time=None) -> List[Dict[str, Any]]:
|
|
"""
|
|
Fetch new/updated documents from the data source since last sync time
|
|
|
|
Args:
|
|
last_sync_time: Last synchronization time
|
|
|
|
Returns:
|
|
List of new/updated documents
|
|
"""
|
|
# Use the existing fetch_all_documents method which already supports last_sync_time
|
|
return self.fetch_all_documents(last_sync_time)
|
|
|
|
def doc_to_llamaindex_doc(self, doc: Dict) -> 'Document':
|
|
"""
|
|
Convert git document to LlamaIndex Document
|
|
|
|
Args:
|
|
doc: Git document dictionary
|
|
|
|
Returns:
|
|
LlamaIndex Document object
|
|
"""
|
|
content = doc.get('content', "")
|
|
doc_id = doc.get('id', "")
|
|
metadata = doc.get('metadata', {})
|
|
|
|
# Ensure metadata has source information
|
|
metadata['source'] = 'git'
|
|
metadata['repository'] = metadata.get('repository', 'unknown')
|
|
|
|
# Create Document
|
|
return Document(
|
|
text=content,
|
|
id_=doc_id,
|
|
metadata=metadata
|
|
)
|
|
|
|
def get_synced_document_ids(self) -> Set[str]:
|
|
"""
|
|
Get IDs of all files in git repositories
|
|
|
|
Returns:
|
|
Set of document IDs
|
|
"""
|
|
doc_ids = set()
|
|
|
|
try:
|
|
if self.config.git_mode == "single":
|
|
# Get files from single repository
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
self._clone_repository(self.config.git_repo_url, temp_dir)
|
|
self._checkout_branch(temp_dir, self.config.git_branch)
|
|
files = self._get_all_files(temp_dir)
|
|
for file_path in files:
|
|
if self._should_process_file(file_path):
|
|
doc_id = self._generate_doc_id(self.config.git_repo_url, file_path)
|
|
doc_ids.add(doc_id)
|
|
else:
|
|
# Get files from multiple repositories
|
|
repos = self._get_server_repositories()
|
|
for repo_name, repo_path in repos:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
repo_url = f"ssh://git@{self.config.git_server_url}:{repo_path}"
|
|
self._clone_repository(repo_url, temp_dir)
|
|
branch = self._get_default_branch(temp_dir)
|
|
self._checkout_branch(temp_dir, branch)
|
|
files = self._get_all_files(temp_dir)
|
|
for file_path in files:
|
|
if self._should_process_file(file_path):
|
|
doc_id = self._generate_doc_id(repo_url, file_path)
|
|
doc_ids.add(doc_id)
|
|
except Exception as e:
|
|
logger.error(f"Error getting synced document IDs: {e}")
|
|
|
|
return doc_ids
|
|
|
|
@staticmethod
|
|
def check_data_source_exists(config: BaseDataSourceConfig) -> bool:
|
|
"""
|
|
Check if the git repository or server exists and is accessible
|
|
|
|
Args:
|
|
config: Git configuration
|
|
|
|
Returns:
|
|
True if git repository/server exists and is accessible, False otherwise
|
|
"""
|
|
try:
|
|
git_sync = GitSync(config)
|
|
return git_sync.test_connection()
|
|
except Exception as e:
|
|
logger.error(f"Error checking git data source: {e}")
|
|
return False |