GitSpec Initial Deployment

.gitignore

@@ -0,0 +1,9 @@
+.env
+__pycache__/
+*.pyc
+.DS_Store
+/tmp/
+venv/
+.venv/
+.jsonl
+_repo/

Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.10-slim
+
+RUN apt-get update && apt-get install -y git && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 8501
+
+CMD ["streamlit", "run", "streamlitUI.py", "--server.port=8501", "--server.address=0.0.0.0"]

README.md

@@ -0,0 +1 @@
+GitSpec - AI Powered Git Inspection Tool

backend/clone.py

@@ -0,0 +1,31 @@
+import os
+import shutil
+import stat
+import tempfile
+from git import Repo
+
+def readOnlyHandler(func, path, execinfo):
+    os.chmod(path, stat.S_IWRITE)
+    func(path)
+
+def cloneRepo(repository):
+    basePath = tempfile.gettempdir() 
+    folderName = repository.rstrip('/').split('/')[-1]
+    destinationPath = os.path.join(basePath, folderName)
+
+    if os.path.exists(destinationPath):
+        shutil.rmtree(destinationPath, onerror=readOnlyHandler)
+    
+    try:
+        repo = Repo.clone_from(repository, destinationPath)
+        return destinationPath, repo 
+    except Exception as e:
+        print("Error: ", e)
+        return None, None
+        
+def cleanupRepo(path):
+    try:
+        if os.path.exists(path):
+            shutil.rmtree(path, onerror=readOnlyHandler)
+    except Exception as e:
+        print(f"Cleanup warning: {e}")

backend/crawl.py

@@ -0,0 +1,11 @@
+import os 
+
+def repoCrawler(folderName):
+    scriptDir = os.path.dirname(os.path.abspath(__file__))
+    pythonFiles = []
+
+    for (root,dirs,files) in os.walk(os.path.join(scriptDir, folderName),topdown=True):
+        for file in files: 
+            if file.lower().endswith('.py'):
+                pythonFiles.append(os.path.join(root, file))
+    return pythonFiles

backend/extract.py

@@ -0,0 +1,164 @@
+import libcst as cst
+import shortuuid
+import json
+import textwrap
+import os 
+
+class CallVisitor(cst.CSTVisitor):
+    def __init__(self):
+        super().__init__()
+        self.calls = set()
+
+    def visit_Call(self, node: cst.Call):
+        if isinstance(node.func, cst.Name):
+            self.calls.add(node.func.value)
+        elif isinstance(node.func, cst.Attribute):
+            if isinstance(node.func.attr, cst.Name):
+                self.calls.add(node.func.attr.value)
+
+class MethodRemover(cst.CSTTransformer):
+    """Removes top-level methods from a class body to create the 'Shell'."""
+    def leave_FunctionDef(self, originalNode, updatedNode): 
+        return cst.RemoveFromParent()
+
+class GlobalRemover(cst.CSTTransformer):
+    """Isolates global variables by stripping functions, classes, and imports."""
+    def leave_FunctionDef(self, originalNode, updatedNode): return cst.RemoveFromParent()
+    def leave_ClassDef(self, originalNode, updatedNode): return cst.RemoveFromParent()
+    def leave_Import(self, originalNode, updatedNode): return cst.RemoveFromParent()
+    def leave_ImportFrom(self, originalNode, updatedNode): return cst.RemoveFromParent()
+    def leave_If(self, originalNode, updatedNode):
+        if isinstance(originalNode.test, cst.Comparison):
+            left = originalNode.test.left
+            if isinstance(left, cst.Name) and left.value == "__name__":
+                return cst.RemoveFromParent()
+        return updatedNode
+
+# Utility Functions
+def get_calls(node):
+    visitor = CallVisitor()
+    node.visit(visitor)
+    return sorted(list(visitor.calls))
+
+def get_module_names(module_node):
+    names = set()
+    for node in module_node.body:
+        if isinstance(node, cst.SimpleStatementLine):
+            for s in node.body:
+                if isinstance(s, (cst.Import, cst.ImportFrom)):
+                    if isinstance(s, cst.Import):
+                        for alias in s.names:
+                            fullName = module_node.code_for_node(alias.name)
+                            names.add(fullName.split('.')[0])
+                    elif isinstance(s, cst.ImportFrom):
+                        if s.module:
+                            fullName = module_node.code_for_node(s.module)
+                            names.add(fullName.split('.')[0])
+    return sorted(list(names))
+
+# Main Processor
+def processPythonFile(files, folderName, repo, outputFile="code.jsonl"):
+    results = []
+
+    for file in files:
+        relPath = os.path.relpath(file, repo.working_tree_dir)
+        
+        # Git Metadata
+        commits = list(repo.iter_commits(paths=relPath))
+        if commits:
+            lastCommit = commits[0]
+            firstCommit = commits[-1]
+            historyMetadata = {
+                "lastCommit": {
+                    "hash": lastCommit.hexsha,
+                    "msg": lastCommit.message.strip(),
+                    "author": lastCommit.author.name,
+                    "date": str(lastCommit.authored_datetime)
+                },
+                "firstCommit": {
+                    "hash": firstCommit.hexsha,
+                    "msg": firstCommit.message.strip(),
+                    "author": firstCommit.author.name,
+                    "date": str(firstCommit.authored_datetime)
+                }
+            }
+        else:
+            historyMetadata = {"lastCommit": None, "firstCommit": None}
+
+        try:
+            with open(file, 'r', encoding='utf-8') as f:
+                sourceCode = f.read()
+            module = cst.parse_module(sourceCode)
+        except Exception as e:
+            print(f"Could not read/parse file {file}: {e}")
+            continue
+
+        external_deps = get_module_names(module)
+        globalVarsCode = module.visit(GlobalRemover()).code.strip()
+        filepath = folderName + file.split(folderName)[-1]
+
+        importLines = []
+        for node in module.body:
+            if isinstance(node, cst.SimpleStatementLine):
+                if any(isinstance(s, (cst.Import, cst.ImportFrom)) for s in node.body):
+                    importLines.append(module.code_for_node(node).strip())
+        importedModulesCode = "\n".join(importLines)
+
+        for node in module.body:
+            # CASE 1: Top-Level Functions
+            if isinstance(node, cst.FunctionDef):
+                calls = get_calls(node)
+                header = (
+                    f"# FILE: {filepath}\n"
+                    f"# TYPE: Global Function\n"
+                    f"# CALLS: {', '.join(calls)}\n"
+                    f"# DEPS: {', '.join(external_deps)}\n"
+                    f"# LAST COMMIT: {historyMetadata['lastCommit']['msg'] if historyMetadata['lastCommit'] else 'N/A'}"
+                )
+                
+                results.append({
+                    'id': shortuuid.uuid(),
+                    'name': node.name.value,
+                    'filePath': filepath,
+                    'calls': calls,
+                    'external_deps': external_deps,
+                    'history': historyMetadata,
+                    'modules': importedModulesCode,
+                    'globalVariables': globalVarsCode,
+                    'parentClass': 'no parent',
+                    'isAsync': node.asynchronous is not None,
+                    'content': header + "\n\n" + module.code_for_node(node).strip(),
+                })
+
+            # CASE 2: Classes
+            elif isinstance(node, cst.ClassDef):
+                classShell = node.visit(MethodRemover())
+                classShellCode = module.code_for_node(classShell).strip()
+                
+                for item in node.body.body:
+                    if isinstance(item, cst.FunctionDef):
+                        calls = get_calls(item)
+                        header = (
+                            f"# FILE: {filepath}\n"
+                            f"# CLASS: {node.name.value}\n"
+                            f"# TYPE: Method\n"
+                            f"# CALLS: {', '.join(calls)}\n"
+                            f"# DEPS: {', '.join(external_deps)}\n"
+                            f"# LAST COMMIT: {historyMetadata['lastCommit']['msg'] if historyMetadata['lastCommit'] else 'N/A'}"
+                        )
+                        
+                        results.append({
+                            'id': shortuuid.uuid(),
+                            'name': item.name.value,
+                            'filePath': filepath,
+                            'calls': calls,
+                            'external_deps': external_deps,
+                            'history': historyMetadata,
+                            'modules': importedModulesCode,
+                            'globalVariables': globalVarsCode,
+                            'parentClass': classShellCode,
+                            'isAsync': item.asynchronous is not None,
+                            'content': header + "\n\n" + textwrap.dedent(module.code_for_node(item)).strip(),
+                        })
+            
+    return results

backend/historyExtractor.py

@@ -0,0 +1,92 @@
+import json
+from git import Repo, GitCommandError, NULL_TREE
+import os
+
+BOT_AUTHORS = ['github-actions[bot]', 'dependabot[bot]', 'pre-commit-ci[bot]', 'github-actions']
+CODE_EXTENSIONS = ['.py']
+MAX_FILES_PER_COMMIT = 20
+MAX_COMMITS_TO_SCAN = 200
+
+def splitDiffs(raw_diff):
+    if not raw_diff: return [], []
+    lines = raw_diff.split('\n')
+    added, removed = [], []
+    for line in lines:
+        if line.startswith('+++') or line.startswith('---') or line.startswith('\\'):
+            continue
+        if line.startswith('+'): added.append(line[1:])
+        elif line.startswith('-'): removed.append(line[1:])
+    return added, removed
+
+def isCodeCommit(files_info):
+    return any(f['file'].endswith(ext) for ext in CODE_EXTENSIONS for f in files_info)
+
+def extractHistory(repo, destPath, outputFile="commits_history.jsonl"):
+    try:
+        commits = list(repo.iter_commits(max_count=MAX_COMMITS_TO_SCAN))
+    except (GitCommandError, ValueError):
+        commits = []
+
+    indexedCount = 0
+    with open(outputFile, 'w', encoding='utf-8') as f:
+        for commit in commits:
+            if commit.author.name in BOT_AUTHORS:
+                continue
+
+            parent = None
+            if commit.parents:
+                try:
+                    p = commit.parents[0]
+                    repo.git.cat_file('-e', p.hexsha)
+                    parent = p
+                except GitCommandError:
+                    parent = None
+
+            if parent:
+                diffs = parent.diff(commit, create_patch=True)
+            else:
+                diffs = commit.diff(NULL_TREE, create_patch=True, reverse=True)
+
+            files_info = []
+            for d in diffs:
+                filePath = d.b_path if d.b_path else d.a_path
+                rawPatch = d.diff.decode('utf-8', errors='replace') if d.diff else ""
+                addedLines, removedLines = splitDiffs(rawPatch)
+
+                status = d.change_type
+                if not status:
+                    if addedLines and not removedLines: status = 'A'
+                    elif removedLines and not addedLines: status = 'D'
+                    else: status = 'M'
+
+                files_info.append({
+                    'file': filePath,
+                    'status': status,
+                    'additions': addedLines[:20],
+                    'removals': removedLines[:20]
+                })
+
+            if len(files_info) > MAX_FILES_PER_COMMIT:
+                continue
+            if not isCodeCommit(files_info):
+                continue
+
+            filesTouched = ", ".join(fi['file'] for fi in files_info)
+            embedString = (f"{commit.summary} — files: {filesTouched} — "
+                           f"author: {commit.author.name} — "
+                           f"date: {commit.authored_datetime.date()}")
+
+            commit_info = {
+                'sha': commit.hexsha[:7],
+                'author': commit.author.name,
+                'summary': commit.summary,
+                'date': commit.authored_datetime.isoformat(),
+                'changes': files_info,
+                'embedText': embedString
+            }
+
+            f.write(json.dumps(commit_info) + '\n')
+            indexedCount += 1
+
+    print(f"Extracted {indexedCount} commits into {outputFile}")
+    return outputFile  

backend/historyIndexer.py

@@ -0,0 +1,42 @@
+import json
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, VectorParams, PointStruct
+from sentence_transformers import SentenceTransformer
+
+def indexHistory(commitsFile, client, model):
+    """
+    Reads a JSONL file of filtered commits and indexes them into Qdrant.
+    """
+    collectionName = "historyIndex"
+
+    if client.collection_exists(collectionName):
+        client.delete_collection(collectionName)
+        
+    client.create_collection(
+        collection_name=collectionName,
+        vectors_config=VectorParams(size=384, distance=Distance.COSINE),
+    )
+
+    points = []
+    
+    with open(commitsFile, 'r', encoding='utf-8') as f:
+        for i, line in enumerate(f):
+            commit = json.loads(line)
+            
+            vector = model.encode(commit['embedText']).tolist()
+            
+            points.append(
+                PointStruct(
+                    id=i, 
+                    vector=vector, 
+                    payload=commit
+                )
+            )
+
+    if points:
+        client.upsert(collection_name=collectionName, points=points)
+        print(f"Successfully indexed {len(points)} commits into {collectionName}")
+    else:
+        print("No commits found in file to index.")
+
+    return client, model 

backend/indexer.py

@@ -0,0 +1,30 @@
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, VectorParams, PointStruct
+from sentence_transformers import SentenceTransformer
+
+def indexer(results):
+    # Qdrant IN RAM
+    client = QdrantClient(":memory:")
+
+    # Code-Strong model
+    model = SentenceTransformer('all-MiniLM-L6-v2')
+
+    if client.collection_exists("tempCollection"):
+        client.delete_collection("tempCollection")
+        
+    client.create_collection(
+        collection_name="tempCollection",
+        vectors_config=VectorParams(size=384, distance=Distance.COSINE),
+    )
+    points = [] 
+    for i, entry in enumerate(results):
+        vector = model.encode(entry['content']).tolist()
+        points.append(
+            PointStruct(
+                id=i, 
+                vector=vector, 
+                payload=entry
+            )
+        )
+    client.upsert(collection_name="tempCollection", points=points)
+    return client, model

backend/router.py

@@ -0,0 +1,55 @@
+from groq import Groq
+import os
+import json
+from dotenv import load_dotenv
+
+load_dotenv()
+
+groqClient = Groq(api_key=os.getenv("GROQ_API_KEY"))
+
+def routeQuery(query: str) -> str:
+    """
+    Classifies a user query into one of three routes:
+    CODE     - search the code index (current structure, how things work)
+    HISTORY  - search the history index (why things changed, who changed what)
+    BOTH     - search both indexes and merge results
+    """
+
+    prompt = f"""You are a query router for a code intelligence tool called GitSpec.
+GitSpec has two knowledge bases:
+
+1. CODE INDEX — contains the current source code: functions, classes, methods, imports.
+   Use this for questions about how something works RIGHT NOW.
+   Examples: "how does authentication work?", "what does process_order do?", "what classes exist in the payment module?"
+
+2. HISTORY INDEX — contains git commit history: what changed, when, who changed it, and why.
+   Use this for questions about change over time, blame, or reasoning.
+   Examples: "why did the auth module change?", "who introduced rate limiting?", "what changed last month?"
+
+3. BOTH — use when the question needs current code understanding AND historical context.
+   Examples: "how does auth work and why was it redesigned?", "what does checkout do and when was it last changed?"
+
+Classify this query into exactly one of: CODE, HISTORY, BOTH
+
+Query: {query}
+
+Respond with valid JSON only. No explanation. No markdown.
+Format: {{"route": "CODE"}} or {{"route": "HISTORY"}} or {{"route": "BOTH"}}"""
+
+    response = groqClient.chat.completions.create(
+        model="llama-3.3-70b-versatile",
+        messages=[{"role": "user", "content": prompt}],
+        temperature=0.0,  
+        max_tokens=20,    
+    )
+
+    raw = response.choices[0].message.content.strip()
+
+    try:
+        result = json.loads(raw)
+        route = result.get("route", "BOTH").upper()
+        if route not in ["CODE", "HISTORY", "BOTH"]:
+            route = "BOTH"  
+        return route
+    except json.JSONDecodeError:
+        return "BOTH"

requirements.txt

@@ -0,0 +1,6 @@
+streamlit
+qdrant-client
+sentence-transformers
+libcst
+shortuuid
+GitPython

streamlitUI.py

@@ -0,0 +1,208 @@
+import streamlit as st
+from backend.clone import cloneRepo, cleanupRepo
+from backend.crawl import repoCrawler
+from backend.extract import processPythonFile
+from backend.indexer import indexer
+from backend.historyExtractor import extractHistory
+from backend.historyIndexer import indexHistory
+from backend.router import routeQuery
+from groq import Groq
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+groqApiKey = os.getenv("GROQ_API_KEY")
+
+st.set_page_config(page_title="GitSpec", layout="wide")
+st.title("GitSpec")
+
+if "client" not in st.session_state:
+    st.session_state.client = None
+    st.session_state.model = None
+    st.session_state.repo_name = ""
+
+if "messages" not in st.session_state:
+    st.session_state.messages = []
+
+# Sidebar Controls
+with st.sidebar:
+    st.header("Settings")
+    if st.button("Clear Chat History"):
+        st.session_state.messages = []
+        st.rerun()
+    st.divider()
+    contextLimit = st.slider("Context Depth (Snippets)", 3, 10, 6)
+
+repoUrl = st.text_input("Enter GitHub Repository URL:")
+
+if st.button("Analyze Repository"):
+    if repoUrl:
+        st.session_state.repo_name = repoUrl.split('/')[-1].replace('.git', '')
+        progressBar = st.progress(0)
+        statusText = st.empty()
+
+        statusText.text("Cloning repository...")
+        progressBar.progress(10)
+        destPath, repo = cloneRepo(repoUrl)
+
+        if destPath and repo:
+            progressBar.progress(25)
+            statusText.text("Scanning files...")
+            files = repoCrawler(destPath)
+
+            progressBar.progress(40)
+            statusText.text("Extracting code structures...")
+            results = processPythonFile(files, destPath, repo)
+
+            progressBar.progress(55)
+            statusText.text("Indexing code...")
+            client, model = indexer(results)
+
+            progressBar.progress(70)
+            statusText.text("Extracting commit history...")
+            commitsFile = extractHistory(repo, destPath)
+
+            progressBar.progress(85)
+            statusText.text("Indexing history...")
+            client, model = indexHistory(commitsFile, client, model)
+
+            st.session_state.client = client
+            st.session_state.model = model
+            st.session_state.messages = []
+
+            progressBar.progress(100)
+            statusText.text("Cleaning up...")
+            cleanupRepo(destPath)
+
+            statusText.empty()
+            progressBar.empty()
+            st.success(f"'{st.session_state.repo_name}' Repository indexed successfully")
+        else:
+            statusText.empty()
+            progressBar.empty()
+            st.error("Failed to clone repository.")
+    else:
+        st.warning("Please enter a URL.")
+
+if st.session_state.client:
+    st.divider()
+
+    for message in st.session_state.messages:
+        with st.chat_message(message["role"]):
+            st.markdown(message["content"])
+
+    if query := st.chat_input("Ask anything about the codebase or its history..."):
+
+        with st.chat_message("user"):
+            st.markdown(query)
+        st.session_state.messages.append({"role": "user", "content": query})
+
+        groqClient = Groq(api_key=groqApiKey)
+
+        with st.spinner("Thinking..."):
+
+            # Route the query
+            route = routeQuery(query)
+
+            queryVector = st.session_state.model.encode(query).tolist()
+            contextBlocks = []
+            searchResults = []
+
+            # Search the right index(es)
+            if route in ["CODE", "BOTH"]:
+                codeResponse = st.session_state.client.query_points(
+                    collection_name="tempCollection",
+                    query=queryVector,
+                    limit=contextLimit
+                )
+                for res in codeResponse.points:
+                    p = res.payload
+                    block = (
+                        f"[CODE] FILE: {p.get('filePath')}\n"
+                        f"METADATA: Parent={p.get('parentClass')}, Imports={p.get('modules')}\n"
+                        f"LAST COMMIT: {p.get('history', {}).get('lastCommit', {}).get('msg')}\n"
+                        f"CODE:\n{p.get('content')}"
+                    )
+                    contextBlocks.append(block)
+                    searchResults.append(("code", res))
+
+            if route in ["HISTORY", "BOTH"]:
+                historyResponse = st.session_state.client.query_points(
+                    collection_name="historyIndex",
+                    query=queryVector,
+                    limit=contextLimit
+                )
+                for res in historyResponse.points:
+                    p = res.payload
+                    block = (
+                        f"[HISTORY] COMMIT: {p.get('sha')} by {p.get('author')} on {p.get('date', '')[:10]}\n"
+                        f"SUMMARY: {p.get('summary')}\n"
+                        f"FILES: {', '.join(f['file'] for f in p.get('changes', []))}\n"
+                        f"CHANGES: {p.get('embedText')}"
+                    )
+                    contextBlocks.append(block)
+                    searchResults.append(("history", res))
+
+            if not contextBlocks:
+                fullResponse = "I couldn't find relevant information. Try rephrasing your question."
+            else:
+                formattedContext = "\n---\n".join(contextBlocks)
+
+                systemPrompts = (
+                    "You are a Senior Software Architect and code historian for GitSpec. "
+                    "You have access to two knowledge sources: current source code (CODE) and git commit history (HISTORY). "
+                    "Use whichever is relevant to answer the question accurately."
+                    "\n\nSTRICT FORMATTING RULES:"
+                    "\n- Start immediately with the answer."
+                    "\n- Use H3 headers (###) for distinct sections."
+                    "\n- For history answers, always cite the commit SHA and author."
+                    "\n- For code answers, reference the file and function name."
+                    "\n- Do NOT include absolute local file paths."
+                    "\n- Keep the tone professional and concise."
+                    "\n- Be concise. No 'In conclusion' or summary sections — end when the answer is complete."
+
+                )
+
+                llm_messages = [
+                    {"role": "system", "content": systemPrompts},
+                    *st.session_state.messages[-5:],
+                    {"role": "user", "content": f"Context:\n{formattedContext}\n\nQuestion: {query}"}
+                ]
+                try:
+                    llmResponse = groqClient.chat.completions.create(
+                        model="llama-3.3-70b-versatile",
+                        messages=llm_messages,
+                        temperature=0.1
+                    )
+                    fullResponse = llmResponse.choices[0].message.content
+                except Exception as e:
+                    if "rate_limit_exceeded" in str(e).lower() or "413" in str(e):
+                        fullResponse = (
+                            "**Rate Limit Reached:** The context for this repository is quite large for the free tier. "
+                            "I've tried to answer, but Groq is busy. Please try: \n"
+                            "1. Reducing the **Context Depth** slider in the sidebar.\n"
+                            "2. Asking a more specific question.\n"
+                            "3. Waiting 60 seconds and trying again."
+                        )
+                    else:
+                        fullResponse = f"An unexpected error occurred: {str(e)}"
+                        
+
+        # Display answer
+        with st.chat_message("assistant"):
+            st.markdown(fullResponse)
+            st.caption(f"Route: `{route}`") 
+
+            if searchResults:
+                with st.expander("Explore Reference Sources"):
+                    for sourceType, res in searchResults:
+                        p = res.payload
+                        if sourceType == "code":
+                            st.caption(f"[CODE] {p.get('filePath')}")
+                            st.code(p.get('content'), language='python')
+                        else:
+                            st.caption(f"[HISTORY] {p.get('sha')} — {p.get('summary')}")
+                            for change in p.get('changes', []):
+                                st.caption(f"File: {change['file']} ({change['status']})")
+
+        st.session_state.messages.append({"role": "assistant", "content": fullResponse})