first commit

This commit is contained in:
jsahn
2026-01-08 15:50:54 +09:00
commit 64b646666f
13 changed files with 1943 additions and 0 deletions
+48
View File
@@ -0,0 +1,48 @@
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
.venv/
venv/
ENV/
env/
# IDE
.vscode/
.idea/
*.swp
*.swo
# Git
.git/
.gitignore
# Data and logs (will be mounted as volumes)
data/
logs/
legacy_data/
pickle/
# Notebooks and experiments
*.ipynb
experiments/
experiments 2/
get_embedding.ipynb
# Documentation
docs/
*.md
!README.md
# OS files
.DS_Store
Thumbs.db
# Claude
.claude/
# Misc
*.tmp
*.bak
Executable
+16
View File
@@ -0,0 +1,16 @@
.env
**/.venv/
**/__pycache__/
.claude/
.DS_Store
._.DS_Store
**/.DS_Store
**/._.DS_Store
.gitea
data/
experiments/
experiments\ 2/
logs/
legacy_data/
pickles/
result.md
Executable
+66
View File
@@ -0,0 +1,66 @@
import math
from collections import Counter
import numpy as np
import nltk
from nltk.corpus import stopwords
nltk.download('stopwords')
def preprocess_text(text):
stop_words = set(stopwords.words('english'))
words = nltk.word_tokenize(text.lower())
return [w for w in words if w.isalnum() and w not in stop_words]
def compute_idf(corpus):
idf = {}
N = len(corpus)
for document in corpus:
words = set(document)
for word in words:
if word in idf:
idf[word] += 1
else:
idf[word] = 1
for word, freq in idf.items():
idf[word] = math.log((N - freq + 0.5) / (freq + 0.5) + 1)
return idf
def compute_average_length(corpus):
total_length = sum(len(document) for document in corpus)
return total_length / len(corpus)
def get_bm25_vectors(corpus):
processed_corpus = [preprocess_text(doc) for doc in corpus]
idf = compute_idf(processed_corpus)
avg_length = compute_average_length(processed_corpus)
document_vectors = {}
for doc_,doc in zip(corpus,processed_corpus):
vector = []
document_counts = Counter(doc)
document_length = len(doc)
for term, idf_value in idf.items():
tf = document_counts.get(term, 0)
numerator = idf_value * tf * (1.5 + 1)
denominator = tf + 1.5 * (1 - 0.75 + 0.75 * (document_length / avg_length))
vector.append(numerator / denominator)
document_vectors[doc_] = vector
return document_vectors
# # Example usage
# corpus = [
# "This is the first document.",
# "This document is the second document.",
# "And this is the third one.",
# "Is this the first document?",
# ]
# bm25_vectors = get_bm25_vectors(corpus)
# print("BM25 Vectors:")
# print(bm25_vectors)
+782
View File
@@ -0,0 +1,782 @@
#!/usr/bin/env python3
"""
MCP server that crawls GitHub issues, embeds the reports, and recommends bug
locations by reusing the crawl/embedding utilities plus the RQ1 workflow.
Uses sqlite-vec for efficient vector similarity search with minimal memory footprint.
"""
from __future__ import annotations
import json
import logging
import os
import re
import sqlite3
import struct
import time
from collections import Counter, defaultdict
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence, Tuple
import numpy as np
import sqlite_vec
from dotenv import load_dotenv
from openai import OpenAI
from tqdm import tqdm
from crawl_issues_With_diffs import crawl_issues_with_diffs
try:
from fastmcp import FastMCP, Context
except ImportError as exc: # pragma: no cover - handled at runtime
raise SystemExit(
"The modelcontextprotocol package is required. "
"Install it with `pip install modelcontextprotocol`."
) from exc
load_dotenv()
# Configure logging
LOG_DIR = Path(os.getenv("BUGFS_LOG_DIR", "logs"))
LOG_DIR.mkdir(parents=True, exist_ok=True)
LOG_FILE = LOG_DIR / "bugfs_mcp.log"
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
handlers=[
logging.FileHandler(LOG_FILE, encoding="utf-8"),
],
)
logger = logging.getLogger("bugfs_mcp")
DEFAULT_MODEL = os.getenv("BUGFS_EMBED_MODEL", "text-embedding-3-small")
CHUNK_CHAR_LIMIT = int(os.getenv("BUGFS_CHUNK_CHAR_LIMIT", "3000"))
DATA_DIR = Path(os.getenv("BUGFS_DATA_DIR", "data"))
DATA_DIR.mkdir(parents=True, exist_ok=True)
# Embedding dimension for text-embedding-3-small
EMBEDDING_DIM = 1536
# Regex to extract hunk header: line info and context
HUNK_HEADER_REGEX = re.compile(
r"@@ -\d+,\d+ \+(?P<start>\d+),(?P<length>\d+) @@(?:\s+(?P<context>.+))?$",
flags=re.MULTILINE,
)
# Patterns to extract method/function name from the hunk context
# Python: def function_name
PYTHON_FUNC_PATTERN = re.compile(r"def\s+(?P<name>\w+)")
# Java/Groovy/Kotlin method: modifiers + return_type + method_name(
# e.g., "public void methodName(", "private CloseableHttpClient create("
JAVA_METHOD_PATTERN = re.compile(
r"(?:public|private|protected|static|final|abstract|synchronized|void|\w+)\s+(?P<name>\w+)\s*\("
)
# Class/interface/enum definition
CLASS_PATTERN = re.compile(r"(?:class|interface|enum)\s+(?P<name>\w+)")
def slugify_repo(owner: str, repo: str) -> str:
return f"{owner}_{repo}".replace("/", "_")
def ensure_text(value: Optional[str]) -> str:
return (value or "").strip()
def build_issue_text(issue: Dict[str, Any]) -> str:
title = ensure_text(issue.get("title"))
body = ensure_text(issue.get("body"))
text = f"{title}\n\n{body}".strip()
return text
def split_into_chunks(text: str, limit: int = CHUNK_CHAR_LIMIT) -> List[str]:
if not text:
return []
return [text[i : i + limit] for i in range(0, len(text), limit)]
def sanitize_issue_body(issue: Dict[str, Any], min_word_count: int = 1) -> Optional[str]:
text = build_issue_text(issue)
text = re.sub(r"\s+", " ", text)
if len(text.split()) < min_word_count:
return None
return text
def normalize_filename(path_str: str) -> str:
path_str = path_str.replace("\\", "/")
parts = [p for p in path_str.split("/") if p]
if not parts:
return path_str
return parts[-1]
@dataclass(frozen=True)
class IssueLocation:
start: int
length: int
function: str
file: str
@property
def label(self) -> str:
return f"{self.file}:{self.function}"
def extract_method_from_context(context: str) -> Optional[str]:
"""Extract method/function name from git diff hunk context."""
if not context:
return None
# Try Python function pattern
match = PYTHON_FUNC_PATTERN.search(context)
if match:
return match.group("name")
# Try Java/Groovy/Kotlin method pattern
match = JAVA_METHOD_PATTERN.search(context)
if match:
name = match.group("name")
# Filter out common false positives (keywords that look like method names)
if name not in ("if", "for", "while", "switch", "catch", "synchronized", "return", "new", "throw"):
return name
# Try class/interface/enum pattern
match = CLASS_PATTERN.search(context)
if match:
return match.group("name")
return None
def extract_locations(issue: Dict[str, Any]) -> List[IssueLocation]:
"""Extract file:function locations from git diff patches.
Parses the hunk header context (text after @@ ... @@) to find the enclosing
function/method name. Supports Python, Java, Groovy, Kotlin, and class definitions.
"""
locations: List[IssueLocation] = []
seen: set[tuple[str, str]] = set() # Avoid duplicates
for commit in issue.get("commits", []):
filename = normalize_filename(commit.get("filename", "unknown"))
patch = commit.get("patch")
if not patch:
continue
# Parse each hunk header
for match in HUNK_HEADER_REGEX.finditer(patch):
start = int(match.group("start"))
length = int(match.group("length"))
context = match.group("context") or ""
# Try to extract method/function name from context
method_name = extract_method_from_context(context)
if method_name:
key = (filename, method_name)
if key not in seen:
seen.add(key)
locations.append(
IssueLocation(start=start, length=length, function=method_name, file=filename)
)
return locations
class Embedder:
def __init__(self, model: str = DEFAULT_MODEL):
self.model = model
self._client: Optional[OpenAI] = None
@property
def client(self) -> OpenAI:
if self._client is None:
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise RuntimeError("OPENAI_API_KEY is not set. Please configure your OpenAI API key.")
self._client = OpenAI(api_key=api_key)
return self._client
def embed_text(self, text: str) -> np.ndarray:
chunks = split_into_chunks(text)
if not chunks:
raise ValueError("Cannot embed empty text.")
last_error: Optional[Exception] = None
for attempt in range(5):
try:
response = self.client.embeddings.create(model=self.model, input=chunks)
vectors = [np.array(item.embedding, dtype=np.float32) for item in response.data]
avg_vector = np.mean(vectors, axis=0)
return avg_vector.astype(np.float32)
except Exception as exc:
time.sleep(1 + attempt)
last_error = exc
raise RuntimeError(f"Failed to embed text after retries: {last_error}")
def serialize_f32(vector: np.ndarray) -> bytes:
"""Serialize a numpy float32 vector to bytes for sqlite-vec."""
return struct.pack(f"{len(vector)}f", *vector)
def normalize_vector(vector: np.ndarray) -> np.ndarray:
"""Normalize vector for cosine similarity."""
norm = np.linalg.norm(vector)
if norm > 0:
return vector / norm
return vector
class SqliteVecCorpus:
"""Efficient vector search using sqlite-vec with minimal memory footprint."""
def __init__(self, db_path: Path):
self.db_path = db_path
self._conn: Optional[sqlite3.Connection] = None
@property
def conn(self) -> sqlite3.Connection:
if self._conn is None:
self._conn = sqlite3.connect(str(self.db_path))
self._conn.enable_load_extension(True)
sqlite_vec.load(self._conn)
self._conn.enable_load_extension(False)
return self._conn
def close(self):
if self._conn:
self._conn.close()
self._conn = None
def _init_schema(self):
"""Initialize database schema."""
self.conn.executescript(f"""
-- Raw issues table (stores full issue data as JSON)
CREATE TABLE IF NOT EXISTS raw_issues (
id INTEGER PRIMARY KEY,
issue_id TEXT UNIQUE,
data TEXT NOT NULL,
created_at TEXT,
updated_at TEXT
);
-- Indexed issues table (for vector search)
CREATE TABLE IF NOT EXISTS issues (
id INTEGER PRIMARY KEY,
issue_id TEXT UNIQUE,
content TEXT,
locations TEXT
);
-- Vector index
CREATE VIRTUAL TABLE IF NOT EXISTS issue_vectors USING vec0(
id INTEGER PRIMARY KEY,
embedding float[{EMBEDDING_DIM}]
);
CREATE INDEX IF NOT EXISTS idx_issue_id ON issues(issue_id);
CREATE INDEX IF NOT EXISTS idx_raw_issue_id ON raw_issues(issue_id);
""")
self.conn.commit()
def count(self) -> int:
"""Get number of indexed issues."""
cursor = self.conn.execute("SELECT COUNT(*) FROM issues")
return cursor.fetchone()[0]
def count_raw_issues(self) -> int:
"""Get number of raw issues stored."""
cursor = self.conn.execute("SELECT COUNT(*) FROM raw_issues")
return cursor.fetchone()[0]
def save_raw_issues(self, issues: Sequence[Dict[str, Any]]) -> int:
"""Save raw issues to database. Returns number of issues saved."""
cursor = self.conn.cursor()
saved = 0
for issue in tqdm(issues, desc="Saving raw issues", unit="issue"):
issue_id = str(issue.get("id", ""))
if not issue_id:
continue
created_at = issue.get("created_at")
updated_at = issue.get("updated_at")
cursor.execute(
"""INSERT OR REPLACE INTO raw_issues (issue_id, data, created_at, updated_at)
VALUES (?, ?, ?, ?)""",
(issue_id, json.dumps(issue, ensure_ascii=False), created_at, updated_at)
)
saved += 1
if saved % 1000 == 0:
self.conn.commit()
self.conn.commit()
return saved
def load_raw_issues(self) -> List[Dict[str, Any]]:
"""Load all raw issues from database."""
cursor = self.conn.execute("SELECT data FROM raw_issues")
return [json.loads(row[0]) for row in cursor.fetchall()]
def get_raw_issue(self, issue_id: str) -> Optional[Dict[str, Any]]:
"""Get a single raw issue by ID."""
cursor = self.conn.execute(
"SELECT data FROM raw_issues WHERE issue_id = ?", (issue_id,)
)
row = cursor.fetchone()
return json.loads(row[0]) if row else None
def get_raw_issue_ids(self) -> set:
"""Get set of all raw issue IDs."""
cursor = self.conn.execute("SELECT issue_id FROM raw_issues")
return {row[0] for row in cursor.fetchall()}
def save_embedding(self, issue_id: str, embedding: List[float], model: str) -> None:
"""Save embedding for an issue."""
cursor = self.conn.cursor()
# Get existing data, update it, and save back
cursor.execute("SELECT data FROM raw_issues WHERE issue_id = ?", (issue_id,))
row = cursor.fetchone()
if row:
data = json.loads(row[0])
data["embedding"] = embedding
data["embedding_model"] = model
cursor.execute(
"UPDATE raw_issues SET data = ? WHERE issue_id = ?",
(json.dumps(data, ensure_ascii=False), issue_id)
)
self.conn.commit()
def get_embedded_issue_ids(self) -> set:
"""Get set of issue IDs that have embeddings."""
cursor = self.conn.execute(
"SELECT issue_id FROM raw_issues WHERE json_extract(data, '$.embedding') IS NOT NULL"
)
return {row[0] for row in cursor.fetchall()}
def iter_raw_issues(self, batch_size: int = 100):
"""Iterate over raw issues in batches to reduce memory usage."""
cursor = self.conn.execute("SELECT COUNT(*) FROM raw_issues")
total = cursor.fetchone()[0]
for offset in range(0, total, batch_size):
cursor = self.conn.execute(
"SELECT issue_id, data FROM raw_issues LIMIT ? OFFSET ?",
(batch_size, offset)
)
for row in cursor.fetchall():
yield row[0], json.loads(row[1])
def rebuild_vector_index(self, min_word_count: int = 1) -> int:
"""Rebuild vector index from raw_issues with embeddings."""
# Clear existing index
self.conn.execute("DELETE FROM issues")
self.conn.execute("DELETE FROM issue_vectors")
self.conn.commit()
batch_issues = []
batch_vectors = []
indexed = 0
for issue_id, issue in tqdm(
self.iter_raw_issues(batch_size=500),
desc="Building vector index",
unit="issue",
total=self.count_raw_issues()
):
embedding = issue.get("embedding")
if not embedding:
continue
# Handle double-encoded JSON strings
if isinstance(embedding, str):
try:
embedding = json.loads(embedding)
except json.JSONDecodeError:
continue
locations = extract_locations(issue)
if not locations:
continue
text = sanitize_issue_body(issue, min_word_count=min_word_count)
if not text:
continue
locations_as_dicts = [
{"start": loc.start, "length": loc.length, "function": loc.function, "file": loc.file}
for loc in locations
]
vec = normalize_vector(np.asarray(embedding, dtype=np.float32))
batch_issues.append((issue_id, text, json.dumps(locations_as_dicts)))
batch_vectors.append(serialize_f32(vec))
indexed += 1
if len(batch_issues) >= 1000:
self._batch_insert(batch_issues, batch_vectors)
batch_issues = []
batch_vectors = []
if batch_issues:
self._batch_insert(batch_issues, batch_vectors)
return indexed
@classmethod
def build_from_issues(
cls, db_path: Path, issues: Sequence[Dict[str, Any]], min_word_count: int = 1
) -> "SqliteVecCorpus":
"""Build sqlite-vec index from issues with embeddings."""
# Remove existing db if exists
if db_path.exists():
db_path.unlink()
corpus = cls(db_path)
corpus._init_schema()
batch_issues = []
batch_vectors = []
for issue in tqdm(issues, desc="Indexing issues", unit="issue"):
embedding = issue.get("embedding")
if not embedding:
continue
locations = extract_locations(issue)
if not locations:
continue
text = sanitize_issue_body(issue, min_word_count=min_word_count)
if not text:
continue
locations_as_dicts = [
{"start": loc.start, "length": loc.length, "function": loc.function, "file": loc.file}
for loc in locations
]
# Normalize embedding for cosine similarity
vec = normalize_vector(np.asarray(embedding, dtype=np.float32))
batch_issues.append((
str(issue["id"]),
text,
json.dumps(locations_as_dicts)
))
batch_vectors.append(serialize_f32(vec))
# Batch insert every 1000 records
if len(batch_issues) >= 1000:
corpus._batch_insert(batch_issues, batch_vectors)
batch_issues = []
batch_vectors = []
# Insert remaining
if batch_issues:
corpus._batch_insert(batch_issues, batch_vectors)
return corpus
def _batch_insert(self, issues: List[Tuple], vectors: List[bytes]):
"""Batch insert issues and vectors."""
cursor = self.conn.cursor()
# Insert issues
cursor.executemany(
"INSERT OR REPLACE INTO issues (issue_id, content, locations) VALUES (?, ?, ?)",
issues
)
# Get the rowids
issue_ids = [i[0] for i in issues]
placeholders = ",".join("?" * len(issue_ids))
cursor.execute(
f"SELECT id, issue_id FROM issues WHERE issue_id IN ({placeholders})",
issue_ids
)
id_map = {row[1]: row[0] for row in cursor.fetchall()}
# Insert vectors
vector_data = [(id_map[issues[i][0]], vectors[i]) for i in range(len(issues))]
cursor.executemany(
"INSERT OR REPLACE INTO issue_vectors (id, embedding) VALUES (?, ?)",
vector_data
)
self.conn.commit()
def add_issue(self, issue_id: str, content: str, locations: List[Dict], embedding: np.ndarray):
"""Add a single issue to the index."""
vec = normalize_vector(embedding.astype(np.float32))
cursor = self.conn.cursor()
cursor.execute(
"INSERT OR REPLACE INTO issues (issue_id, content, locations) VALUES (?, ?, ?)",
(issue_id, content, json.dumps(locations))
)
rowid = cursor.lastrowid
cursor.execute(
"INSERT OR REPLACE INTO issue_vectors (id, embedding) VALUES (?, ?)",
(rowid, serialize_f32(vec))
)
self.conn.commit()
def recommend(
self,
query_vector: np.ndarray,
neighbor_count: int = 10,
max_candidates: int = 10,
) -> Dict[str, Any]:
if self.count() == 0:
raise RuntimeError("No issues with embeddings and locations are available.")
# Normalize query vector for cosine similarity
query = normalize_vector(query_vector.astype(np.float32))
query_bytes = serialize_f32(query)
# Search using sqlite-vec
cursor = self.conn.execute("""
SELECT
i.issue_id,
i.locations,
v.distance
FROM issue_vectors v
JOIN issues i ON i.id = v.id
WHERE v.embedding MATCH ? AND k = ?
ORDER BY v.distance
""", (query_bytes, neighbor_count))
neighbors: List[Dict[str, Any]] = []
location_votes: Counter[str] = Counter()
location_sources: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
for row in cursor.fetchall():
issue_id, locations_json, distance = row
# sqlite-vec returns L2 distance, convert to similarity
# For normalized vectors: cosine_similarity = 1 - (distance^2 / 2)
similarity = float(1 - (distance ** 2) / 2)
neighbors.append({"issue_id": issue_id, "similarity": similarity})
locations = json.loads(locations_json)
for loc in locations:
label = f"{loc['file']}:{loc['function']}"
location_votes[label] += 1
location_sources[label].append({
"issue_id": issue_id,
"file": loc["file"],
"function": loc["function"],
"start": loc["start"],
"length": loc["length"],
"similarity": similarity,
})
recommendations: List[Dict[str, Any]] = []
for label, votes in sorted(location_votes.items(), key=lambda item: (-item[1], item[0]))[:max_candidates]:
sources = location_sources[label][:5]
recommendations.append({"location": label, "votes": votes, "sources": sources})
return {"neighbors": neighbors, "recommendations": recommendations}
class DatasetManager:
def __init__(self, data_dir: Path):
self.data_dir = data_dir
def db_path(self, slug: str, model: str = DEFAULT_MODEL) -> Path:
"""Path for sqlite-vec database."""
safe_model = model.replace("/", "_")
return self.data_dir / f"{slug}_{safe_model}.db"
def get_or_create_corpus(self, slug: str, model: str = DEFAULT_MODEL) -> SqliteVecCorpus:
"""Get or create a corpus for the given slug and model."""
db_file = self.db_path(slug, model)
corpus = SqliteVecCorpus(db_file)
corpus._init_schema()
return corpus
def crawl(self, owner: str, repo: str, token: str) -> Tuple[Path, int]:
slug = slugify_repo(owner, repo)
# Get existing issues from sqlite
db_file = self.db_path(slug)
corpus = self.get_or_create_corpus(slug)
existing_ids = corpus.get_raw_issue_ids()
# Convert to list format for crawler compatibility
existing = None
if existing_ids:
existing = corpus.load_raw_issues()
results = crawl_issues_with_diffs(
owner=owner,
repo=repo,
token=token,
output_path=str(db_file), # Not used for saving, just for logging
existing_results=existing,
)
# Save to sqlite
corpus.save_raw_issues(results)
corpus.close()
return db_file, len(results)
async def embed(
self, owner: str, repo: str, embedder: Embedder, ctx: Optional[Context] = None
) -> Dict[str, Any]:
slug = slugify_repo(owner, repo)
db_file = self.db_path(slug, embedder.model)
corpus = self.get_or_create_corpus(slug, embedder.model)
if corpus.count_raw_issues() == 0:
raise FileNotFoundError(f"No crawled issues found. Expected {db_file}.")
# Get already embedded issue IDs
embedded_ids = corpus.get_embedded_issue_ids()
total_issues = corpus.count_raw_issues()
embedded = 0
processed = 0
for issue_id, issue in corpus.iter_raw_issues(batch_size=100):
processed += 1
# Report progress to MCP client
if ctx:
await ctx.report_progress(progress=processed, total=total_issues)
# Skip already embedded
if issue_id in embedded_ids:
continue
text = sanitize_issue_body(issue)
if not text:
continue
vector = embedder.embed_text(text)
corpus.save_embedding(issue_id, vector.tolist(), embedder.model)
embedded += 1
logger.info(f"Embedded {processed}/{total_issues} (new: {embedded})")
# Rebuild vector index
logger.info(f"Building sqlite-vec index for {owner}/{repo}...")
if ctx:
await ctx.report_progress(progress=total_issues, total=total_issues)
indexed_count = corpus.rebuild_vector_index()
logger.info(f"sqlite-vec index saved to {db_file}")
corpus.close()
return {
"output_path": str(db_file),
"total_issues": total_issues,
"embedded_issues": len(embedded_ids) + embedded,
"indexed_issues": indexed_count,
"new_embeddings": embedded,
}
def load_corpus(self, owner: str, repo: str, model: str, min_word_count: int = 1) -> SqliteVecCorpus:
slug = slugify_repo(owner, repo)
db_file = self.db_path(slug, model)
if not db_file.exists():
raise FileNotFoundError(
f"No sqlite-vec database found at {db_file}. Please run embed_repo first."
)
logger.info(f"Loading sqlite-vec index from {db_file}")
corpus = SqliteVecCorpus(db_file)
# Rebuild index if empty but has raw issues with embeddings
if corpus.count() == 0 and corpus.count_raw_issues() > 0:
logger.info("Vector index empty, rebuilding from raw issues...")
corpus.rebuild_vector_index(min_word_count=min_word_count)
return corpus
dataset_manager = DatasetManager(DATA_DIR)
embedder = Embedder(DEFAULT_MODEL)
server = FastMCP("bugfs")
@server.tool()
def crawl_repo(
owner: str,
repo: str,
state: str = "closed",
) -> Dict[str, Any]:
"""Crawl GitHub issues plus associated commit diffs for the target repository."""
token = os.getenv("GITHUB_TOKEN")
if not token:
raise RuntimeError("GITHUB_TOKEN is required to crawl issues.")
path, count = dataset_manager.crawl(owner, repo, token)
return {"issues_path": str(path), "issues_crawled": count, "state": state}
@server.tool()
async def embed_repo(
owner: str,
repo: str,
model: str = "",
ctx: Context = None,
) -> Dict[str, Any]:
"""Embed every crawled issue (title + body) using the chosen embedding model."""
if model and model != embedder.model:
custom_embedder = Embedder(model)
else:
custom_embedder = embedder
result = await dataset_manager.embed(owner, repo, custom_embedder, ctx=ctx)
return result
@server.tool(name="run_BUGFS")
def run_BUGFS(
owner: str,
repo: str,
title: str,
body: str,
neighbors: int = 10,
max_candidates: int = 10,
model: str = "",
) -> Dict[str, Any]:
"""
Recommend bug locations (file:function) by retrieving similar historical issues.
"""
chosen_model = model if model else embedder.model
corpus = dataset_manager.load_corpus(owner, repo, chosen_model)
query_text = re.sub(r"\s+", " ", f"{title.strip()}\n\n{body.strip()}".strip()).strip()
if not query_text:
raise ValueError("title+body must contain text.")
query_embedder = embedder if chosen_model == embedder.model else Embedder(chosen_model)
query_vector = query_embedder.embed_text(query_text)
return corpus.recommend(query_vector, neighbor_count=neighbors, max_candidates=max_candidates)
def main() -> None:
import argparse
parser = argparse.ArgumentParser(description="BUGFS MCP Server")
parser.add_argument("--host", default="0.0.0.0", help="Host to bind (default: 0.0.0.0)")
parser.add_argument("--port", type=int, default=8000, help="Port to bind (default: 8000)")
args = parser.parse_args()
server.run(transport="sse", host=args.host, port=args.port)
if __name__ == "__main__":
main()
+31
View File
@@ -0,0 +1,31 @@
FROM python:3.11-slim
WORKDIR /app
# Install system dependencies for sqlite-vec
RUN apt-get update && apt-get install -y --no-install-recommends \
build-essential \
&& rm -rf /var/lib/apt/lists/*
# Copy requirements first for better caching
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY BUGFS_mcp_server.py .
COPY crawl_issues_With_diffs.py .
COPY BM25.py .
COPY embedding.py .
# Create data and logs directories
RUN mkdir -p /app/data /app/logs
# Set environment variables
ENV BUGFS_DATA_DIR=/app/data
ENV BUGFS_LOG_DIR=/app/logs
# Expose port 1026
EXPOSE 1026
# Run the MCP server
CMD ["python", "BUGFS_mcp_server.py", "--host", "0.0.0.0", "--port", "1026"]
Executable
+40
View File
@@ -0,0 +1,40 @@
# BUGFS: Method-Level Bug Localization by Counting Method Modifications of Similar Bug Reports
## 📖 Overview
BUGFS (**BUG** location recommendation using method **F**requency of **S**imilar bug reports) is a novel approach for **method-level bug localization**.
It combines **semantic similarity** between bug reports and the **historical frequency of method modifications** to recommend the most likely buggy methods.
> 🧠 **Key idea:**
> If a method was frequently modified in similar past bug reports, it is likely to require modification again for a new bug.
Compared to state-of-the-art methods such as **FineLocator** and **MblShl**, BUGFS achieves:
- **1.5× higher Precision@10**, **1.2× higher Recall@10**
- **3.0× higher MAP**, **3.4× higher MRR**
- Consistent performance across **Java (Ant, AspectJ, Maven)** and **Python (Django, Flask, Pandas)** projects.
<p align="center">
<img src="docs/bugfs_overview.png" width="500" alt="BUGFS Overview">
</p>
## ⚙️ MCP Bug Localization Server
The new `bug_locator_mcp_server.py` exposes a Model Context Protocol (MCP) server that automates everything from crawling issues to recommending buggy methods:
| Tool | Purpose |
| --- | --- |
| `crawl_repo` | Uses `crawl_issues_With_diffs.py` to download GitHub issues plus their fixing commit patches. |
| `embed_repo` | Reuses `embedding.py` logic to embed each issue’s title/body with OpenAI embeddings. |
| `run_BUGFS` | Implements the `run_rq1` methodology to retrieve similar issues and rank file:function locations by vote frequency. |
### Setup
1. `pip install modelcontextprotocol openai python-dotenv requests numpy`
2. Add `OPENAI_API_KEY` and `GITHUB_TOKEN` to your environment (or a `.env` file).
### Running locally
```
python bug_locator_mcp_server.py
```
Once the server is running you can add it to any MCP-compatible client. Typical workflow:
1. `crawl_repo(owner="pallets", repo="flask")`
2. `embed_repo(owner="pallets", repo="flask", model="text-embedding-3-small")`
3. `run_BUGFS(owner="pallets", repo="flask", title="Bug title", body="Bug body...")`
+344
View File
@@ -0,0 +1,344 @@
#!/usr/bin/env python3
"""
GitHub Issue + Commit Diff Crawler
- flask_issues.json 과 100% 동일한 구조로 JSON 생성
[
{
"id": <issue_id>,
"url": "<issue_api_url>",
"title": "<issue_title>",
"body": "<issue_body>",
"commits": [
{
"filename": "<file_path>",
"patch": "<unified_diff_patch>",
"date": "<ISO8601_commit_date>"
},
...
]
},
...
]
사용 예:
export GITHUB_TOKEN=xxxx
python crawl_issues_with_diffs.py pallets flask flask_issues.json
"""
import os
import sys
import time
import json
import requests
from typing import List, Dict, Any, Set, Optional
import dotenv
from tqdm import tqdm
dotenv.load_dotenv()
os.environ["GITHUB_TOKEN"] = os.getenv("GITHUB_TOKEN", "")
GITHUB_API = "https://api.github.com"
def make_session(token: str) -> requests.Session:
s = requests.Session()
s.headers.update(
{
"Authorization": f"token {token}",
"Accept": "application/vnd.github+json",
}
)
return s
def fetch_issues(
session: requests.Session,
owner: str,
repo: str,
state: str = "all",
per_page: int = 100,
) -> List[Dict[str, Any]]:
"""모든 이슈(이슈 + PR)를 페이징 하면서 가져옴."""
issues: List[Dict[str, Any]] = []
page = 1
pbar = tqdm(desc=f"Fetching issues from {owner}/{repo}", unit="page")
while True:
url = f"{GITHUB_API}/repos/{owner}/{repo}/issues"
params = {
"state": state,
"per_page": per_page,
"page": page,
}
try:
resp = session.get(url, params=params)
except requests.RequestException as e:
tqdm.write(f"[ERROR] fetch_issues page={page} exception: {e}")
break
if resp.status_code != 200:
tqdm.write(f"[WARN] fetch_issues page={page} status={resp.status_code}")
break
batch = resp.json()
if not batch:
break
issues.extend(batch)
pbar.update(1)
pbar.set_postfix({"total_issues": len(issues)})
page += 1
time.sleep(0.2)
pbar.close()
return issues
def fetch_issue_timeline(
session: requests.Session,
owner: str,
repo: str,
issue_number: int,
) -> List[Dict[str, Any]]:
"""
Issue timeline API 로 cross-referenced PR 등을 찾음.
"""
url = f"{GITHUB_API}/repos/{owner}/{repo}/issues/{issue_number}/timeline"
headers = {
"Accept": "application/vnd.github.mockingbird-preview+json"
}
try:
resp = session.get(url, headers=headers)
except requests.RequestException as e:
print(f"[WARN] fetch_issue_timeline #{issue_number} exception: {e}")
return []
if resp.status_code != 200:
print(f"[WARN] fetch_issue_timeline #{issue_number} status={resp.status_code}")
return []
events = resp.json()
time.sleep(0.2)
return events
def collect_pr_numbers_for_issue(
session: requests.Session,
owner: str,
repo: str,
issue: Dict[str, Any],
) -> Set[int]:
"""
한 issue 에 대해 연관된 PR 번호들을 수집:
1) 이슈 자체가 PR인 경우
2) timeline cross-referenced 로 연결된 PR들
"""
pr_numbers: Set[int] = set()
issue_number = issue["number"]
# 1) 이슈가 곧 PR인 경우
if "pull_request" in issue:
pr_numbers.add(issue_number)
# 2) 타임라인에서 cross-referenced PR 찾기
events = fetch_issue_timeline(session, owner, repo, issue_number)
for ev in events:
if ev.get("event") == "cross-referenced":
src_issue = ev.get("source", {}).get("issue", {})
if src_issue.get("pull_request"):
pr_num = src_issue.get("number")
if isinstance(pr_num, int):
pr_numbers.add(pr_num)
return pr_numbers
def fetch_pr_commits(
session: requests.Session,
owner: str,
repo: str,
pr_number: int,
) -> List[Dict[str, Any]]:
"""PR에 포함된 commit 리스트 가져오기."""
url = f"{GITHUB_API}/repos/{owner}/{repo}/pulls/{pr_number}/commits"
try:
resp = session.get(url)
except requests.RequestException as e:
print(f"[WARN] fetch_pr_commits PR#{pr_number} exception: {e}")
return []
if resp.status_code != 200:
print(f"[WARN] fetch_pr_commits PR#{pr_number} status={resp.status_code}")
return []
commits = resp.json()
time.sleep(0.2)
return commits
def fetch_commit_files(
session: requests.Session,
owner: str,
repo: str,
sha: str,
) -> Optional[Dict[str, Any]]:
"""commit 상세(파일 + patch 포함) 가져오기."""
url = f"{GITHUB_API}/repos/{owner}/{repo}/commits/{sha}"
try:
resp = session.get(url)
except requests.RequestException as e:
print(f"[WARN] fetch_commit_files sha={sha} exception: {e}")
return None
if resp.status_code != 200:
print(f"[WARN] fetch_commit_files sha={sha} status={resp.status_code}")
return None
commit_detail = resp.json()
time.sleep(0.2)
return commit_detail
def atomic_save_json(path: str, data: Any) -> None:
"""임시 파일에 저장 후 rename 으로 교체 (중간에 죽어도 파일 깨지는 것 방지)."""
tmp_path = path + ".tmp"
with open(tmp_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False)
os.replace(tmp_path, path)
def crawl_issues_with_diffs(
owner: str,
repo: str,
token: str,
output_path: str,
existing_results: Optional[List[Dict[str, Any]]] = None,
) -> List[Dict[str, Any]]:
"""
- 기존에 저장된 결과가 있으면 이어서 크롤링.
- 각 issue 처리 후마다 output_path 에 바로 저장.
"""
session = make_session(token)
all_issues = fetch_issues(session, owner, repo)
# 이미 저장된 결과가 있으면 이어서
if existing_results is None:
results: List[Dict[str, Any]] = []
else:
results = list(existing_results)
processed_ids: Set[int] = {item["id"] for item in results}
# Filter issues that need processing
issues_to_process = [i for i in all_issues if i["id"] not in processed_ids]
pbar = tqdm(
issues_to_process,
desc=f"Crawling {owner}/{repo}",
unit="issue",
initial=len(processed_ids),
total=len(all_issues),
)
for issue in pbar:
issue_id = issue["id"]
issue_url = issue["url"]
title = issue.get("title", "")
body = issue.get("body", "")
pbar.set_postfix({"issue": issue["number"]})
pr_numbers = collect_pr_numbers_for_issue(session, owner, repo, issue)
commits_entries: List[Dict[str, Any]] = []
for pr_num in pr_numbers:
pr_commits = fetch_pr_commits(session, owner, repo, pr_num)
for c in pr_commits:
sha = c["sha"]
commit_detail = fetch_commit_files(session, owner, repo, sha)
if not commit_detail:
continue
commit_date = (
commit_detail
.get("commit", {})
.get("author", {})
.get("date")
)
for f in commit_detail.get("files", []):
patch = f.get("patch")
if not patch:
continue
filename = f.get("filename")
commits_entries.append(
{
"filename": filename,
"patch": patch,
"date": commit_date,
}
)
issue_record = {
"id": issue_id,
"url": issue_url,
"title": title,
"body": body,
"commits": commits_entries,
}
results.append(issue_record)
processed_ids.add(issue_id)
atomic_save_json(output_path, results)
pbar.close()
return results
def main():
if len(sys.argv) != 4:
print(
"Usage: python crawl_issues_with_diffs.py <owner> <repo> <output_json>\n"
"Example:\n"
' python crawl_issues_with_diffs.py pallets flask flask_issues.json'
)
sys.exit(1)
owner = sys.argv[1]
repo = sys.argv[2]
output_path = sys.argv[3]
token = os.environ.get("GITHUB_TOKEN")
if not token:
print("ERROR: GITHUB_TOKEN 환경변수를 설정해 주세요.")
sys.exit(1)
# 🔹 기존 결과가 있으면 먼저 로드해서 이어서 크롤링
existing_results: Optional[List[Dict[str, Any]]] = None
if os.path.exists(output_path):
try:
with open(output_path, "r", encoding="utf-8") as f:
existing_results = json.load(f)
print(f"[INFO] loaded existing results from {output_path}")
except Exception as e:
print(f"[WARN] failed to load existing {output_path}: {e}")
existing_results = None
results = crawl_issues_with_diffs(owner, repo, token, output_path, existing_results)
# 마지막으로 한 번 더 저장 (안전)
atomic_save_json(output_path, results)
print(f"[DONE] saved {len(results)} issues to {output_path}")
if __name__ == "__main__":
main()
+17
View File
@@ -0,0 +1,17 @@
services:
bugfs:
build:
context: .
dockerfile: Dockerfile
container_name: bugfs
restart: unless-stopped
ports:
- "1026:1026"
volumes:
- ./data:/app/data
- ./logs:/app/logs
environment:
- GITHUB_TOKEN=${GITHUB_TOKEN}
- OPENAI_API_KEY=${OPENAI_API_KEY}
- BUGFS_DATA_DIR=/app/data
- BUGFS_LOG_DIR=/app/logs
BIN
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 145 KiB

Executable
+158
View File
@@ -0,0 +1,158 @@
import json
from openai import OpenAI
from dotenv import load_dotenv
import os
import time
import numpy as np
load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")
INPUT_JSON = "spinnaker_issues_.json"
OUTPUT_JSON = "spinnaker_embeddings_.json"
MODEL = "text-embedding-3-small"
CHUNK_CHAR_LIMIT = 3000 # 필요하면 조절
def load_json(path):
if not os.path.exists(path):
return {}
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
print(f"[WARN] failed to load existing {path}: {e}")
return {}
def atomic_save(path, data):
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False)
os.replace(tmp, path)
def load_issues(path):
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
def build_issue_text(issue):
"""title + body 묶어서 임베딩 입력 텍스트 생성"""
title = issue.get("title", "")
body = issue.get("body", "")
return f"{title}\n\n{body}".strip()
def split_into_chunks(text: str, limit: int = CHUNK_CHAR_LIMIT):
"""긴 텍스트를 limit 길이만큼 청크로 나눔."""
return [text[i:i + limit] for i in range(0, len(text), limit)]
def embed_text_chunks(client, chunks):
"""여러 chunk를 embedding 후 평균 벡터 반환"""
res = client.embeddings.create(
model=MODEL,
input=chunks
)
vectors = [np.array(item.embedding) for item in res.data]
avg_vec = np.mean(vectors, axis=0)
return avg_vec.tolist()
def format_seconds(sec: float) -> str:
"""초를 'HH:MM:SS' 문자열로 포맷"""
sec = int(sec)
h = sec // 3600
m = (sec % 3600) // 60
s = sec % 60
return f"{h:02d}:{m:02d}:{s:02d}"
def main():
client = OpenAI()
issues = load_issues(INPUT_JSON)
# 🔥 기존 결과 로드 후 이어서 진행
embeddings = load_json(OUTPUT_JSON)
total_issues = len(issues)
already_embedded = len(embeddings)
print(f"[INFO] 전체 이슈 수: {total_issues}")
print(f"[INFO] 이미 임베딩된 이슈 수: {already_embedded}")
print(f"[INFO] 남은 이슈 수: {total_issues - already_embedded}")
start_time = time.time()
newly_processed = 0 # 이번 실행에서 새로 임베딩한 개수
for idx, issue in enumerate(issues, start=1):
issue_id = str(issue["id"])
# 🔥 이미 처리한 issue는 스킵
if issue_id in embeddings:
# 진행도 표시용: 스킵도 '이미 완료된 것'으로 취급
progress_count = already_embedded + newly_processed
progress_percent = (progress_count / total_issues) * 100 if total_issues > 0 else 0.0
print(f"[SKIP] issue {issue_id} already embedded "
f"({progress_count}/{total_issues}, {progress_percent:.2f}%)")
continue
text = build_issue_text(issue)
if not text.strip():
print(f"[WARN] issue {issue_id} has empty text. skipping")
continue
progress_count = already_embedded + newly_processed
progress_percent = (progress_count / total_issues) * 100 if total_issues > 0 else 0.0
print(f"[INFO] embedding issue {issue_id}, index={idx}/{total_issues}, "
f"진행도={progress_percent:.2f}%, text length={len(text)}")
chunks = split_into_chunks(text)
print(f"[INFO] → split into {len(chunks)} chunks")
# 🔥 API 재시도 로직
for attempt in range(5):
try:
t0 = time.time()
vec = embed_text_chunks(client, chunks)
t1 = time.time()
elapsed_for_issue = t1 - t0
break
except Exception as e:
print(f"[ERROR] embedding issue {issue_id}, attempt {attempt+1}/5: {e}")
time.sleep(2)
else:
print(f"[FAIL] issue {issue_id} skipped after 5 retries")
continue
embeddings[issue_id] = vec
newly_processed += 1
# ✅ ETA 계산
total_elapsed = time.time() - start_time
if newly_processed > 0:
avg_per_issue = total_elapsed / newly_processed
remaining_issues = total_issues - (already_embedded + newly_processed)
eta_seconds = max(0, remaining_issues * avg_per_issue)
eta_str = format_seconds(eta_seconds)
else:
eta_str = "N/A"
print(f"[OK] Embedded issue {issue_id}, vector length={len(vec)}, "
f"소요시간(방금): {elapsed_for_issue:.2f}s, 예상 남은 시간: {eta_str}")
# 🔥 즉시 저장
atomic_save(OUTPUT_JSON, embeddings)
print(f"[SAVE] 현재까지 {already_embedded + newly_processed}/{total_issues}개 저장 완료 → {OUTPUT_JSON}")
print(f"[DONE] total embedded issues: {len(embeddings)}")
atomic_save(OUTPUT_JSON, embeddings)
print("[FINAL SAVE] completed")
if __name__ == "__main__":
main()
+185
View File
@@ -0,0 +1,185 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": 1,
"metadata": {},
"outputs": [],
"source": [
"from openai import OpenAI\n",
"import json\n",
"import re\n",
"import time\n",
"import tiktoken\n",
"import shelve\n",
"from tqdm.auto import tqdm"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"client = OpenAI(\n",
" api_key='', # this is also the default, it can be omitted\n",
")"
]
},
{
"cell_type": "code",
"execution_count": 3,
"metadata": {},
"outputs": [],
"source": [
"project = \"maven\""
]
},
{
"cell_type": "code",
"execution_count": 4,
"metadata": {},
"outputs": [],
"source": [
"# functions = json.load(open(\"p_functions_with_times.json\",\"r\",encoding=\"utf-8\"))\n",
"issues = json.load(open(f\"{project}_issues_new.json\",\"r\"))"
]
},
{
"cell_type": "code",
"execution_count": 5,
"metadata": {},
"outputs": [],
"source": [
"tokenizer = tiktoken.encoding_for_model(\"text-embedding-3-small\")\n",
"def estimate_tokens_tiktoken(text):\n",
" tokens = tokenizer.encode(text)\n",
" return len(tokens)\n",
"\n",
"def get_text_embedding(text):\n",
" return client.embeddings.create(input = [text], model=\"text-embedding-3-small\").data[0].embedding\n",
"\n",
"def split_text(text, max_tokens=8192):\n",
" words = text.split()\n",
" chunks = []\n",
" current_chunk = []\n",
" current_tokens = 0\n",
"\n",
" for word in words:\n",
" current_tokens += estimate_tokens_tiktoken(word + ' ') # 단어와 공백 포함하여 토큰 계산\n",
" if current_tokens > max_tokens:\n",
" chunks.append(\" \".join(current_chunk))\n",
" current_chunk = []\n",
" current_tokens = estimate_tokens_tiktoken(word + ' ') # 새로운 청크의 첫 단어 토큰 수로 초기화\n",
" current_chunk.append(word)\n",
"\n",
" if current_chunk:\n",
" chunks.append(\" \".join(current_chunk))\n",
"\n",
" return chunks\n",
"\n",
"\n",
"def get_long_text_embedding(text, max_tokens=8192, delay_per_token=0.02):\n",
" chunks = split_text(text, max_tokens)\n",
" # embeddings = [(len(chunk),get_text_embedding(chunk)) for chunk in chunks]\n",
" embeddings = []\n",
" for chunk in chunks:\n",
" tokens = estimate_tokens_tiktoken(chunk)\n",
" embeddings.append((len(chunk),get_text_embedding(chunk)))\n",
" time.sleep(tokens * delay_per_token)\n",
" return embeddings"
]
},
{
"cell_type": "code",
"execution_count": 6,
"metadata": {},
"outputs": [],
"source": [
"# new_issues = []\n",
"# for issue in tqdm(issues):\n",
"# # print(issue.keys())\n",
"# if issue[\"title\"] and issue[\"body\"]:\n",
"# title = re.sub(r\"\\s+\",\" \",issue[\"title\"])\n",
"# description = re.sub(r\"\\s+\",\" \",issue[\"body\"])\n",
"# embeddings = get_long_text_embedding(\" \".join([title,description]))\n",
"# # print(embeddings)\n",
"# issue[\"embeddings\"] = embeddings\n",
"# new_issues.append(issue)\n",
"# # break\n",
"\n",
"# json.dump(new_issues,open(f\"{project}_issues_with_text-emberdding-3-small.json\",\"w\"))"
]
},
{
"cell_type": "code",
"execution_count": 8,
"metadata": {},
"outputs": [
{
"data": {
"application/vnd.jupyter.widget-view+json": {
"model_id": "af7c03b537a9445f88a613bf355e1279",
"version_major": 2,
"version_minor": 0
},
"text/plain": [
" 0%| | 0/283 [00:00<?, ?it/s]"
]
},
"metadata": {},
"output_type": "display_data"
}
],
"source": [
"with shelve.open(f\"{project}_issues_shelve.db\") as db:\n",
" for issue in tqdm(issues):\n",
" # Issue 식별자 설정, 예: issue ID 사용\n",
" issue_id = str(issue[\"id\"]) \n",
"\n",
" # 이전에 이미 처리된 issue 건너뛰기\n",
" if issue_id in db:\n",
" continue\n",
"\n",
" # title과 body가 모두 있는 경우에만 처리\n",
" if issue.get(\"title\",None) and issue.get(\"body\",None):\n",
" title = re.sub(r\"\\s+\", \" \", issue[\"title\"])\n",
" description = re.sub(r\"\\s+\", \" \", issue[\"body\"])\n",
" \n",
" # 여기서 임베딩을 계산하는 함수는 구현 필요\n",
" embeddings = get_long_text_embedding(\" \".join([title, description]))\n",
"\n",
" # Issue에 embeddings 추가 및 shelve에 저장\n",
" issue[\"embeddings\"] = embeddings\n",
" db[issue_id] = issue # shelve에 저장\n",
"\n",
"# 결과를 파일로 저장\n",
"with open(f\"{project}_issues_with_text-embedding-3-small.json\", \"w\") as file:\n",
" # shelve 데이터베이스에서 모든 저장된 issue를 읽어 JSON 파일로 저장\n",
" with shelve.open(f\"{project}_issues_shelve.db\") as db:\n",
" json.dump(list(db.values()), file)"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.2"
}
},
"nbformat": 4,
"nbformat_minor": 2
}
+248
View File
@@ -0,0 +1,248 @@
#!/usr/bin/env python3
"""
Migration script: Convert existing JSON embedding files to FAISS format.
"""
import json
import pickle
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional
import faiss
import numpy as np
DATA_DIR = Path("data")
EMBEDDING_DIM = 1536
# Regex patterns from BUGFS_mcp_server.py
HUNK_HEADER_REGEX = re.compile(
r"@@ -\d+,\d+ \+(?P<start>\d+),(?P<length>\d+) @@(?:\s+(?P<context>.+))?$",
flags=re.MULTILINE,
)
PYTHON_FUNC_PATTERN = re.compile(r"def\s+(?P<name>\w+)")
JAVA_METHOD_PATTERN = re.compile(
r"(?:public|private|protected|static|final|abstract|synchronized|void|\w+)\s+(?P<name>\w+)\s*\("
)
CLASS_PATTERN = re.compile(r"(?:class|interface|enum)\s+(?P<name>\w+)")
@dataclass(frozen=True)
class IssueLocation:
start: int
length: int
function: str
file: str
@property
def label(self) -> str:
return f"{self.file}:{self.function}"
def normalize_filename(path_str: str) -> str:
path_str = path_str.replace("\\", "/")
parts = [p for p in path_str.split("/") if p]
if not parts:
return path_str
return parts[-1]
def extract_method_from_context(context: str) -> Optional[str]:
if not context:
return None
match = PYTHON_FUNC_PATTERN.search(context)
if match:
return match.group("name")
match = JAVA_METHOD_PATTERN.search(context)
if match:
name = match.group("name")
if name not in ("if", "for", "while", "switch", "catch", "synchronized", "return", "new", "throw"):
return name
match = CLASS_PATTERN.search(context)
if match:
return match.group("name")
return None
def extract_locations(issue: Dict[str, Any]) -> List[IssueLocation]:
locations: List[IssueLocation] = []
seen: set[tuple[str, str]] = set()
for commit in issue.get("commits", []):
filename = normalize_filename(commit.get("filename", "unknown"))
patch = commit.get("patch")
if not patch:
continue
for match in HUNK_HEADER_REGEX.finditer(patch):
start = int(match.group("start"))
length = int(match.group("length"))
context = match.group("context") or ""
method_name = extract_method_from_context(context)
if method_name:
key = (filename, method_name)
if key not in seen:
seen.add(key)
locations.append(
IssueLocation(start=start, length=length, function=method_name, file=filename)
)
return locations
def build_issue_text(issue: Dict[str, Any]) -> str:
title = (issue.get("title") or "").strip()
body = (issue.get("body") or "").strip()
return f"{title}\n\n{body}".strip()
def sanitize_issue_body(issue: Dict[str, Any], min_word_count: int = 1) -> Optional[str]:
text = build_issue_text(issue)
text = re.sub(r"\s+", " ", text)
if len(text.split()) < min_word_count:
return None
return text
def migrate_json_to_faiss(json_path: Path) -> None:
"""Convert a JSON embedding file to FAISS format."""
print(f"\n{'='*60}")
print(f"Migrating: {json_path.name}")
print(f"File size: {json_path.stat().st_size / (1024*1024):.1f} MB")
print(f"{'='*60}")
# Parse filename to get slug and model
# Format: {slug}_issues_with_{model}.json
name = json_path.stem # remove .json
if "_issues_with_" not in name:
print(f"⚠️ Skipping {json_path.name} - not an embedding file")
return
parts = name.split("_issues_with_")
slug = parts[0]
model = parts[1]
print(f"Slug: {slug}")
print(f"Model: {model}")
# Load JSON
print("Loading JSON...")
with json_path.open("r", encoding="utf-8") as f:
issues = json.load(f)
print(f"Loaded {len(issues)} issues")
# Build records and embeddings
print("Processing issues...")
records: List[Dict[str, Any]] = []
embeddings: List[np.ndarray] = []
embeddings_cache: Dict[str, Dict[str, Any]] = {}
skipped_no_embedding = 0
skipped_no_locations = 0
skipped_no_text = 0
for issue in issues:
embedding = issue.get("embedding")
if not embedding:
skipped_no_embedding += 1
continue
# Cache all embeddings (even without locations)
issue_id = str(issue["id"])
embeddings_cache[issue_id] = issue
locations = extract_locations(issue)
if not locations:
skipped_no_locations += 1
continue
text = sanitize_issue_body(issue)
if not text:
skipped_no_text += 1
continue
# Convert IssueLocation objects to dicts for pickle compatibility
locations_as_dicts = [
{"start": loc.start, "length": loc.length, "function": loc.function, "file": loc.file}
for loc in locations
]
records.append({
"id": issue["id"],
"locations": locations_as_dicts,
"content": text,
})
embeddings.append(np.asarray(embedding, dtype=np.float32))
print(f" - With embeddings: {len(embeddings_cache)}")
print(f" - Skipped (no embedding): {skipped_no_embedding}")
print(f" - Skipped (no locations): {skipped_no_locations}")
print(f" - Skipped (no text): {skipped_no_text}")
print(f" - Valid for index: {len(records)}")
if not records:
print("⚠️ No valid records to index!")
return
# Build FAISS index
print("Building FAISS index...")
matrix = np.vstack(embeddings).astype(np.float32)
faiss.normalize_L2(matrix)
index = faiss.IndexFlatIP(matrix.shape[1])
index.add(matrix)
print(f"Index created with {index.ntotal} vectors, dim={index.d}")
# Save files
faiss_path = DATA_DIR / f"{slug}_{model}.faiss"
metadata_path = DATA_DIR / f"{slug}_{model}_metadata.pkl"
embeddings_path = DATA_DIR / f"{slug}_{model}_embeddings.pkl"
print(f"Saving FAISS index to {faiss_path.name}...")
faiss.write_index(index, str(faiss_path))
print(f"Saving metadata to {metadata_path.name}...")
with metadata_path.open("wb") as f:
pickle.dump(records, f, protocol=pickle.HIGHEST_PROTOCOL)
print(f"Saving embeddings cache to {embeddings_path.name}...")
with embeddings_path.open("wb") as f:
pickle.dump(embeddings_cache, f, protocol=pickle.HIGHEST_PROTOCOL)
# Print size comparison
old_size = json_path.stat().st_size
new_size = faiss_path.stat().st_size + metadata_path.stat().st_size + embeddings_path.stat().st_size
print(f"\n📊 Size comparison:")
print(f" - Original JSON: {old_size / (1024*1024):.1f} MB")
print(f" - FAISS index: {faiss_path.stat().st_size / (1024*1024):.1f} MB")
print(f" - Metadata: {metadata_path.stat().st_size / (1024*1024):.1f} MB")
print(f" - Embeddings: {embeddings_path.stat().st_size / (1024*1024):.1f} MB")
print(f" - Total new: {new_size / (1024*1024):.1f} MB")
print(f" - Reduction: {(1 - new_size/old_size)*100:.1f}%")
print(f"\n✅ Migration complete for {slug}")
def main():
print("🚀 FAISS Migration Tool")
print(f"Data directory: {DATA_DIR.absolute()}")
# Find all embedding JSON files
json_files = list(DATA_DIR.glob("*_issues_with_*.json"))
if not json_files:
print("No embedding files found to migrate!")
return
print(f"\nFound {len(json_files)} embedding file(s) to migrate:")
for f in json_files:
print(f" - {f.name} ({f.stat().st_size / (1024*1024):.1f} MB)")
for json_path in json_files:
migrate_json_to_faiss(json_path)
print("\n" + "="*60)
print("🎉 All migrations complete!")
print("="*60)
if __name__ == "__main__":
main()
+8
View File
@@ -0,0 +1,8 @@
numpy>=1.24.0
python-dotenv>=1.0.0
openai>=1.0.0
fastmcp>=0.1.0
requests>=2.28.0
faiss-cpu>=1.7.4
tqdm>=4.65.0
sqlite-vec>=0.1.1