# MiniGit integrated into RP Assistant
import os
import json
import hashlib
import zlib
import difflib
import datetime
import subprocess
import sqlite3
import urllib.request
import urllib.error
from rp.config import DB_PATH
# Constants
IGNORE_SET = {
".minigit",
".git",
".hg",
".svn",
"__pycache__",
".DS_Store",
}
# AI constants for commit message generation
API_URL = "https://static.molodetz.nl/rp.cgi/api/v1/chat/completions"
MODEL = "google/gemma-3-12b-it:free"
TEMPERATURE = 1.0
MAX_TOKENS = None
IGNORE_SET = {
".minigit",
".git",
".hg",
".svn",
"__pycache__",
".DS_Store",
}
def init_storage():
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS minigit_objects (
hash TEXT PRIMARY KEY,
data BLOB
)
"""
)
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS minigit_commits (
hash TEXT PRIMARY KEY,
data TEXT
)
"""
)
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS minigit_trees (
hash TEXT PRIMARY KEY,
data TEXT
)
"""
)
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS minigit_meta (
key TEXT PRIMARY KEY,
value TEXT
)
"""
)
# Check if initial commit exists
cursor.execute("SELECT value FROM minigit_meta WHERE key = 'head'")
if not cursor.fetchone():
initial_commit_hash = "0" * 40
initial_commit_data = {
"parent_hash": None,
"message": "Initial commit",
"author": "System",
"timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"tree_hash": None,
"project_name": os.path.basename(os.getcwd()),
"changed_files": [],
}
save_commit(initial_commit_hash, initial_commit_data)
cursor.execute(
"INSERT INTO minigit_meta (key, value) VALUES ('head', ?)", (initial_commit_hash,)
)
conn.commit()
conn.close()
return True
def get_db_connection():
return sqlite3.connect(DB_PATH)
def save_object(table, obj_hash, data):
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute(f"INSERT OR REPLACE INTO {table} (hash, data) VALUES (?, ?)", (obj_hash, data))
conn.commit()
conn.close()
def get_object(table, obj_hash):
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute(f"SELECT data FROM {table} WHERE hash = ?", (obj_hash,))
row = cursor.fetchone()
conn.close()
if not row:
raise ValueError(f"Object not found: {obj_hash}")
return row[0]
def object_exists(table, obj_hash):
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute(f"SELECT 1 FROM {table} WHERE hash = ?", (obj_hash,))
exists = cursor.fetchone() is not None
conn.close()
return exists
def save_blob(content: bytes) -> str:
blob_hash = hash_content(content)
if not object_exists("minigit_objects", blob_hash):
compressed_content = zlib.compress(content)
save_object("minigit_objects", blob_hash, compressed_content)
return blob_hash
def get_blob_content(blob_hash: str) -> bytes:
compressed_content = get_object("minigit_objects", blob_hash)
return zlib.decompress(compressed_content)
def save_commit(commit_hash, commit_data):
save_object("minigit_commits", commit_hash, json.dumps(commit_data, indent=2))
def get_commit(commit_hash: str) -> dict:
return json.loads(get_object("minigit_commits", commit_hash))
def save_tree(tree_hash, tree_data):
save_object("minigit_trees", tree_hash, json.dumps(tree_data, indent=2))
def get_tree(tree_hash: str) -> dict:
if not tree_hash:
return {}
return json.loads(get_object("minigit_trees", tree_hash))
def get_tree_for_commit(commit_hash: str) -> dict:
commit = get_commit(commit_hash)
return get_tree(commit.get("tree_hash"))
def get_latest_commit_hash() -> str:
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT value FROM minigit_meta WHERE key = 'head'")
row = cursor.fetchone()
conn.close()
return row[0] if row else "0" * 40
def update_head(commit_hash: str):
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT OR REPLACE INTO minigit_meta (key, value) VALUES ('head', ?)", (commit_hash,)
)
conn.commit()
conn.close()
def get_all_commit_hashes():
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT hash FROM minigit_commits")
for row in cursor.fetchall():
yield row[0]
conn.close()
def get_full_commit_hash(partial_hash: str) -> str | None:
if not partial_hash:
return None
if len(partial_hash) == 40:
return partial_hash if object_exists(COMMITS_DIR, partial_hash) else None
matches = [h for h in get_all_commit_hashes() if h.startswith(partial_hash)]
if len(matches) == 0:
return None
if len(matches) > 1:
return None
return matches[0]
def safe_run(cmd, default=""):
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
return result.stdout.strip() if result.returncode == 0 else default
except Exception:
return default
def call_ai(prompt):
try:
api_key = os.environ.get("MOLODETZ_API_KEY", "retoorded")
if not api_key:
return None
data = {
"model": MODEL,
"messages": [{"role": "user", "content": prompt}],
"temperature": TEMPERATURE,
}
if MAX_TOKENS is not None:
data["max_tokens"] = MAX_TOKENS
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"HTTP-Referer": "https://github.com/commit-hook",
"X-Title": "Git Commit Hook",
}
req = urllib.request.Request(
API_URL, data=json.dumps(data).encode("utf-8"), headers=headers
)
with urllib.request.urlopen(req, timeout=60) as response:
result = json.loads(response.read().decode("utf-8"))
return result["choices"][0]["message"]["content"].strip()
except Exception:
return None
def get_diff_string():
parent_hash = get_latest_commit_hash()
if parent_hash == "0" * 40:
return ""
tree_a = get_tree_for_commit(parent_hash)
tree_b = {}
for file_path, content in scan_working_dir():
tree_b[file_path] = hash_content(content)
diff_lines = []
files_a = set(tree_a.keys())
files_b = set(tree_b.keys())
all_files = sorted(list(files_a.union(files_b)))
for file_path in all_files:
if file_path not in files_a:
try:
content_b = get_blob_content(tree_b[file_path]).decode("utf-8")
diff_lines.append(f"+++ Added file: {file_path}")
for line in content_b.splitlines():
diff_lines.append(f"+{line}")
except:
diff_lines.append(f"+++ Added file: {file_path} [binary]")
elif file_path not in files_b:
try:
content_a = get_blob_content(tree_a[file_path]).decode("utf-8")
diff_lines.append(f"--- Removed file: {file_path}")
for line in content_a.splitlines():
diff_lines.append(f"-{line}")
except:
diff_lines.append(f"--- Removed file: {file_path} [binary]")
else:
if tree_a[file_path] != tree_b[file_path]:
try:
content_a = get_blob_content(tree_a[file_path]).decode("utf-8").splitlines()
content_b = get_blob_content(tree_b[file_path]).decode("utf-8").splitlines()
diff = difflib.unified_diff(
content_a,
content_b,
fromfile=f"a/{file_path}",
tofile=f"b/{file_path}",
lineterm="",
)
diff_lines.append(f"~~~ Modified file: {file_path}")
diff_lines.extend(diff)
except:
diff_lines.append(f"~~~ Modified file: {file_path} [binary]")
return "\n".join(diff_lines)
def get_changed_files_list():
parent_hash = get_latest_commit_hash()
if parent_hash == "0" * 40:
return [file_path for file_path, _ in scan_working_dir()]
tree_a = get_tree_for_commit(parent_hash)
tree_b = {}
for file_path, content in scan_working_dir():
tree_b[file_path] = hash_content(content)
changed = []
files_a = set(tree_a.keys())
files_b = set(tree_b.keys())
all_files = files_a.union(files_b)
for file_path in all_files:
if (
file_path not in files_a
or file_path not in files_b
or tree_a.get(file_path) != tree_b.get(file_path)
):
changed.append(file_path)
return changed
def generate_commit_message(diff, files):
try:
files_list = "\n".join([f"- {f}" for f in files[:20]])
prompt = f"""You write commit messages for code changes.
Changed files:
{files_list}
Code changes:
{diff[:12000]}
Write a commit message with this format:
<prefix>: <description>
<prefix>: <description>
... (use multiple lines for multiple distinct changes)
Format rules:
- All lowercase for the prefix
- Colon and space after prefix
- Start description with lowercase letter
- No period at the end
- Use imperative mood (Add not Added)
- Each line should be a separate change
Choose prefixes from:
- fix: for bug fixes
Example: fix: resolve null pointer error in user login
Example: fix: correct date format in export function
- feat: for new features
Example: feat: add dark mode toggle to settings
Example: feat: implement search filter for products
- docs: for documentation changes
Example: docs: update api endpoint descriptions
Example: docs: add setup guide for development
- perf: for performance improvements
Example: perf: reduce database query time by 40%
Example: perf: optimize image loading with lazy load
- refactor: for code restructuring
Example: refactor: simplify user validation logic
Example: refactor: extract common functions to utils
- maintenance: for routine updates and maintenance
Example: maintenance: update dependencies to latest versions
Example: maintenance: clean up unused imports and files
Reply with ONLY the commit message, nothing else. No other text. No explanations or reasoning."""
message = call_ai(prompt)
if not message:
return generate_fallback_message(files)
message = message.strip().strip('"').strip("\"'")
lines = message.split("\n")
processed_lines = []
prefixes = ["fix:", "feat:", "docs:", "perf:", "refactor:", "maintenance:"]
for line in lines:
line = line.strip()
if line and not any(line.startswith(p) for p in prefixes):
line = f"feat: {line}"
if line:
processed_lines.append(line)
if not processed_lines:
return generate_fallback_message(files)
return "\n".join(processed_lines)
except Exception:
return generate_fallback_message(files)
def generate_fallback_message(files):
try:
if not files:
return "feat: update project"
exts = set()
for f in files:
ext = os.path.splitext(f)[1]
if ext:
exts.add(ext[1:])
if exts:
return f"feat: update {', '.join(sorted(exts)[:3])} files"
return "feat: update project files"
except Exception:
return "feat: update project"
def hash_content(content: bytes) -> str:
return hashlib.sha1(content).hexdigest()
def scan_working_dir():
for root, dirs, files in os.walk(".", topdown=True):
dirs[:] = [d for d in dirs if d not in IGNORE_SET and not d.startswith(".")]
for file in files:
if file in IGNORE_SET:
continue
file_path = os.path.normpath(os.path.join(root, file))
file_path = file_path.replace(os.path.sep, "/")
if file_path.startswith("./"):
file_path = file_path[2:]
try:
with open(file_path, "rb") as f:
content = f.read()
yield file_path, content
except IOError:
pass
def cmd_commit(message=None, author="rp"):
init_storage() # Ensure tables exist
parent_hash = get_latest_commit_hash()
current_tree = {}
files_scanned = 0
for file_path, content in scan_working_dir():
files_scanned += 1
blob_hash = save_blob(content)
current_tree[file_path] = blob_hash
if files_scanned == 0 and parent_hash != "0" * 40:
return
parent_tree = get_tree_for_commit(parent_hash)
if parent_tree == current_tree:
return
changed_files = get_changed_files_list()
if message is None:
diff = get_diff_string()
message = generate_commit_message(diff, changed_files)
tree_data_str = json.dumps(current_tree, sort_keys=True)
tree_hash = hash_content(tree_data_str.encode())
save_tree(tree_hash, current_tree)
commit_timestamp = datetime.datetime.now(datetime.timezone.utc)
project_name = os.path.basename(os.getcwd())
commit_data_to_hash = (
f"tree:{tree_hash}\n"
f"parent:{parent_hash}\n"
f"author:{author}\n"
f"timestamp:{commit_timestamp.isoformat()}\n"
f"message:{message}\n"
f"project:{project_name}\n"
)
commit_hash = hash_content(commit_data_to_hash.encode())
commit_data = {
"parent_hash": parent_hash,
"message": message,
"author": author,
"timestamp": commit_timestamp.isoformat(),
"tree_hash": tree_hash,
"project_name": project_name,
"changed_files": changed_files,
}
save_commit(commit_hash, commit_data)
update_head(commit_hash)
return commit_hash
def minigit_commit(message=None):
"""Commit current state to MiniGit."""
try:
init_storage() # Ensure repo exists
return cmd_commit(message)
except Exception:
return None
def checkout_commit(commit_hash):
"""Checkout to a specific commit."""
commit = get_commit(commit_hash)
tree = get_tree(commit["tree_hash"])
current_files = set(scan_working_dir_files())
tree_files = set(tree.keys())
# Remove files not in tree
for file_path in current_files - tree_files:
if os.path.exists(file_path):
os.remove(file_path)
# Add or update files in tree
for file_path, blob_hash in tree.items():
content = get_blob_content(blob_hash)
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, "wb") as f:
f.write(content)
update_head(commit_hash)
def scan_working_dir_files():
for root, dirs, files in os.walk(".", topdown=True):
dirs[:] = [d for d in dirs if d not in IGNORE_SET and not d.startswith(".")]
for file in files:
if file in IGNORE_SET:
continue
file_path = os.path.normpath(os.path.join(root, file))
file_path = file_path.replace(os.path.sep, "/")
if file_path.startswith("./"):
file_path = file_path[2:]
yield file_path
def minigit_revert(description):
"""Revert to the latest commit whose message contains the description."""
try:
init_storage()
matching_commits = []
for commit_hash in get_all_commit_hashes():
commit = get_commit(commit_hash)
if description.lower() in commit["message"].lower():
matching_commits.append((commit_hash, commit))
if not matching_commits:
return None
# Sort by timestamp, latest first
matching_commits.sort(key=lambda x: x[1]["timestamp"], reverse=True)
commit_hash, _ = matching_commits[0]
checkout_commit(commit_hash)
return commit_hash
except Exception:
return None
# For integration, this will be called before file modifications
def pre_commit():
"""Pre-commit hook for file modifications."""
return minigit_commit("maintenance: prepare for file modification")
Reference in New Issue View Git Blame Copy Permalink