From 0696e92ae3ef94a6c5b8d2e62ba86d371730cb2a Mon Sep 17 00:00:00 2001 From: Russell Ballestrini Date: Fri, 2 Jan 2026 14:37:53 -0500 Subject: [PATCH] Add VCS repo cloning support (git/hg/svn/fossil) Smart source detection: like Hydra for RSS feeds, auto-detect repository URLs and clone instead of HTTP crawling. - repo.py: detect_vcs(), clone_repo(), walk_files(), symlink_to_vault() - Symlink storage: files in repo_vault/ symlink to content-addressed vault/ - Database: repo_uri, repo_path, commit_hash, vcs_type columns - Auto-detect GitHub, GitLab, Bitbucket, sr.ht, hg.mozilla.org, etc. --- CLAUDE.md | 47 +++++ database.py | 18 +- neopig.py | 162 +++++++++++++++++ repo.py | 492 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 718 insertions(+), 1 deletion(-) create mode 100644 repo.py diff --git a/CLAUDE.md b/CLAUDE.md index 8403dee..23ac529 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -59,12 +59,14 @@ async_web_fetcher.py # Async HTTP client, CrawlMode enum, robots.txt handling database.py # SQLite schema: crawl_jobs, media, media_sources tables filevault.py # FileVault 2.0 - content-addressed storage (Vault, AsyncVault) domain_vault.py # Triple vault: DomainHtmlVault, DomainMediaVault, DomainLinkpeekVault +repo.py # VCS detection, cloning, symlink-based indexing (git/hg/svn) screenshot.py # ScreenshotCapture wrapper for uri2png serp.py # FastAPI SERP server with search, live feed, crawl UI data/ # Database and state files directory neopig.db # Main SQLite database vault/ # Content-addressed media storage (9-deep hex pairs) +repo_vault/ # VCS repos with symlinks to vault (git/hg/svn clones) ``` ## Key Classes @@ -292,6 +294,51 @@ State file: `data/state/hydra-{domain}.json` - `feeds`: Discovered feed URLs with discovery timestamp - `seen_urls`: URLs already crawled (for delta detection) +## VCS Mode (repo.py) + +Like Hydra mode for feeds, VCS detection is a "smart source" that bypasses slow HTTP crawling. When neopig detects a git/hg/svn repository URL, it clones directly instead of scraping the web UI. + +```bash +# Auto-detect and clone git repo +python neopig.py https://github.com/user/repo + +# Mercurial repo +python neopig.py https://hg.mozilla.org/mozilla-central + +# SSH URL +python neopig.py git@github.com:user/repo.git + +# SVN (legacy) +python neopig.py https://svn.apache.org/repos/asf/project/trunk +``` + +**Storage Strategy: Symlinks + Content-Addressed** + +``` +repo_vault/ + github.com/ + user/ + repo/ + src/main.py -> ../../../vault/ab/cd/.../hash.py (symlink) + README.md -> ../../vault/12/34/.../hash.md + .git/ (preserved for pull) + +vault/ + ab/cd/.../hash.py (actual content, deduplicated) +``` + +**Supported VCS:** +- git (GitHub, GitLab, Bitbucket, Codeberg, sr.ht, etc.) +- hg (Mercurial) +- svn (Subversion) +- fossil + +**Database columns** (MediaSource): +- `repo_uri`: Clone URL +- `repo_path`: File path within repo +- `commit_hash`: Commit when indexed +- `vcs_type`: git/hg/svn/fossil + ## Job Management ```bash diff --git a/database.py b/database.py index 25798ca..81a1583 100644 --- a/database.py +++ b/database.py @@ -110,6 +110,11 @@ class MediaSource(Base): searchable_text = Column(Text) crawl_job_id = Column(Integer, ForeignKey('crawl_jobs.id')) discovered_at = Column(Text, nullable=False) + # VCS repo metadata (for code files from git/hg/svn clones) + repo_uri = Column(Text) # Clone URL + repo_path = Column(Text) # File path within repo + commit_hash = Column(Text) # Commit/revision when indexed + vcs_type = Column(Text) # git, hg, svn, fossil media = relationship('Media', back_populates='sources') crawl_job = relationship('CrawlJob', back_populates='media_sources') @@ -120,6 +125,7 @@ class MediaSource(Base): Index('idx_sources_job', 'crawl_job_id'), Index('idx_sources_media_uri', 'media_uri'), Index('idx_sources_page_uri', 'page_uri'), + Index('idx_sources_repo', 'repo_uri'), ) @@ -485,6 +491,12 @@ class Database: detail_content: str = "", searchable_text: str = "", score: int = SCORE_THUMBNAIL, + # VCS repo metadata + repo_uri: str = "", + repo_path: str = "", + commit_hash: str = "", + vcs_type: str = "", + keywords: list = None, ) -> None: """Create a new media record and add source context.""" now = datetime.now(timezone.utc).isoformat() @@ -521,7 +533,11 @@ class Database: detail_content=detail_content, searchable_text=searchable_text, crawl_job_id=crawl_job_id, - discovered_at=now + discovered_at=now, + repo_uri=repo_uri or None, + repo_path=repo_path or None, + commit_hash=commit_hash or None, + vcs_type=vcs_type or None, ).on_conflict_do_nothing() await session.execute(source_stmt) await session.commit() diff --git a/neopig.py b/neopig.py index 441fe8b..fa435c5 100644 --- a/neopig.py +++ b/neopig.py @@ -46,6 +46,7 @@ from filevault import AsyncVault, hash_to_path from database import Database, SCORE_SCREENSHOT, SCORE_OG_IMAGE, SCORE_THUMBNAIL, SCORE_FULL_RES from screenshot import ScreenshotCapture, ScreenshotConfig from domain_vault import VaultManager, DomainHtmlVault, DomainMediaVault, DomainLinkpeekVault, extract_media_urls +from repo import detect_vcs, clone_repo_async, pull_repo_async, get_repo_path, walk_files, get_commit_hash, get_file_language, is_binary_file from tqdm import tqdm logger = logging.getLogger(__name__) @@ -911,6 +912,19 @@ class NeoPig: """ keywords = keywords or [] + # Check if target is a VCS repository (git, hg, svn, etc.) + vcs_type, clone_url = detect_vcs(target_uri) + if vcs_type: + logger.info(f"Detected {vcs_type} repository: {clone_url}") + return await self.clone_and_index( + target_uri=target_uri, + clone_url=clone_url, + vcs_type=vcs_type, + keywords=keywords, + job_id=job_id, + quiet=quiet, + ) + # Create crawl job (unless one was provided) if job_id is None: job_id = await self.db.create_crawl_job( @@ -1430,6 +1444,154 @@ class NeoPig: except Exception as e: logger.warning(f"Screenshot failed for {page_uri}: {e}") + async def clone_and_index( + self, + target_uri: str, + clone_url: str, + vcs_type: str, + keywords: List[str] = None, + job_id: int = None, + quiet: bool = False, + pull: bool = False, + ) -> Dict[str, Any]: + """ + Clone a VCS repository and index its files. + + Like Hydra mode for feeds, VCS detection is a "smart source" that + bypasses slow HTTP crawling for high-priority ingestion. + + Args: + target_uri: Original URI + clone_url: Clone URL (may differ from target_uri) + vcs_type: 'git', 'hg', 'svn', or 'fossil' + keywords: Keywords to tag files with + job_id: Optional existing job ID + quiet: Disable progress output + pull: Update existing clone instead of fresh clone + + Returns: + Indexing statistics + """ + from filevault import content_hash + + keywords = keywords or [] + repo_vault_base = Path(self.vault.path).parent / 'repo_vault' + + # Create job + if job_id is None: + job_id = await self.db.create_crawl_job( + target_uri=target_uri, + keywords=keywords, + mode='code', + ) + + start_job_logging(job_id) + logger.info(f"Starting VCS clone job {job_id}") + logger.info(f"Repository: {clone_url}") + logger.info(f"VCS type: {vcs_type}") + logger.info(f"Keywords: {keywords}") + + start_time = datetime.now(timezone.utc) + stats = { + 'vcs_type': vcs_type, + 'clone_url': clone_url, + 'files_indexed': 0, + 'files_skipped': 0, + 'bytes_stored': 0, + 'errors': 0, + 'commit_hash': None, + } + + # Determine repo path + repo_path = get_repo_path(clone_url, repo_vault_base) + logger.info(f"Repo path: {repo_path}") + + # Clone or pull + if repo_path.exists() and (repo_path / '.git').exists(): + logger.info(f"Repository exists, pulling updates...") + success, msg = await pull_repo_async(repo_path) + if not success: + logger.error(f"Pull failed: {msg}") + stats['errors'] += 1 + else: + logger.info(f"Pull: {msg}") + else: + logger.info(f"Cloning repository...") + success, msg = await clone_repo_async(clone_url, repo_path, vcs_type, shallow=True) + if not success: + logger.error(f"Clone failed: {msg}") + await self.db.complete_crawl_job(job_id, stats) + return stats + logger.info(f"Clone: {msg}") + + # Get commit hash + stats['commit_hash'] = get_commit_hash(repo_path) + logger.info(f"Commit: {stats['commit_hash']}") + + # Walk files and index + files = list(walk_files(repo_path, include_binary=True)) + logger.info(f"Found {len(files)} files to index") + + pbar = tqdm(files, unit="files", disable=quiet) + for file_path in pbar: + try: + content = file_path.read_bytes() + h = content_hash(content) + ext = file_path.suffix or '.txt' + file_size = len(content) + + # Store in vault + vault_path = await self.vault.store(h, content, ext) + stats['bytes_stored'] += file_size + + # Replace file with symlink + rel_path = os.path.relpath(vault_path, file_path.parent) + file_path.unlink() + file_path.symlink_to(rel_path) + + # Get file metadata + repo_rel_path = str(file_path.relative_to(repo_path)) + language = get_file_language(file_path) + is_binary = is_binary_file(Path(vault_path)) + + # Create media record with repo metadata + await self.db.create_media_record( + md5_hash=h, + media_uri=f"{clone_url}/blob/HEAD/{repo_rel_path}", + page_uri=target_uri, + page_title=repo_rel_path, + media_type='code', + mime_type=f"text/{language}" if language and not is_binary else "application/octet-stream", + file_size=file_size, + alt_text=language or '', + keywords=keywords, + crawl_job_id=job_id, + score=SCORE_FULL_RES, + # VCS repo metadata + repo_uri=clone_url, + repo_path=repo_rel_path, + commit_hash=stats['commit_hash'], + vcs_type=vcs_type, + ) + + stats['files_indexed'] += 1 + pbar.set_postfix_str(f"indexed: {stats['files_indexed']}, {stats['bytes_stored'] // 1024}KB") + + except Exception as e: + logger.warning(f"Error indexing {file_path}: {e}") + stats['errors'] += 1 + stats['files_skipped'] += 1 + + pbar.close() + + # Complete job + duration = (datetime.now(timezone.utc) - start_time).total_seconds() + stats['duration_seconds'] = duration + await self.db.complete_crawl_job(job_id, stats) + + logger.info(f"VCS indexing complete: {stats['files_indexed']} files, {stats['bytes_stored'] // 1024}KB stored") + return stats + async def backfill_missing_screenshots(self, domain: str = None, crawl_job_id: int = None, fast_mode: bool = False, quiet: bool = False, create_job: bool = True): """Capture screenshots for pages that don't have them yet. diff --git a/repo.py b/repo.py new file mode 100644 index 0000000..29aea60 --- /dev/null +++ b/repo.py @@ -0,0 +1,492 @@ +""" +repo.py - VCS Repository Detection, Cloning, and Indexing + +Detects git/hg/svn/fossil repositories and clones them for indexing. +Files are stored in content-addressed vault with symlinks preserving tree structure. + +Like Hydra mode for RSS feeds, VCS detection is a "smart source" that +bypasses slow HTTP crawling for high-priority ingestion. +""" + +import asyncio +import os +import re +import subprocess +from pathlib import Path +from typing import Optional, Tuple, Iterator, AsyncIterator +from urllib.parse import urlparse + +# VCS type detection by domain +VCS_HOSTS = { + # Git hosts + 'github.com': 'git', + 'gitlab.com': 'git', + 'git.unturf.com': 'git', + 'bitbucket.org': 'git', + 'codeberg.org': 'git', + 'sr.ht': 'git', + 'git.sr.ht': 'git', + 'gitea.com': 'git', + 'gitee.com': 'git', + 'salsa.debian.org': 'git', + 'git.savannah.gnu.org': 'git', + 'git.kernel.org': 'git', + 'git.zx2c4.com': 'git', + 'git.openwrt.org': 'git', + + # Mercurial hosts + 'hg.mozilla.org': 'hg', + 'hg.python.org': 'hg', + 'hg.sr.ht': 'hg', + 'foss.heptapod.net': 'hg', + + # SVN hosts (legacy) + 'svn.apache.org': 'svn', + 'svn.code.sf.net': 'svn', +} + +# URL patterns that indicate VCS type +VCS_URL_PATTERNS = [ + (r'\.git/?$', 'git'), + (r'git@[^:]+:', 'git'), # git@github.com:user/repo + (r'/trunk/?$', 'svn'), + (r'/branches/?', 'svn'), + (r'/tags/?$', 'svn'), + (r'\.fossil$', 'fossil'), +] + +# Directories to skip when walking files +VCS_DIRS = {'.git', '.hg', '.svn', '_FOSSIL_', '.fossil-settings'} + +# Binary file extensions to skip for text indexing +BINARY_EXTENSIONS = { + '.png', '.jpg', '.jpeg', '.gif', '.bmp', '.ico', '.webp', '.avif', + '.mp4', '.webm', '.mov', '.avi', '.mkv', + '.mp3', '.wav', '.ogg', '.flac', '.m4a', + '.zip', '.tar', '.gz', '.bz2', '.xz', '.7z', '.rar', + '.exe', '.dll', '.so', '.dylib', '.a', '.o', + '.pyc', '.pyo', '.class', '.jar', + '.pdf', '.doc', '.docx', '.xls', '.xlsx', + '.ttf', '.otf', '.woff', '.woff2', '.eot', + '.sqlite', '.db', '.sqlite3', +} + + +def detect_vcs(uri: str) -> Tuple[Optional[str], Optional[str]]: + """ + Detect VCS type from URI. + + Returns: + (vcs_type, clone_url) or (None, None) if not a repo + + Examples: + >>> detect_vcs('https://github.com/user/repo') + ('git', 'https://github.com/user/repo.git') + >>> detect_vcs('git@github.com:user/repo.git') + ('git', 'git@github.com:user/repo.git') + >>> detect_vcs('https://hg.mozilla.org/mozilla-central') + ('hg', 'https://hg.mozilla.org/mozilla-central') + """ + # Check SSH-style URLs first (git@host:path) + ssh_match = re.match(r'^(git|hg)@([^:]+):(.+)$', uri) + if ssh_match: + vcs_type = ssh_match.group(1) + return vcs_type, uri + + # Parse URL + parsed = urlparse(uri) + host = parsed.netloc.lower() + + # Remove port if present + if ':' in host: + host = host.split(':')[0] + + # Check URL patterns + for pattern, vcs_type in VCS_URL_PATTERNS: + if re.search(pattern, uri): + return vcs_type, uri + + # Check known hosts + if host in VCS_HOSTS: + vcs_type = VCS_HOSTS[host] + clone_url = uri + + # Ensure .git suffix for git repos if not present + if vcs_type == 'git' and not uri.endswith('.git'): + # Strip trailing slashes and add .git + clone_url = uri.rstrip('/') + '.git' + + return vcs_type, clone_url + + return None, None + + +def get_repo_path(uri: str, base_path: Path) -> Path: + """ + Generate local path for a repository. + + Structure: base_path / domain / org / repo + + Examples: + >>> get_repo_path('https://github.com/user/repo', Path('repo_vault')) + Path('repo_vault/github.com/user/repo') + """ + # Handle SSH URLs + ssh_match = re.match(r'^(?:git|hg)@([^:]+):(.+?)(?:\.git)?$', uri) + if ssh_match: + host = ssh_match.group(1) + path = ssh_match.group(2) + else: + parsed = urlparse(uri) + host = parsed.netloc.lower() + if ':' in host: + host = host.split(':')[0] + path = parsed.path.strip('/') + + # Remove .git suffix + if path.endswith('.git'): + path = path[:-4] + + return base_path / host / path + + +def run_cmd(cmd: list, cwd: Optional[Path] = None, timeout: int = 300) -> Tuple[int, str, str]: + """Run a command and return (returncode, stdout, stderr).""" + try: + result = subprocess.run( + cmd, + cwd=cwd, + capture_output=True, + text=True, + timeout=timeout + ) + return result.returncode, result.stdout, result.stderr + except subprocess.TimeoutExpired: + return -1, '', f'Command timed out after {timeout}s' + except Exception as e: + return -1, '', str(e) + + +async def run_cmd_async(cmd: list, cwd: Optional[Path] = None, timeout: int = 300) -> Tuple[int, str, str]: + """Run a command asynchronously.""" + return await asyncio.to_thread(run_cmd, cmd, cwd, timeout) + + +def clone_repo(uri: str, dest: Path, vcs_type: str, shallow: bool = True) -> Tuple[bool, str]: + """ + Clone a repository. + + Args: + uri: Clone URL + dest: Destination path + vcs_type: 'git', 'hg', 'svn', or 'fossil' + shallow: Use shallow clone if supported (git only) + + Returns: + (success, message) + """ + dest.parent.mkdir(parents=True, exist_ok=True) + + if vcs_type == 'git': + cmd = ['git', 'clone'] + if shallow: + cmd.extend(['--depth', '1']) + cmd.extend([uri, str(dest)]) + + elif vcs_type == 'hg': + cmd = ['hg', 'clone', uri, str(dest)] + + elif vcs_type == 'svn': + cmd = ['svn', 'checkout', uri, str(dest)] + + elif vcs_type == 'fossil': + # Fossil needs a two-step process + fossil_file = dest.parent / f'{dest.name}.fossil' + ret, out, err = run_cmd(['fossil', 'clone', uri, str(fossil_file)]) + if ret != 0: + return False, f'fossil clone failed: {err}' + dest.mkdir(parents=True, exist_ok=True) + ret, out, err = run_cmd(['fossil', 'open', str(fossil_file)], cwd=dest) + if ret != 0: + return False, f'fossil open failed: {err}' + return True, 'Cloned successfully' + + else: + return False, f'Unknown VCS type: {vcs_type}' + + ret, out, err = run_cmd(cmd, timeout=600) + if ret != 0: + return False, f'{vcs_type} clone failed: {err}' + + return True, 'Cloned successfully' + + +async def clone_repo_async(uri: str, dest: Path, vcs_type: str, shallow: bool = True) -> Tuple[bool, str]: + """Clone a repository asynchronously.""" + return await asyncio.to_thread(clone_repo, uri, dest, vcs_type, shallow) + + +def pull_repo(path: Path) -> Tuple[bool, str]: + """ + Update an existing repository. + + Auto-detects VCS type from directory contents. + + Returns: + (success, message) + """ + if (path / '.git').exists(): + ret, out, err = run_cmd(['git', 'pull'], cwd=path) + if ret != 0: + return False, f'git pull failed: {err}' + return True, out.strip() or 'Already up to date' + + elif (path / '.hg').exists(): + ret, out, err = run_cmd(['hg', 'pull', '-u'], cwd=path) + if ret != 0: + return False, f'hg pull failed: {err}' + return True, out.strip() + + elif (path / '.svn').exists(): + ret, out, err = run_cmd(['svn', 'update'], cwd=path) + if ret != 0: + return False, f'svn update failed: {err}' + return True, out.strip() + + elif (path / '_FOSSIL_').exists() or (path / '.fslckout').exists(): + ret, out, err = run_cmd(['fossil', 'update'], cwd=path) + if ret != 0: + return False, f'fossil update failed: {err}' + return True, out.strip() + + else: + return False, 'No VCS directory found' + + +async def pull_repo_async(path: Path) -> Tuple[bool, str]: + """Update a repository asynchronously.""" + return await asyncio.to_thread(pull_repo, path) + + +def get_commit_hash(path: Path) -> Optional[str]: + """Get current commit/revision hash.""" + if (path / '.git').exists(): + ret, out, err = run_cmd(['git', 'rev-parse', 'HEAD'], cwd=path) + if ret == 0: + return out.strip() + + elif (path / '.hg').exists(): + ret, out, err = run_cmd(['hg', 'id', '-i'], cwd=path) + if ret == 0: + return out.strip() + + elif (path / '.svn').exists(): + ret, out, err = run_cmd(['svn', 'info', '--show-item', 'revision'], cwd=path) + if ret == 0: + return out.strip() + + elif (path / '_FOSSIL_').exists() or (path / '.fslckout').exists(): + ret, out, err = run_cmd(['fossil', 'info'], cwd=path) + if ret == 0: + for line in out.split('\n'): + if line.startswith('checkout:'): + return line.split()[1] + + return None + + +def walk_files(repo_path: Path, include_binary: bool = False) -> Iterator[Path]: + """ + Walk files in a repository, excluding VCS directories. + + Args: + repo_path: Path to repository root + include_binary: Include binary files (default: skip them) + + Yields: + Path objects for each file + """ + for root, dirs, files in os.walk(repo_path): + # Skip VCS directories + dirs[:] = [d for d in dirs if d not in VCS_DIRS] + + for name in files: + file_path = Path(root) / name + + # Skip binary files unless requested + if not include_binary: + if file_path.suffix.lower() in BINARY_EXTENSIONS: + continue + + # Skip empty files + try: + if file_path.stat().st_size == 0: + continue + except OSError: + continue + + yield file_path + + +async def symlink_to_vault( + repo_path: Path, + vault, # AsyncVault + on_progress: Optional[callable] = None +) -> AsyncIterator[Tuple[str, Path, Path]]: + """ + Replace repo files with symlinks to content-addressed vault. + + Args: + repo_path: Path to repository root + vault: AsyncVault instance for content storage + on_progress: Optional callback(hash, file_path, vault_path) + + Yields: + (md5_hash, original_path, vault_path) for each file + """ + from filevault import content_hash + + for file_path in walk_files(repo_path, include_binary=True): + try: + content = file_path.read_bytes() + h = content_hash(content) + ext = file_path.suffix or '.txt' + + # Store in vault + vault_path = await vault.store(h, content, ext) + + # Replace file with symlink + file_path.unlink() + rel_path = os.path.relpath(vault_path, file_path.parent) + file_path.symlink_to(rel_path) + + if on_progress: + on_progress(h, file_path, vault_path) + + yield h, file_path, vault_path + + except Exception as e: + # Log error but continue with other files + print(f'Error processing {file_path}: {e}') + continue + + +def is_binary_file(path: Path) -> bool: + """Check if a file is binary by extension or content.""" + if path.suffix.lower() in BINARY_EXTENSIONS: + return True + + # Check first bytes for null characters + try: + with open(path, 'rb') as f: + chunk = f.read(8192) + if b'\x00' in chunk: + return True + except Exception: + pass + + return False + + +def get_file_language(path: Path) -> Optional[str]: + """Guess programming language from file extension.""" + ext_to_lang = { + '.py': 'python', + '.js': 'javascript', + '.ts': 'typescript', + '.jsx': 'javascript', + '.tsx': 'typescript', + '.rb': 'ruby', + '.go': 'go', + '.rs': 'rust', + '.c': 'c', + '.h': 'c', + '.cpp': 'cpp', + '.hpp': 'cpp', + '.cc': 'cpp', + '.java': 'java', + '.kt': 'kotlin', + '.scala': 'scala', + '.swift': 'swift', + '.m': 'objective-c', + '.php': 'php', + '.pl': 'perl', + '.pm': 'perl', + '.sh': 'shell', + '.bash': 'shell', + '.zsh': 'shell', + '.fish': 'shell', + '.lua': 'lua', + '.r': 'r', + '.R': 'r', + '.jl': 'julia', + '.ex': 'elixir', + '.exs': 'elixir', + '.erl': 'erlang', + '.hs': 'haskell', + '.ml': 'ocaml', + '.fs': 'fsharp', + '.clj': 'clojure', + '.lisp': 'lisp', + '.el': 'elisp', + '.vim': 'vim', + '.sql': 'sql', + '.html': 'html', + '.htm': 'html', + '.css': 'css', + '.scss': 'scss', + '.sass': 'sass', + '.less': 'less', + '.json': 'json', + '.yaml': 'yaml', + '.yml': 'yaml', + '.toml': 'toml', + '.xml': 'xml', + '.md': 'markdown', + '.rst': 'rst', + '.txt': 'text', + '.ini': 'ini', + '.cfg': 'ini', + '.conf': 'conf', + '.dockerfile': 'dockerfile', + '.makefile': 'makefile', + '.cmake': 'cmake', + '.gradle': 'gradle', + '.groovy': 'groovy', + '.tf': 'terraform', + '.nix': 'nix', + '.zig': 'zig', + '.v': 'v', + '.nim': 'nim', + '.d': 'd', + '.pas': 'pascal', + '.asm': 'assembly', + '.s': 'assembly', + '.wasm': 'wasm', + '.wat': 'wat', + } + + ext = path.suffix.lower() + if ext in ext_to_lang: + return ext_to_lang[ext] + + # Check filename for special cases + name = path.name.lower() + if name == 'makefile': + return 'makefile' + elif name == 'dockerfile': + return 'dockerfile' + elif name == 'jenkinsfile': + return 'groovy' + elif name == 'gemfile': + return 'ruby' + elif name == 'rakefile': + return 'ruby' + elif name == 'vagrantfile': + return 'ruby' + elif name == 'procfile': + return 'text' + elif name.startswith('.') and name.endswith('rc'): + return 'shell' + + return None