import os import uuid from flask import Flask, request, jsonify, Response, Blueprint, render_template_string import google.generativeai as genai import json import logging import threading import time from datetime import datetime from huggingface_hub import HfApi, hf_hub_download, create_repo, snapshot_download from huggingface_hub.utils import RepositoryNotFoundError, HfHubHTTPError from werkzeug.utils import secure_filename # Unused in final version, but kept for context if file uploads were needed from dotenv import load_dotenv import requests import importlib.util import sys import shutil # For deleting directories # Load environment variables from .env file load_dotenv() app = Flask(__name__) # --- Configuration from Code 2 and New/Modified for Code 1 --- REPO_ID = "Kgshop/testsynk" HF_TOKEN_WRITE = os.getenv("HF_TOKEN") # Ensure this is set for write access HF_TOKEN_READ = os.getenv("HF_TOKEN_READ") # Optional, falls back to WRITE if not set GENERATED_SITES_DIR = 'generated_sites' GENERATED_APPS_METADATA_FILE = os.path.join(GENERATED_SITES_DIR, 'generated_apps.json') # Global dictionary to hold dynamically loaded blueprints # Key: app_uuid, Value: blueprint_object _GENERATED_BLUEPRINTS = {} # Ensure base directory for generated sites exists if not os.path.exists(GENERATED_SITES_DIR): os.makedirs(GENERATED_SITES_DIR) # --- Logging Setup from Code 2 --- logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # --- Hugging Face Utility Functions (Adapted from Code 2 for directory sync) --- DOWNLOAD_RETRIES = 3 DOWNLOAD_DELAY = 5 def create_hf_repo_if_not_exists(): """Ensures the Hugging Face dataset repository exists.""" if not HF_TOKEN_WRITE: logging.warning("HF_TOKEN for writing not set. Cannot create Hugging Face repo.") return False try: api = HfApi(token=HF_TOKEN_WRITE) api.create_repo(repo_id=REPO_ID, repo_type="dataset", exist_ok=True, private=True) logging.info(f"Hugging Face dataset repo '{REPO_ID}' ensured to exist.") return True except Exception as e: logging.error(f"Failed to create/ensure Hugging Face repo '{REPO_ID}': {e}") return False def download_data_from_hf(target_dir="."): """Downloads the entire repository snapshot from Hugging Face.""" if not REPO_ID: logging.warning("REPO_ID not set. Skipping Hugging Face download.") return False token_to_use = HF_TOKEN_READ if HF_TOKEN_READ else HF_TOKEN_WRITE if not token_to_use: logging.warning("No Hugging Face token available for download. Skipping.") return False logging.info(f"Attempting full repo snapshot download for '{REPO_ID}' to '{target_dir}'...") success = False for attempt in range(DOWNLOAD_RETRIES + 1): try: # `snapshot_download` downloads the entire repo to the specified local_dir logging.info(f"Downloading snapshot of repo '{REPO_ID}' (Attempt {attempt + 1}/{DOWNLOAD_RETRIES + 1})...") # Clear existing content in target_dir if it's the `generated_sites` directory if os.path.exists(target_dir) and target_dir == GENERATED_SITES_DIR: logging.info(f"Clearing existing local '{target_dir}' before download to ensure full sync.") # Keep the base directory but remove its contents for item in os.listdir(target_dir): item_path = os.path.join(target_dir, item) if os.path.isfile(item_path): os.remove(item_path) elif os.path.isdir(item_path): shutil.rmtree(item_path) # Download the snapshot into the current working directory, then copy `generated_sites` downloaded_repo_path = snapshot_download( repo_id=REPO_ID, repo_type="dataset", token=token_to_use, local_dir="hf_download_temp", # Download to a temp directory first local_dir_use_symlinks=False, # force_download=True # Uncomment to always redownload, useful for development ) logging.info(f"Repo snapshot downloaded to {downloaded_repo_path}") # Move contents of `generated_sites` from temp download to actual GENERATED_SITES_DIR source_generated_sites = os.path.join(downloaded_repo_path, 'generated_sites') if os.path.exists(source_generated_sites): for item in os.listdir(source_generated_sites): s = os.path.join(source_generated_sites, item) d = os.path.join(GENERATED_SITES_DIR, item) if os.path.isdir(s): if os.path.exists(d): shutil.rmtree(d) # Remove existing sub-dir before moving shutil.move(s, d) else: if os.path.exists(d): os.remove(d) # Remove existing file before moving shutil.move(s, d) logging.info(f"Moved generated_sites content from {source_generated_sites} to {GENERATED_SITES_DIR}") else: logging.warning(f"No 'generated_sites' folder found in the downloaded repo {downloaded_repo_path}.") # Clean up the temporary download directory if os.path.exists(downloaded_repo_path): shutil.rmtree(downloaded_repo_path) logging.info(f"Cleaned up temporary download directory {downloaded_repo_path}.") success = True break except RepositoryNotFoundError: logging.error(f"Repository {REPO_ID} not found on Hugging Face. Ensuring its creation...") if create_hf_repo_if_not_exists(): logging.info(f"Repository {REPO_ID} created. Retrying download after creation.") continue # Retry the download as repo now exists return False # Cannot proceed without repo except HfHubHTTPError as e: logging.error(f"HTTP error downloading repo (Attempt {attempt + 1}): {e}. Retrying in {DOWNLOAD_DELAY}s...") except requests.exceptions.RequestException as e: logging.error(f"Network error downloading repo (Attempt {attempt + 1}): {e}. Retrying in {DOWNLOAD_DELAY}s...", exc_info=True) except Exception as e: logging.error(f"Unexpected error during repo snapshot download (Attempt {attempt + 1}): {e}. Retrying in {DOWNLOAD_DELAY}s...", exc_info=True) if attempt < DOWNLOAD_RETRIES: time.sleep(DOWNLOAD_DELAY) if not success: logging.error(f"Failed to download repo '{REPO_ID}' after {DOWNLOAD_RETRIES + 1} attempts.") return success def upload_data_to_hf(target_dir="."): """Uploads the specified directory to Hugging Face.""" if not HF_TOKEN_WRITE: logging.warning("HF_TOKEN for writing not set. Skipping upload to Hugging Face.") return try: api = HfApi(token=HF_TOKEN_WRITE) logging.info(f"Starting upload of directory '{target_dir}' to HF repo '{REPO_ID}'...") # Use upload_folder to sync the entire directory # path_in_repo='.' means upload contents of target_dir to root of repo. # If target_dir is 'generated_sites', it will upload 'generated_sites' itself to the repo root. # This is desired to keep the folder structure consistent. api.upload_folder( folder_path=target_dir, repo_id=REPO_ID, repo_type="dataset", token=HF_TOKEN_WRITE, commit_message=f"Sync {target_dir} {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" ) logging.info(f"Directory '{target_dir}' successfully uploaded to Hugging Face.") except Exception as e: logging.error(f"Error during Hugging Face folder upload for '{target_dir}': {e}", exc_info=True) def load_generated_apps_metadata(): """Loads metadata about generated apps from generated_apps.json.""" if not os.path.exists(GENERATED_APPS_METADATA_FILE): logging.warning(f"Metadata file {GENERATED_APPS_METADATA_FILE} not found. Creating empty structure.") return {"apps": {}} try: with open(GENERATED_APPS_METADATA_FILE, 'r', encoding='utf-8') as f: data = json.load(f) if not isinstance(data, dict) or "apps" not in data: logging.warning(f"Invalid format in {GENERATED_APPS_METADATA_FILE}. Resetting to empty.") return {"apps": {}} return data except json.JSONDecodeError as e: logging.error(f"Error decoding JSON from {GENERATED_APPS_METADATA_FILE}: {e}. File might be corrupt. Resetting to empty.") return {"apps": {}} except Exception as e: logging.error(f"Unexpected error loading metadata from {GENERATED_APPS_METADATA_FILE}: {e}. Resetting to empty.", exc_info=True) return {"apps": {}} def save_generated_apps_metadata(metadata): """Saves metadata about generated apps to generated_apps.json and triggers HF upload.""" try: with open(GENERATED_APPS_METADATA_FILE, 'w', encoding='utf-8') as f: json.dump(metadata, f, ensure_ascii=False, indent=4) logging.info(f"Metadata saved to {GENERATED_APPS_METADATA_FILE}") # Trigger upload of the entire generated_sites directory including the metadata file upload_data_to_hf(GENERATED_SITES_DIR) except Exception as e: logging.error(f"Error saving metadata to {GENERATED_APPS_METADATA_FILE}: {e}", exc_info=True) def load_and_register_blueprint(app_uuid, app_path): """ Dynamically loads a Flask Blueprint from 'app.py' within a given app_path and registers it with the main Flask app. """ if app_uuid in _GENERATED_BLUEPRINTS: logging.info(f"Blueprint {app_uuid} already loaded.") return True module_path = os.path.join(app_path, 'app.py') if not os.path.exists(module_path): logging.error(f"App module not found at {module_path} for UUID {app_uuid}. Skipping.") return False try: # Create a module spec from the file path # Use a unique name for the module to avoid conflicts, e.g., 'generated_app_UUID' spec = importlib.util.spec_from_file_location(f"generated_app_{app_uuid}", module_path) if spec is None: raise ImportError(f"Could not create module spec for {module_path}") # Create a new module based on the spec module = importlib.util.module_from_spec(spec) # Add to sys.modules for proper import behavior and to allow other modules to import it if needed sys.modules[spec.name] = module # Execute the module's code spec.loader.exec_module(module) # Look for the 'generated_blueprint' instance in the module blueprint_found = getattr(module, 'generated_blueprint', None) if isinstance(blueprint_found, Blueprint): app.register_blueprint(blueprint_found, url_prefix=f'/generated_sites/{app_uuid}') _GENERATED_BLUEPRINTS[app_uuid] = blueprint_found logging.info(f"Blueprint '{blueprint_found.name}' for UUID {app_uuid} loaded and registered under /generated_sites/{app_uuid}") return True else: logging.error(f"No Flask Blueprint named 'generated_blueprint' found in {module_path} for UUID {app_uuid}.") return False except Exception as e: logging.error(f"Failed to load and register blueprint from {module_path} for UUID {app_uuid}: {e}", exc_info=True) return False def load_all_generated_blueprints(): """Loads all generated blueprints listed in metadata on application startup.""" logging.info("Attempting to load all previously generated blueprints...") metadata = load_generated_apps_metadata() if not metadata.get("apps"): logging.info("No generated apps metadata found to load.") return for app_uuid, app_info in metadata.get("apps", {}).items(): app_dir = os.path.join(GENERATED_SITES_DIR, app_uuid) if not os.path.exists(app_dir): logging.warning(f"Directory for app {app_uuid} not found locally at {app_dir}. Skipping load.") continue load_and_register_blueprint(app_uuid, app_dir) logging.info("Completed loading of existing blueprints.") def periodic_backup(): """Starts a periodic backup thread for the generated_sites directory.""" backup_interval = 1800 # 30 minutes logging.info(f"Setting up periodic backup of generated_sites directory every {backup_interval} seconds.") while True: time.sleep(backup_interval) logging.info("Starting periodic backup of generated_sites directory...") upload_data_to_hf(GENERATED_SITES_DIR) logging.info("Periodic backup finished.") # --- Flask App Routes and AI Generation Logic --- # Initial HTML template for the generator UI (from Code 1, slightly adapted for clarity) html_template = """ EVA - Генератор Сайтов

EVA

Генератор сайтов на базе ИИ

""" def generate_website_code_from_prompt(user_prompt): """ Generates Flask Blueprint Python code based on the user prompt. """ try: # Using GEMINI_API_KEY from .env gemini_api_key = os.getenv("GEMINI_API_KEY") if not gemini_api_key: raise ValueError("GEMINI_API_KEY environment variable not set. Please add it to your .env file.") genai.configure(api_key=gemini_api_key) except Exception as e: logging.error(f"Error configuring GenAI: {e}") raise ValueError(f"Не удалось настроить Google AI. Проблема с конфигурацией. Ошибка: {e}") if not user_prompt or not user_prompt.strip(): raise ValueError("Текстовый запрос (промпт) не может быть пустым.") # Updated system instruction for generating a Flask Blueprint system_instruction = ( "You are an expert web developer specializing in Flask. Your task is to generate a complete, " "self-contained Python Flask Blueprint code string based on the user's request. " "This Blueprint will be dynamically loaded into a larger Flask application.\n" "The generated Python string must adhere to the following rules:\n" "1. It must start with necessary imports: `from flask import Blueprint, render_template_string, request, jsonify`.\n" " If managing data via JSON file, also include `import json` and `import os`.\n" "2. It must define a Flask Blueprint instance, named `generated_blueprint`. For example: `generated_blueprint = Blueprint('my_app_name', __name__)`.\n" "3. All HTML, CSS (in `