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.

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.