Create app.py

app.py

@@ -0,0 +1,248 @@
+import os, tempfile, subprocess, shutil, fnmatch, json, requests, re, pickle, time, threading
+import gradio as gr
+from dotenv import load_dotenv
+
+# ---------------------------------
+# Setup
+# ---------------------------------
+load_dotenv()
+OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
+MEMORY_FILE = "repogpt_memory.pkl"
+MEMORY_TTL = 600  # 10 minutes (600 seconds)
+_last_active_time = time.time()
+
+# ---------------------------------
+# Persistent Memory Management
+# ---------------------------------
+def save_memory(data):
+    data["_timestamp"] = time.time()
+    with open(MEMORY_FILE, "wb") as f:
+        pickle.dump(data, f)
+
+def load_memory():
+    if os.path.exists(MEMORY_FILE):
+        try:
+            with open(MEMORY_FILE, "rb") as f:
+                data = pickle.load(f)
+            # Expire memory if inactive > TTL
+            if time.time() - data.get("_timestamp", 0) > MEMORY_TTL:
+                print("🧹 Memory expired (10 min inactive). Resetting RepoGPT state.")
+                os.remove(MEMORY_FILE)
+                return {"chat_history": [], "repo_context": ""}
+            return data
+        except Exception:
+            return {"chat_history": [], "repo_context": ""}
+    return {"chat_history": [], "repo_context": ""}
+
+def clear_memory_periodically():
+    """Background thread to auto-clear memory after TTL."""
+    global _last_active_time
+    while True:
+        time.sleep(60)
+        if time.time() - _last_active_time > MEMORY_TTL:
+            if os.path.exists(MEMORY_FILE):
+                try:
+                    os.remove(MEMORY_FILE)
+                    print("🧹 Auto-cleared RepoGPT memory after 10 minutes of inactivity.")
+                except:
+                    pass
+            break
+
+MEMORY = load_memory()
+threading.Thread(target=clear_memory_periodically, daemon=True).start()
+
+# ---------------------------------
+# Repo Handling
+# ---------------------------------
+def clone_repo(repo_url):
+    tmp_dir = tempfile.mkdtemp()
+    subprocess.run(
+        ["git", "clone", "--depth", "1", repo_url, tmp_dir],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE
+    )
+    return tmp_dir
+
+def read_repo_files(repo_dir):
+    patterns = ["*.py", "*.js", "*.jsx", "*.ts", "*.tsx", "*.html", "*.json",
+                "*.go", "*.java", "*.vue", "*.md", "*.css"]
+    repo_content = []
+    for root, _, files in os.walk(repo_dir):
+        for pattern in patterns:
+            for filename in fnmatch.filter(files, pattern):
+                filepath = os.path.join(root, filename)
+                try:
+                    with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
+                        content = f.read()
+                        if len(content) > 4000:
+                            content = content[:4000]
+                        repo_content.append({"filename": filename, "content": content})
+                except:
+                    pass
+    return repo_content
+
+# ---------------------------------
+# Tech Stack Auto-Detection
+# ---------------------------------
+def detect_tech_stack(repo_files):
+    all_text = " ".join(f["content"] for f in repo_files).lower()
+    stack = set()
+    tech_keywords = {
+        "python": ["def ", "import ", "flask", "django", "fastapi"],
+        "javascript": ["function", "const ", "react", "vue", "next.js", "node"],
+        "typescript": ["typescript", "tsconfig", ".tsx"],
+        "java": ["public class", "springboot", "maven", "gradle"],
+        "go": ["package main", "func main", "go mod"],
+        "html": ["<html", "<body", "<div"],
+        "css": ["color:", "background", "font-family"],
+        "docker": ["from ", "dockerfile"],
+        "database": ["sql", "select", "insert", "mongodb", "mongoose", "prisma"],
+        "ai/ml": ["torch", "tensorflow", "sklearn", "transformers", "model"],
+    }
+    for tech, keywords in tech_keywords.items():
+        if any(k in all_text for k in keywords):
+            stack.add(tech)
+    if not stack:
+        return "No clear tech stack detected."
+    return "🧰 Detected Tech Stack: " + ", ".join(sorted(stack))
+
+# ---------------------------------
+# Code Search
+# ---------------------------------
+def search_code_snippets(repo_files, query, top_k=3):
+    query = query.lower()
+    matches = []
+    for file in repo_files:
+        lines = file["content"].splitlines()
+        for i, line in enumerate(lines):
+            if query in line.lower():
+                start = max(0, i - 2)
+                end = min(len(lines), i + 3)
+                snippet = "\n".join(lines[start:end])
+                matches.append((file["filename"], snippet))
+    matches = sorted(matches, key=lambda x: len(x[1]), reverse=True)[:top_k]
+    if not matches:
+        return ""
+    return "\n\n".join([f"📄 {m[0]}:\n{m[1]}" for m in matches])
+
+# ---------------------------------
+# OpenRouter API
+# ---------------------------------
+def call_openrouter(messages, model="google/gemma-3-27b-it:free"):
+    try:
+        res = requests.post(
+            "https://openrouter.ai/api/v1/chat/completions",
+            headers={
+                "Authorization": f"Bearer {OPENROUTER_API_KEY}",
+                "Content-Type": "application/json"
+            },
+            json={"model": model, "messages": messages},
+            timeout=180
+        )
+        res.raise_for_status()
+        return res.json()["choices"][0]["message"]["content"]
+    except Exception as e:
+        return f"⚠️ API Error: {e}"
+
+# ---------------------------------
+# Chat Logic
+# ---------------------------------
+def create_system_prompt(repo_context=None):
+    base = (
+        "You are RepoGPT — a helpful AI assistant for GitHub repositories. "
+        "You can explain code, summarize architecture, identify frameworks, search code, and generate ideas. "
+        "Be conversational and beginner-friendly."
+    )
+    if repo_context:
+        base += f"\n\nRepository context:\n{repo_context[:8000]}"
+    return {"role": "system", "content": base}
+
+def load_repository(repo_url):
+    global _last_active_time
+    _last_active_time = time.time()
+    if not repo_url or "github.com" not in repo_url:
+        return "⚠️ Please provide a valid GitHub repository URL.", "", ""
+    repo_dir = None
+    try:
+        repo_dir = clone_repo(repo_url)
+        repo_files = read_repo_files(repo_dir)
+        stack_info = detect_tech_stack(repo_files)
+        file_summary = "\n".join(f["filename"] for f in repo_files[:50])
+        context = f"{stack_info}\n\nFiles found:\n{file_summary}"
+        MEMORY["repo_context"] = "\n".join(f["content"] for f in repo_files)
+        save_memory(MEMORY)
+        return f"✅ Repo loaded successfully!\n\n{stack_info}", context, repo_files
+    except Exception as e:
+        return f"❌ Failed to load repo: {e}", "", []
+    finally:
+        if repo_dir and os.path.exists(repo_dir):
+            shutil.rmtree(repo_dir, ignore_errors=True)
+
+def chat_with_repo(message, history, repo_context, repo_files):
+    global _last_active_time
+    _last_active_time = time.time()
+
+    MEMORY["chat_history"] = history
+    save_memory(MEMORY)
+
+    code_snippet = ""
+    if any(k in message.lower() for k in ["function", "class", "api", "code", "def ", "explain"]):
+        code_snippet = search_code_snippets(repo_files, message)
+
+    messages = [create_system_prompt(repo_context)]
+    for user, bot in history[-6:]:
+        messages.append({"role": "user", "content": user})
+        messages.append({"role": "assistant", "content": bot})
+    if code_snippet:
+        message += f"\n\nRelevant Code Snippet:\n{code_snippet}"
+    messages.append({"role": "user", "content": message})
+
+    reply = call_openrouter(messages)
+    MEMORY["chat_history"].append((message, reply))
+    save_memory(MEMORY)
+    return reply
+
+# ---------------------------------
+# Gradio UI
+# ---------------------------------
+with gr.Blocks(theme=gr.themes.Soft(), title="RepoGPT") as demo:
+    gr.Markdown("""
+    # 🤖 RepoGPT  
+    **ChatGPT for GitHub Repositories**
+    
+    🧩 Features:
+    - 🔍 Smart Code Search  
+    - 💾 Persistent Memory (auto-clears after 10 min)  
+    - 🧠 Tech-Stack Detection
+    """)
+
+    with gr.Row():
+        repo_input = gr.Textbox(label="🔗 GitHub Repository URL", placeholder="https://github.com/yourname/yourproject")
+        load_btn = gr.Button("📂 Load Repo")
+
+    repo_status = gr.Markdown("ℹ️ No repository loaded yet.")
+    repo_context_state = gr.State(MEMORY.get("repo_context", ""))
+    repo_files_state = gr.State([])
+
+    def on_load_repo(link):
+        return load_repository(link)
+
+    load_btn.click(on_load_repo, inputs=[repo_input], outputs=[repo_status, repo_context_state, repo_files_state])
+
+    chatbot = gr.ChatInterface(
+        fn=lambda msg, history, repo_ctx, repo_files: chat_with_repo(msg, history, repo_ctx, repo_files),
+        additional_inputs=[repo_context_state, repo_files_state],
+        title="RepoGPT",
+        description="Your AI companion for understanding and improving GitHub projects.",
+        examples=[
+            ["Summarize this repo."],
+            ["Explain the backend structure."],
+            ["Find where login is implemented."],
+            ["What technologies are used here?"],
+            ["Generate a LinkedIn post for this project."]
+        ],
+    )
+
+if __name__ == "__main__":
+    port = int(os.environ.get("PORT", 7860))
+    demo.launch(server_name="0.0.0.0", server_port=port, share=False)