| | |
| | |
| |
|
| | import json |
| | import shutil |
| | import hashlib |
| | from pathlib import Path |
| | from git import Repo |
| | from datetime import datetime |
| | import argparse |
| |
|
| | |
| | REPO_LIST_PATH = Path("data/metadata/repos_filtered.json") |
| | CLONE_DIR = Path("temp_repos") |
| | OUTPUT_FILE = Path("data/raw/dockerfiles.jsonl") |
| | OUTPUT_FILE.parent.mkdir(parents=True, exist_ok=True) |
| |
|
| | REMOVE_DIRS = [".git", ".github", "docs", "tests", "__pycache__", ".idea", ".vscode"] |
| |
|
| | def clean_repo(path: Path): |
| | for d in REMOVE_DIRS: |
| | shutil.rmtree(path / d, ignore_errors=True) |
| |
|
| | def compute_sha1(text: str) -> str: |
| | return hashlib.sha1(text.encode("utf-8")).hexdigest() |
| |
|
| | def is_valid_dockerfile(path: Path) -> bool: |
| | try: |
| | text = path.read_text(encoding="utf-8").strip() |
| | lines = [l.strip().lower() for l in text.splitlines() if l.strip()] |
| | if len(lines) < 5 or path.stat().st_size > 200_000: |
| | return False |
| | top_lines = lines[:10] |
| | has_from = any(l.startswith("from") for l in top_lines) |
| | has_run = any(l.startswith(("run", "cmd", "copy")) for l in lines) |
| | return has_from and has_run |
| | except Exception as e: |
| | print(f"⚠️ Błąd walidacji pliku {path}: {e}") |
| | return False |
| |
|
| | def find_dockerfiles(repo_path: Path) -> list[Path]: |
| | return [ |
| | f for f in repo_path.rglob("*") |
| | if f.name.lower() == "dockerfile" and f.is_file() |
| | ] |
| |
|
| | def clone_repo(url: str, full_name: str) -> Path | None: |
| | dest = CLONE_DIR / full_name.replace("/", "__") |
| | if dest.exists(): |
| | print(f"⚠️ Repo {full_name} już istnieje – pomijam klonowanie.") |
| | return dest |
| | try: |
| | print(f"⬇️ Klonuję {full_name}...") |
| | Repo.clone_from(url, dest, depth=1) |
| | clean_repo(dest) |
| | return dest |
| | except Exception as e: |
| | print(f"❌ Błąd klonowania {full_name}: {e}") |
| | return None |
| |
|
| | def main(): |
| | parser = argparse.ArgumentParser() |
| | parser.add_argument("--purge", action="store_true", help="Usuń repozytorium po ekstrakcji") |
| | args = parser.parse_args() |
| |
|
| | with open(REPO_LIST_PATH) as f: |
| | repos = json.load(f) |
| |
|
| | saved, skipped = 0, 0 |
| | seen_hashes = set() |
| |
|
| | with open(OUTPUT_FILE, "w", encoding="utf-8") as out_f: |
| | for repo in repos: |
| | full_name = repo["fullName"] |
| | url = repo["url"] |
| | repo_path = clone_repo(url, full_name) |
| | if not repo_path: |
| | continue |
| |
|
| | for file in find_dockerfiles(repo_path): |
| | if not is_valid_dockerfile(file): |
| | skipped += 1 |
| | continue |
| |
|
| | try: |
| | lines = file.read_text(encoding="utf-8").strip().splitlines() |
| | lines = [l.rstrip() for l in lines if l.strip()] |
| | file_id = compute_sha1("\n".join(lines)) |
| | if file_id in seen_hashes: |
| | skipped += 1 |
| | continue |
| | seen_hashes.add(file_id) |
| |
|
| | json.dump({ |
| | "repo": full_name, |
| | "path": str(file.relative_to(repo_path)), |
| | "file_id": file_id, |
| | "content": lines, |
| | "size_bytes": file.stat().st_size, |
| | "line_count": len(lines), |
| | "valid": True, |
| | "cloned_at": datetime.now().isoformat() |
| | }, out_f) |
| | out_f.write("\n") |
| | saved += 1 |
| |
|
| | except Exception as e: |
| | print(f"⚠️ Błąd przy zapisie {file}: {e}") |
| | skipped += 1 |
| |
|
| | if args.purge: |
| | shutil.rmtree(repo_path, ignore_errors=True) |
| |
|
| | print(f"\n✅ Zapisano {saved} poprawnych Dockerfile do {OUTPUT_FILE}") |
| | print(f"🚫 Pominięto {skipped} plików (nieważne, błędne, zduplikowane)") |
| |
|
| | if __name__ == "__main__": |
| | main() |
| |
|
