BobAi/Fss-Rag-Mini
7
0
Fork 0
Code Issues 2 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Implement comprehensive auto-update system

Browse Source 
✨ Features:
- GitHub releases integration with version checking
- TUI update notifications with user-friendly interface
- CLI update commands (check-update, update)
- Discrete notifications that don't interrupt workflow
- Legacy user detection for older versions
- Safe update process with backup and rollback
- Progress bars and user confirmation
- Configurable update preferences

🔧 Technical:
- UpdateChecker class with GitHub API integration
- UpdateConfig for user preferences
- Graceful fallbacks when network unavailable
- Auto-restart after successful updates
- Works with both TUI and CLI interfaces

🎯 User Experience:
- TUI: Shows update banner on startup if available
- CLI: Discrete one-line notice for regular commands
- Commands: 'rag-mini check-update' and 'rag-mini update'
- Non-intrusive design respects user workflow

This provides seamless updates for the critical improvements
we've been implementing while giving users full control.
This commit is contained in:
BobAi 2025-08-15 15:10:59 +10:00
parent 92cb600dd6
commit e7e0f71a35
5 changed files with 744 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
mini_rag/__init__.py
View File

@ -13,6 +13,20 @@ from .indexer import ProjectIndexer
from .search import CodeSearcher
from .watcher import FileWatcher
# Auto-update system (graceful import for legacy versions)
try:
from .updater import UpdateChecker, check_for_updates, get_updater
__all__ = [
"CodeEmbedder",
"CodeChunker",
"ProjectIndexer",
"CodeSearcher",
"FileWatcher",
"UpdateChecker",
"check_for_updates",
"get_updater",
]
except ImportError:
__all__ = [
"CodeEmbedder",
"CodeChunker",

25
mini_rag/config.py
View File

@ -114,6 +114,16 @@ class LLMConfig:
]
@dataclass
class UpdateConfig:
"""Configuration for auto-update system."""
auto_check: bool = True # Check for updates automatically
check_frequency_hours: int = 24 # How often to check (hours)
auto_install: bool = False # Auto-install without asking (not recommended)
backup_before_update: bool = True # Create backup before updating
notify_beta_releases: bool = False # Include beta/pre-releases
@dataclass
class RAGConfig:
"""Main RAG system configuration."""
@ -123,6 +133,7 @@ class RAGConfig:
embedding: EmbeddingConfig = None
search: SearchConfig = None
llm: LLMConfig = None
updates: UpdateConfig = None
def __post_init__(self):
if self.chunking is None:
@ -137,6 +148,8 @@ class RAGConfig:
self.search = SearchConfig()
if self.llm is None:
self.llm = LLMConfig()
if self.updates is None:
self.updates = UpdateConfig()
class ConfigManager:
@ -274,6 +287,18 @@ class ConfigManager:
if len(config_dict['llm']['model_rankings']) > 10:
yaml_lines.append(" # ... (edit config to see all options)")
# Add update settings
yaml_lines.extend([
"",
"# Auto-update system settings",
"updates:",
f" auto_check: {str(config_dict['updates']['auto_check']).lower()} # Check for updates automatically",
f" check_frequency_hours: {config_dict['updates']['check_frequency_hours']} # Hours between update checks",
f" auto_install: {str(config_dict['updates']['auto_install']).lower()} # Auto-install updates (not recommended)",
f" backup_before_update: {str(config_dict['updates']['backup_before_update']).lower()} # Create backup before updating",
f" notify_beta_releases: {str(config_dict['updates']['notify_beta_releases']).lower()} # Include beta releases in checks",
])
return '\n'.join(yaml_lines)
def update_config(self, **kwargs) -> RAGConfig:

458
mini_rag/updater.py Normal file
View File

@ -0,0 +1,458 @@
#!/usr/bin/env python3
"""
FSS-Mini-RAG Auto-Update System
Provides seamless GitHub-based updates with user-friendly interface.
Checks for new releases, downloads updates, and handles installation safely.
"""
import os
import sys
import json
import time
import shutil
import zipfile
import tempfile
import subprocess
from pathlib import Path
from typing import Optional, Dict, Any, Tuple
from datetime import datetime, timedelta
from dataclasses import dataclass
try:
import requests
REQUESTS_AVAILABLE = True
except ImportError:
REQUESTS_AVAILABLE = False
from .config import ConfigManager
@dataclass
class UpdateInfo:
"""Information about an available update."""
version: str
release_url: str
download_url: str
release_notes: str
published_at: str
is_newer: bool
class UpdateChecker:
"""
Handles checking for and applying updates from GitHub releases.
Features:
- Checks GitHub API for latest releases
- Downloads and applies updates safely with backup
- Respects user preferences and rate limiting
- Provides graceful fallbacks if network unavailable
"""
def __init__(self,
repo_owner: str = "FSSCoding",
repo_name: str = "Fss-Mini-Rag",
current_version: str = "2.1.0"):
self.repo_owner = repo_owner
self.repo_name = repo_name
self.current_version = current_version
self.github_api_url = f"https://api.github.com/repos/{repo_owner}/{repo_name}"
self.check_frequency_hours = 24 # Check once per day
# Paths
self.app_root = Path(__file__).parent.parent
self.cache_file = self.app_root / ".update_cache.json"
self.backup_dir = self.app_root / ".backup"
# User preferences
self.config = ConfigManager()
def should_check_for_updates(self) -> bool:
"""
Determine if we should check for updates now.
Respects:
- User preference to disable updates
- Rate limiting (once per day by default)
- Network availability
"""
if not REQUESTS_AVAILABLE:
return False
# Check user preference
if hasattr(self.config, 'updates') and not getattr(self.config.updates, 'auto_check', True):
return False
# Check if we've checked recently
if self.cache_file.exists():
try:
with open(self.cache_file, 'r') as f:
cache = json.load(f)
last_check = datetime.fromisoformat(cache.get('last_check', '2020-01-01'))
if datetime.now() - last_check < timedelta(hours=self.check_frequency_hours):
return False
except (json.JSONDecodeError, ValueError, KeyError):
pass # Ignore cache errors, will check anyway
return True
def check_for_updates(self) -> Optional[UpdateInfo]:
"""
Check GitHub API for the latest release.
Returns:
UpdateInfo if an update is available, None otherwise
"""
if not REQUESTS_AVAILABLE:
return None
try:
# Get latest release from GitHub API
response = requests.get(
f"{self.github_api_url}/releases/latest",
timeout=10,
headers={"Accept": "application/vnd.github.v3+json"}
)
if response.status_code != 200:
return None
release_data = response.json()
# Extract version info
latest_version = release_data.get('tag_name', '').lstrip('v')
release_notes = release_data.get('body', 'No release notes available.')
published_at = release_data.get('published_at', '')
release_url = release_data.get('html_url', '')
# Find download URL for source code
download_url = None
for asset in release_data.get('assets', []):
if asset.get('name', '').endswith('.zip'):
download_url = asset.get('browser_download_url')
break
# Fallback to source code zip
if not download_url:
download_url = f"https://github.com/{self.repo_owner}/{self.repo_name}/archive/refs/tags/v{latest_version}.zip"
# Check if this is a newer version
is_newer = self._is_version_newer(latest_version, self.current_version)
# Update cache
self._update_cache(latest_version, is_newer)
if is_newer:
return UpdateInfo(
version=latest_version,
release_url=release_url,
download_url=download_url,
release_notes=release_notes,
published_at=published_at,
is_newer=True
)
except Exception as e:
# Silently fail for network issues - don't interrupt user experience
pass
return None
def _is_version_newer(self, latest: str, current: str) -> bool:
"""
Compare version strings to determine if latest is newer.
Simple semantic version comparison supporting:
- Major.Minor.Patch (e.g., 2.1.0)
- Major.Minor (e.g., 2.1)
"""
def version_tuple(v):
return tuple(map(int, (v.split("."))))
try:
return version_tuple(latest) > version_tuple(current)
except (ValueError, AttributeError):
# If version parsing fails, assume it's newer to be safe
return latest != current
def _update_cache(self, latest_version: str, is_newer: bool):
"""Update the cache file with check results."""
cache_data = {
'last_check': datetime.now().isoformat(),
'latest_version': latest_version,
'is_newer': is_newer
}
try:
with open(self.cache_file, 'w') as f:
json.dump(cache_data, f, indent=2)
except Exception:
pass # Ignore cache write errors
def download_update(self, update_info: UpdateInfo, progress_callback=None) -> Optional[Path]:
"""
Download the update package to a temporary location.
Args:
update_info: Information about the update to download
progress_callback: Optional callback for progress updates
Returns:
Path to downloaded file, or None if download failed
"""
if not REQUESTS_AVAILABLE:
return None
try:
# Create temporary file for download
with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp_file:
tmp_path = Path(tmp_file.name)
# Download with progress tracking
response = requests.get(update_info.download_url, stream=True, timeout=30)
response.raise_for_status()
total_size = int(response.headers.get('content-length', 0))
downloaded = 0
with open(tmp_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if progress_callback and total_size > 0:
progress_callback(downloaded, total_size)
return tmp_path
except Exception as e:
# Clean up on error
if 'tmp_path' in locals() and tmp_path.exists():
tmp_path.unlink()
return None
def create_backup(self) -> bool:
"""
Create a backup of the current installation.
Returns:
True if backup created successfully
"""
try:
# Remove old backup if it exists
if self.backup_dir.exists():
shutil.rmtree(self.backup_dir)
# Create new backup
self.backup_dir.mkdir(exist_ok=True)
# Copy key files and directories
important_items = [
'mini_rag',
'rag-mini.py',
'rag-tui.py',
'requirements.txt',
'install_mini_rag.sh',
'install_windows.bat',
'README.md',
'assets'
]
for item in important_items:
src = self.app_root / item
if src.exists():
if src.is_dir():
shutil.copytree(src, self.backup_dir / item)
else:
shutil.copy2(src, self.backup_dir / item)
return True
except Exception as e:
return False
def apply_update(self, update_package_path: Path, update_info: UpdateInfo) -> bool:
"""
Apply the downloaded update.
Args:
update_package_path: Path to the downloaded update package
update_info: Information about the update
Returns:
True if update applied successfully
"""
try:
# Extract to temporary directory first
with tempfile.TemporaryDirectory() as tmp_dir:
tmp_path = Path(tmp_dir)
# Extract the archive
with zipfile.ZipFile(update_package_path, 'r') as zip_ref:
zip_ref.extractall(tmp_path)
# Find the extracted directory (may be nested)
extracted_dirs = [d for d in tmp_path.iterdir() if d.is_dir()]
if not extracted_dirs:
return False
source_dir = extracted_dirs[0]
# Copy files to application directory
important_items = [
'mini_rag',
'rag-mini.py',
'rag-tui.py',
'requirements.txt',
'install_mini_rag.sh',
'install_windows.bat',
'README.md'
]
for item in important_items:
src = source_dir / item
dst = self.app_root / item
if src.exists():
if dst.exists():
if dst.is_dir():
shutil.rmtree(dst)
else:
dst.unlink()
if src.is_dir():
shutil.copytree(src, dst)
else:
shutil.copy2(src, dst)
# Update version info
self._update_version_info(update_info.version)
return True
except Exception as e:
return False
def _update_version_info(self, new_version: str):
"""Update version information in the application."""
# Update __init__.py version
init_file = self.app_root / 'mini_rag' / '__init__.py'
if init_file.exists():
try:
content = init_file.read_text()
updated_content = content.replace(
f'__version__ = "{self.current_version}"',
f'__version__ = "{new_version}"'
)
init_file.write_text(updated_content)
except Exception:
pass
def rollback_update(self) -> bool:
"""
Rollback to the backup version if update failed.
Returns:
True if rollback successful
"""
if not self.backup_dir.exists():
return False
try:
# Restore from backup
for item in self.backup_dir.iterdir():
dst = self.app_root / item.name
if dst.exists():
if dst.is_dir():
shutil.rmtree(dst)
else:
dst.unlink()
if item.is_dir():
shutil.copytree(item, dst)
else:
shutil.copy2(item, dst)
return True
except Exception as e:
return False
def restart_application(self):
"""Restart the application after update."""
try:
# Get the current script path
current_script = sys.argv[0]
# Restart with the same arguments
if sys.platform.startswith('win'):
# Windows
subprocess.Popen([sys.executable] + sys.argv)
else:
# Unix-like systems
os.execv(sys.executable, [sys.executable] + sys.argv)
except Exception as e:
# If restart fails, just exit gracefully
print(f"\n✅ Update complete! Please restart the application manually.")
sys.exit(0)
def get_legacy_notification() -> Optional[str]:
"""
Check if this is a legacy version that needs urgent notification.
For users who downloaded before the auto-update system.
"""
try:
# Check if this is a very old version by looking for cache file
# Old versions won't have update cache, so we can detect them
app_root = Path(__file__).parent.parent
cache_file = app_root / ".update_cache.json"
# Also check version in __init__.py to see if it's old
init_file = app_root / 'mini_rag' / '__init__.py'
if init_file.exists():
content = init_file.read_text()
if '__version__ = "2.0.' in content or '__version__ = "1.' in content:
return """
🚨 IMPORTANT UPDATE AVAILABLE 🚨
Your version of FSS-Mini-RAG is missing critical updates!
🔧 Recent improvements include:
• Fixed LLM response formatting issues
• Added context window configuration
• Improved Windows installer reliability
• Added auto-update system (this notification!)
📥 Please update by downloading the latest version:
https://github.com/FSSCoding/Fss-Mini-Rag/releases/latest
💡 After updating, you'll get automatic update notifications!
"""
except Exception:
pass
return None
# Global convenience functions
_updater_instance = None
def check_for_updates() -> Optional[UpdateInfo]:
"""Global function to check for updates."""
global _updater_instance
if _updater_instance is None:
_updater_instance = UpdateChecker()
if _updater_instance.should_check_for_updates():
return _updater_instance.check_for_updates()
return None
def get_updater() -> UpdateChecker:
"""Get the global updater instance."""
global _updater_instance
if _updater_instance is None:
_updater_instance = UpdateChecker()
return _updater_instance

141
rag-mini.py
View File

@ -21,6 +21,12 @@ try:
from mini_rag.ollama_embeddings import OllamaEmbedder
from mini_rag.llm_synthesizer import LLMSynthesizer
from mini_rag.explorer import CodeExplorer
# Update system (graceful import)
try:
from mini_rag.updater import check_for_updates, get_updater
UPDATER_AVAILABLE = True
except ImportError:
UPDATER_AVAILABLE = False
except ImportError as e:
print("❌ Error: Missing dependencies!")
print()
@ -444,6 +450,119 @@ def explore_interactive(project_path: Path):
print("Make sure the project is indexed first: rag-mini index <project>")
sys.exit(1)
def show_discrete_update_notice():
"""Show a discrete, non-intrusive update notice for CLI users."""
if not UPDATER_AVAILABLE:
return
try:
update_info = check_for_updates()
if update_info:
# Very discrete notice - just one line
print(f"🔄 (Update v{update_info.version} available - run 'rag-mini check-update' to learn more)")
except Exception:
# Silently ignore any update check failures
pass
def handle_check_update():
"""Handle the check-update command."""
if not UPDATER_AVAILABLE:
print("❌ Update system not available")
print("💡 Try updating to the latest version manually from GitHub")
return
try:
print("🔍 Checking for updates...")
update_info = check_for_updates()
if update_info:
print(f"\n🎉 Update Available: v{update_info.version}")
print("=" * 50)
print("\n📋 What's New:")
notes_lines = update_info.release_notes.split('\n')[:10] # First 10 lines
for line in notes_lines:
if line.strip():
print(f" {line.strip()}")
print(f"\n🔗 Release Page: {update_info.release_url}")
print(f"\n🚀 To install: rag-mini update")
print("💡 Or update manually from GitHub releases")
else:
print("✅ You're already on the latest version!")
except Exception as e:
print(f"❌ Failed to check for updates: {e}")
print("💡 Try updating manually from GitHub")
def handle_update():
"""Handle the update command."""
if not UPDATER_AVAILABLE:
print("❌ Update system not available")
print("💡 Try updating manually from GitHub")
return
try:
print("🔍 Checking for updates...")
update_info = check_for_updates()
if not update_info:
print("✅ You're already on the latest version!")
return
print(f"\n🎉 Update Available: v{update_info.version}")
print("=" * 50)
# Show brief release notes
notes_lines = update_info.release_notes.split('\n')[:5]
for line in notes_lines:
if line.strip():
print(f" • {line.strip()}")
# Confirm update
confirm = input(f"\n🚀 Install v{update_info.version}? [Y/n]: ").strip().lower()
if confirm in ['', 'y', 'yes']:
updater = get_updater()
print(f"\n📥 Downloading v{update_info.version}...")
# Progress callback
def show_progress(downloaded, total):
if total > 0:
percent = (downloaded / total) * 100
bar_length = 30
filled = int(bar_length * downloaded / total)
bar = "█" * filled + "░" * (bar_length - filled)
print(f"\r [{bar}] {percent:.1f}%", end="", flush=True)
# Download and install
update_package = updater.download_update(update_info, show_progress)
if not update_package:
print("\n❌ Download failed. Please try again later.")
return
print("\n💾 Creating backup...")
if not updater.create_backup():
print("⚠️ Backup failed, but continuing anyway...")
print("🔄 Installing update...")
if updater.apply_update(update_package, update_info):
print("✅ Update successful!")
print("🚀 Restarting...")
updater.restart_application()
else:
print("❌ Update failed.")
print("🔙 Attempting rollback...")
if updater.rollback_update():
print("✅ Rollback successful.")
else:
print("❌ Rollback failed. You may need to reinstall.")
else:
print("Update cancelled.")
except Exception as e:
print(f"❌ Update failed: {e}")
print("💡 Try updating manually from GitHub")
def main():
"""Main CLI interface."""
# Check virtual environment
@ -466,10 +585,10 @@ Examples:
"""
)
parser.add_argument('command', choices=['index', 'search', 'explore', 'status'],
parser.add_argument('command', choices=['index', 'search', 'explore', 'status', 'update', 'check-update'],
help='Command to execute')
parser.add_argument('project_path', type=Path,
help='Path to project directory (REQUIRED)')
parser.add_argument('project_path', type=Path, nargs='?',
help='Path to project directory (REQUIRED except for update commands)')
parser.add_argument('query', nargs='?',
help='Search query (for search command)')
parser.add_argument('--force', action='store_true',
@ -487,6 +606,19 @@ Examples:
if args.verbose:
logging.getLogger().setLevel(logging.INFO)
# Handle update commands first (don't require project_path)
if args.command == 'check-update':
handle_check_update()
return
elif args.command == 'update':
handle_update()
return
# All other commands require project_path
if not args.project_path:
print("❌ Project path required for this command")
sys.exit(1)
# Validate project path
if not args.project_path.exists():
print(f"❌ Project path does not exist: {args.project_path}")
@ -496,6 +628,9 @@ Examples:
print(f"❌ Project path is not a directory: {args.project_path}")
sys.exit(1)
# Show discrete update notification for regular commands (non-intrusive)
show_discrete_update_notice()
# Execute command
if args.command == 'index':
index_project(args.project_path, args.force)

102
rag-tui.py
View File

@ -10,6 +10,13 @@ import json
from pathlib import Path
from typing import Optional, List, Dict, Any
# Update system (graceful import)
try:
from mini_rag.updater import check_for_updates, get_updater, get_legacy_notification
UPDATER_AVAILABLE = True
except ImportError:
UPDATER_AVAILABLE = False
# Simple TUI without external dependencies
class SimpleTUI:
def __init__(self):
@ -2083,12 +2090,107 @@ Your suggested question (under 10 words):"""
input("Press Enter to continue...")
def check_for_updates_notification(self):
"""Check for updates and show notification if available."""
if not UPDATER_AVAILABLE:
return
try:
# Check for legacy notification first
legacy_notice = get_legacy_notification()
if legacy_notice:
print("🔔" + "=" * 58 + "🔔")
print(legacy_notice)
print("🔔" + "=" * 58 + "🔔")
print()
return
# Check for regular updates
update_info = check_for_updates()
if update_info:
print("🎉" + "=" * 58 + "🎉")
print(f"🔄 Update Available: v{update_info.version}")
print()
print("📋 What's New:")
# Show first few lines of release notes
notes_lines = update_info.release_notes.split('\n')[:3]
for line in notes_lines:
if line.strip():
print(f" • {line.strip()}")
print()
# Simple update prompt
update_choice = self.get_input("🚀 Install update now? [y/N]", "n").lower()
if update_choice in ['y', 'yes']:
self.perform_update(update_info)
else:
print("💡 You can update anytime from the Configuration menu!")
print("🎉" + "=" * 58 + "🎉")
print()
except Exception:
# Silently ignore update check errors - don't interrupt user experience
pass
def perform_update(self, update_info):
"""Perform the actual update with progress display."""
try:
updater = get_updater()
print(f"\n📥 Downloading v{update_info.version}...")
# Progress callback
def show_progress(downloaded, total):
if total > 0:
percent = (downloaded / total) * 100
bar_length = 30
filled = int(bar_length * downloaded / total)
bar = "█" * filled + "░" * (bar_length - filled)
print(f"\r [{bar}] {percent:.1f}%", end="", flush=True)
# Download update
update_package = updater.download_update(update_info, show_progress)
if not update_package:
print("\n❌ Download failed. Please try again later.")
input("Press Enter to continue...")
return
print("\n💾 Creating backup...")
if not updater.create_backup():
print("⚠️ Backup failed, but continuing anyway...")
print("🔄 Installing update...")
if updater.apply_update(update_package, update_info):
print("✅ Update successful!")
print("🚀 Restarting application...")
input("Press Enter to restart...")
updater.restart_application()
else:
print("❌ Update failed.")
print("🔙 Attempting rollback...")
if updater.rollback_update():
print("✅ Rollback successful.")
else:
print("❌ Rollback failed. You may need to reinstall.")
input("Press Enter to continue...")
except Exception as e:
print(f"❌ Update error: {e}")
input("Press Enter to continue...")
def main_menu(self):
"""Main application loop."""
first_run = True
while True:
self.clear_screen()
self.print_header()
# Check for updates on first run only (non-intrusive)
if first_run:
self.check_for_updates_notification()
first_run = False
# Show current project status prominently
if self.project_path:
rag_dir = self.project_path / '.mini-rag'