# app.py (updated with better JSON handling) import gradio as gr from transformers import AutoTokenizer, AutoModelForCausalLM import requests import importlib.util import os import hashlib import json from datetime import datetime CONFIG = { "model_name": "deepseek-ai/deepseek-coder-1.3b-instruct", "update_url": "https://raw.githubusercontent.com/[YOUR_USERNAME]/deepseek-updates/main/", "manifest_url": "https://raw.githubusercontent.com/[YOUR_USERNAME]/deepseek-updates/main/manifest.json", "allowed_modules": ["response_handler", "updater"], "update_frequency": 6 # hours } tokenizer = AutoTokenizer.from_pretrained(CONFIG["model_name"]) model = AutoModelForCausalLM.from_pretrained(CONFIG["model_name"]) def verify_file_integrity(content, expected_hash): sha256 = hashlib.sha256() sha256.update(content.encode('utf-8')) return sha256.hexdigest() == expected_hash def check_for_updates(): try: # Get update manifest with explicit error handling response = requests.get(CONFIG["manifest_url"], timeout=10) # Check for valid JSON response try: manifest = response.json() except json.JSONDecodeError as e: # Check if we got an HTML error page instead of JSON if "" in response.text[:50].lower(): return "Update failed: Server returned HTML instead of JSON" return f"JSON parse error: {str(e)}" # Check last update time last_update_path = "last_update.txt" if os.path.exists(last_update_path): with open(last_update_path, 'r') as f: last_update_str = f.read().strip() if last_update_str: last_update = datetime.fromisoformat(last_update_str) time_since_update = (datetime.utcnow() - last_update).total_seconds() / 3600 if time_since_update < CONFIG["update_frequency"]: return "Too soon for update check" # Process updates updates_applied = [] for module_name in CONFIG["allowed_modules"]: if module_name in manifest.get("modules", {}): module_info = manifest["modules"][module_name] file_path = f"{module_name}.py" # Download and verify update file_response = requests.get(CONFIG["update_url"] + file_path, timeout=10) if file_response.status_code == 200: content = file_response.text if verify_file_integrity(content, module_info.get("sha256", "")): with open(file_path, 'w') as f: f.write(content) updates_applied.append(module_name) # Update timestamp with open(last_update_path, 'w') as f: f.write(datetime.utcnow().isoformat()) return f"Updates applied to: {', '.join(updates_applied)}" if updates_applied else "No updates available" except requests.exceptions.RequestException as e: return f"Network error: {str(e)}" except Exception as e: return f"Update failed: {str(e)}" def safe_load_module(module_name): try: spec = importlib.util.spec_from_file_location(module_name, f"{module_name}.py") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module except Exception as e: print(f"Error loading {module_name}: {str(e)}") # Return a fallback module from types import ModuleType return ModuleType(module_name) # Load core modules with fallback response_handler = safe_load_module("response_handler") updater = safe_load_module("updater") def process_query(prompt): if "/update" in prompt: return check_for_updates() # Handle case where response_handler failed to load if not hasattr(response_handler, "generate_response"): return "System error: Response module not loaded properly" return response_handler.generate_response(prompt, tokenizer, model) interface = gr.Interface( fn=process_query, inputs=gr.Textbox(lines=3, placeholder="Enter your query..."), outputs="text", title="Self-Updating DeepSeek AI", description="This AI can update its own code. Type '/update' to check for improvements." ) if __name__ == "__main__": interface.launch(server_port=7860, share=True)