feat: smart git URL parsing for browser URLs

- Add git URL parsing utilities (extract_base_repo_url, parse_git_url)
- Support GitHub, GitLab, Bitbucket browser URL detection
- Add /projects/repositories/parse-url endpoint
- Enhance repository creation to detect browser URLs and suggest corrections
- Add real-time URL validation in frontend with debouncing
- Show visual indicators (green/yellow/red) for URL validity
- Display inline suggestions with 'Use Suggested' button
- Add comprehensive unit tests for URL parsing
- Quality gates: ruff ✓, mypy ✓, typecheck ✓, lint ✓, build ✓
This commit is contained in:
Fusion
2026-05-19 12:25:44 +02:00
parent ac6c97b6ce
commit 8b70daed53
11 changed files with 1150 additions and 73 deletions
+110 -69
View File
@@ -14,6 +14,7 @@ from src.config import Settings
from src.models.git_repository import GitRepository
from src.models.project import Project
from src.models.user import User
from src.utils.git_url_parser import parse_git_url
router = APIRouter(prefix="/projects", tags=["git-repositories"])
@@ -46,6 +47,21 @@ def _get_repo_path(user_id: uuid.UUID, project_id: uuid.UUID, name: str) -> str:
class GitRepositoryCreate(BaseModel):
name: str
remote_url: str | None = None
force_original_url: bool = False
class URLParseRequest(BaseModel):
url: str
class URLParseResponse(BaseModel):
original_url: str
base_url: str | None
is_valid_clone_url: bool
needs_parsing: bool
host: str | None
message: str
error_code: str | None
class GitRepositoryResponse(BaseModel):
@@ -63,75 +79,6 @@ class GitRepositoryResponse(BaseModel):
updated_at: datetime
@router.post("/{project_id}/repositories", response_model=GitRepositoryResponse, status_code=status.HTTP_201_CREATED)
async def create_repository(
project_id: uuid.UUID,
data: GitRepositoryCreate,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> GitRepository:
_user = await _get_user(session, user_id)
_project = await _get_owned_project(project_id, user_id, session)
# Check for duplicate name
existing = await session.execute(
select(GitRepository).where(
GitRepository.project_id == project_id,
GitRepository.name == data.name,
)
)
if existing.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="repository name already exists")
repo_path = _get_repo_path(user_id, project_id, data.name)
# Ensure parent directory exists
os.makedirs(os.path.dirname(repo_path), exist_ok=True)
if data.remote_url:
# Clone as mirror
try:
result = subprocess.run(
["git", "clone", "--mirror", data.remote_url, repo_path],
capture_output=True,
text=True,
timeout=300,
)
if result.returncode != 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"failed to clone repository: {result.stderr}",
)
except subprocess.TimeoutExpired:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="clone operation timed out")
except FileNotFoundError:
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="git command not found")
else:
# Init bare repo
try:
result = subprocess.run(
["git", "init", "--bare", repo_path],
capture_output=True,
text=True,
check=True,
)
except FileNotFoundError:
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="git command not found")
repo = GitRepository(
name=data.name,
path=repo_path,
project_id=project_id,
owner_id=user_id,
is_mirror=bool(data.remote_url),
remote_url=data.remote_url,
)
session.add(repo)
await session.commit()
await session.refresh(repo)
return repo
@router.get("/{project_id}/repositories", response_model=list[GitRepositoryResponse])
async def list_repositories(
project_id: uuid.UUID,
@@ -168,3 +115,97 @@ async def delete_repository(
await session.delete(repo)
await session.commit()
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.post("/repositories/parse-url", response_model=URLParseResponse)
async def parse_repository_url(data: URLParseRequest) -> URLParseResponse:
"""Parse a git URL and detect if it's a browser URL that needs correction."""
result = parse_git_url(data.url)
return URLParseResponse(**result)
@router.post("/{project_id}/repositories", response_model=GitRepositoryResponse, status_code=status.HTTP_201_CREATED)
async def create_repository(
project_id: uuid.UUID,
data: GitRepositoryCreate,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> GitRepository:
_user = await _get_user(session, user_id)
_project = await _get_owned_project(project_id, user_id, session)
# Check for duplicate name
existing = await session.execute(
select(GitRepository).where(
GitRepository.project_id == project_id,
GitRepository.name == data.name,
)
)
if existing.scalar_one_or_none():
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="repository name already exists")
# Validate and potentially correct the URL
remote_url = data.remote_url
if remote_url and not data.force_original_url:
parse_result = parse_git_url(remote_url)
if parse_result["needs_parsing"] and parse_result["base_url"]:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail={
"message": "The provided URL appears to be a browser URL, not a git clone URL",
"suggested_url": parse_result["base_url"],
"original_url": remote_url,
"error_code": "URL_NEEDS_PARSING",
},
)
# Use base_url if it was extracted (for URLs without .git suffix)
if parse_result["base_url"]:
remote_url = parse_result["base_url"]
repo_path = _get_repo_path(user_id, project_id, data.name)
# Ensure parent directory exists
os.makedirs(os.path.dirname(repo_path), exist_ok=True)
if remote_url:
# Clone as mirror
try:
result = subprocess.run(
["git", "clone", "--mirror", remote_url, repo_path],
capture_output=True,
text=True,
timeout=300,
)
if result.returncode != 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"failed to clone repository: {result.stderr}",
)
except subprocess.TimeoutExpired:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="clone operation timed out")
except FileNotFoundError:
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="git command not found")
else:
# Init bare repo
try:
result = subprocess.run(
["git", "init", "--bare", repo_path],
capture_output=True,
text=True,
check=True,
)
except FileNotFoundError:
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="git command not found")
repo = GitRepository(
name=data.name,
path=repo_path,
project_id=project_id,
owner_id=user_id,
is_mirror=bool(remote_url),
remote_url=remote_url,
)
session.add(repo)
await session.commit()
await session.refresh(repo)
return repo
+228
View File
@@ -0,0 +1,228 @@
"""Git URL parsing utilities to extract base repository URLs from browser URLs."""
from urllib.parse import urlparse
def extract_base_repo_url(url: str) -> str | None:
"""Extract base repository URL from a browser/git URL.
Examples:
https://github.com/user/repo/tree/main → https://github.com/user/repo.git
https://github.com/user/repo.git → https://github.com/user/repo.git
git@github.com:user/repo.git → git@github.com:user/repo.git
https://gitlab.com/user/repo/-/blob/main/README.md → https://gitlab.com/user/repo.git
Returns None if URL doesn't match known patterns.
"""
# Handle SSH URLs (pass through unchanged)
if url.startswith("git@"):
return url if url.endswith(".git") else f"{url}.git"
try:
parsed = urlparse(url)
except Exception:
return None
# Remove query parameters
url = f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
# Extract host
host = parsed.netloc.lower()
# Split path
path_parts = [p for p in parsed.path.split("/") if p]
if not path_parts:
return None
# Determine host type and extract base
if "github.com" in host:
return _extract_github_url(url, path_parts)
elif "gitlab.com" in host:
return _extract_gitlab_url(url, path_parts)
elif "bitbucket.org" in host:
return _extract_bitbucket_url(url, path_parts)
else:
# Generic host - try basic extraction
return _extract_generic_url(url, path_parts)
def _extract_github_url(url: str, path_parts: list[str]) -> str | None:
"""Extract base repo URL from GitHub URL."""
# Need at least owner/repo
if len(path_parts) < 2:
return None
# Find the repo name (second path part)
# Remove trailing .git if present
repo_name = path_parts[1]
if repo_name.endswith(".git"):
repo_name = repo_name[:-4]
# Reconstruct base URL
base = f"https://github.com/{path_parts[0]}/{repo_name}"
# Add .git suffix
return f"{base}.git"
def _extract_gitlab_url(url: str, path_parts: list[str]) -> str | None:
"""Extract base repo URL from GitLab URL."""
# Need at least owner/repo
if len(path_parts) < 2:
return None
# Find the repo name (second path part)
repo_name = path_parts[1]
if repo_name.endswith(".git"):
repo_name = repo_name[:-4]
# Reconstruct base URL
base = f"https://gitlab.com/{path_parts[0]}/{repo_name}"
return f"{base}.git"
def _extract_bitbucket_url(url: str, path_parts: list[str]) -> str | None:
"""Extract base repo URL from Bitbucket URL."""
# Need at least owner/repo
if len(path_parts) < 2:
return None
# Find the repo name (second path part)
repo_name = path_parts[1]
if repo_name.endswith(".git"):
repo_name = repo_name[:-4]
# Reconstruct base URL
base = f"https://bitbucket.org/{path_parts[0]}/{repo_name}"
return f"{base}.git"
def _extract_generic_url(url: str, path_parts: list[str]) -> str | None:
"""Extract base repo URL from generic git host URL."""
# Need at least owner/repo
if len(path_parts) < 2:
return None
# Find the repo name (second path part)
repo_name = path_parts[1]
if repo_name.endswith(".git"):
repo_name = repo_name[:-4]
# Reconstruct base URL
parsed = urlparse(url)
base = f"{parsed.scheme}://{parsed.netloc}/{path_parts[0]}/{repo_name}"
return f"{base}.git"
def is_valid_clone_url(url: str) -> bool:
"""Check if URL is already a valid git clone URL.
A valid clone URL:
- Is an SSH URL (git@host:path)
- Ends with .git
- Has no browser-specific path segments
"""
# SSH URLs are always valid
if url.startswith("git@"):
return True
try:
parsed = urlparse(url)
except Exception:
return False
path = parsed.path
# Must end with .git for HTTPS
if not path.endswith(".git"):
return False
# Check for browser-specific paths
browser_paths = ["/tree/", "/blob/", "/pull/", "/issues/", "/actions/",
"-/tree/", "-/blob/", "-/merge_requests/",
"/src/"]
for bp in browser_paths:
if bp in path:
return False
return True
def parse_git_url(url: str) -> dict:
"""Parse a git URL and return detailed information.
Returns:
{
"original_url": str,
"base_url": str | None,
"is_valid_clone_url": bool,
"needs_parsing": bool,
"host": str | None,
"message": str,
"error_code": str | None,
}
"""
result = {
"original_url": url,
"base_url": None,
"is_valid_clone_url": False,
"needs_parsing": False,
"host": None,
"message": "",
"error_code": None,
}
# Check if empty
if not url or not url.strip():
result["message"] = "Please enter a URL"
result["error_code"] = "INVALID_URL"
return result
url = url.strip()
# Try to parse
try:
parsed = urlparse(url)
except Exception:
result["message"] = "Please enter a valid URL"
result["error_code"] = "INVALID_URL"
return result
# Extract host
if parsed.netloc:
result["host"] = parsed.netloc.lower()
elif url.startswith("git@"):
# SSH URL: git@host:path
parts = url.split(":", 1)
if len(parts) == 2:
result["host"] = parts[0].replace("git@", "")
else:
result["message"] = "Please enter a valid URL"
result["error_code"] = "INVALID_URL"
return result
# Check if already valid
if is_valid_clone_url(url):
result["base_url"] = url
result["is_valid_clone_url"] = True
result["needs_parsing"] = False
result["message"] = "Valid git repository URL"
return result
# Try to extract base URL
base = extract_base_repo_url(url)
if base:
result["base_url"] = base
result["needs_parsing"] = True
result["message"] = f"This looks like a browser URL. Did you mean: {base}?"
result["error_code"] = "URL_NEEDS_PARSING"
else:
result["message"] = "Could not parse this URL. Please enter a valid git repository URL."
result["error_code"] = "INVALID_URL"
return result