Spaces:
Sleeping
Sleeping
| # 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 "<!DOCTYPE html>" 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) |