Zeus Cloud Dashboard
Welcome, {{ username }}!
-Current Path: {{ current_path }}
- - + # Generate HF URLs for previews + for file_info in user_files: + try: + # Use hf_hub_url for generating the correct URL (handles private repos if token is set) + # We need the non-download URL for previews like images/iframes + file_info['preview_url'] = hf_hub_url( + repo_id=REPO_ID, + filename=file_info['hf_path'], + repo_type='dataset', + revision='main' # or specific commit hash + ) + # URL for modal (image/video) might need '?download=true' or token if private, handled by JS/direct link + file_info['modal_url'] = file_info['preview_url'] # Start with base url + # Add token parameter IF necessary and repo is private (complex logic, maybe simplify) + # Simplification: Assume public or token handled by browser session/cookies if needed + # OR pass token to JS if needed for fetch/modal + file_info['download_url'] = url_for('download_file', hf_path=file_info['hf_path'], original_filename=file_info['original_filename']) + file_info['delete_url'] = url_for('delete_file', hf_path=file_info['hf_path']) + except Exception as e: + logging.error(f"Error generating HF URL for {file_info.get('hf_path')}: {e}") + file_info['preview_url'] = '#' + file_info['modal_url'] = '#' + file_info['download_url'] = '#' + file_info['delete_url'] = '#' + + + html = ''' + + + + + +
+
Панель управления Zeus Cloud
+Пользователь: {{ username }}
+
+ {% with messages = get_flashed_messages(with_categories=true) %}
+ {% if messages %}
+ {% for category, message in messages %}
+
+
+ {{ message }}
+ {% endfor %}
+ {% endif %}
+ {% endwith %}
+
+
+ 0%
+ Ваши файлы
+Ваши файлы под надежной защитой квантовой криптографии*
(*маркетинговое заявление)
+ {% for file in user_files %}
+
-{% with messages = get_flashed_messages(with_categories=true) %}
- {% if messages %}{% for category, message in messages %}
+
+ {% endfor %}
+ {% if not user_files %}
+
+ {% if file['type'] == 'video' %}
+
+ {% elif file['type'] == 'image' %}
+ 
+ {% elif file['type'] == 'pdf' %}
+
+ {% elif file['type'] == 'text' %}
+
+ {% else %}
+
+
+ Нет предпросмотра для {{ file.type }}
+ {% endif %} +Вы еще не загрузили ни одного файла.
+ {% endif %} +{{ message }}
{% endfor %}{% endif %}
-{% endwith %}
-
-
-
-
-
-
-
Files and Folders
-{% if parent_path is not none %} -⬆️ Go Up -{% endif %} - -
- {% for item in items_in_current_path %}
- ')
-def download_item(item_path):
+@app.route('/download//')
+def download_file(hf_path, original_filename):
if 'username' not in session:
- flash('Please log in.', 'info')
+ flash('Пожалуйста, войдите в систему!', 'error')
return redirect(url_for('login'))
username = session['username']
data = load_data()
- if username not in data.get('users', {}):
- return redirect(url_for('login'))
-
- # Normalize the path passed in the URL
- item_path_normalized = normalize_path(item_path)
-
- user_items = data['users'][username].get('items', [])
- item_to_download = next((item for item in user_items if item['path'] == item_path_normalized and item['type'] == 'file'), None)
-
- is_admin_request = request.args.get('admin_context') == 'true' # Check if admin initiated
-
- if not item_to_download:
- # If admin is trying to download, check across all users (less efficient)
- if is_admin_request:
- found = False
- for uname, udata in data.get('users', {}).items():
- item_to_download = next((item for item in udata.get('items', []) if item['path'] == item_path_normalized and item['type'] == 'file'), None)
- if item_to_download:
- # Check admin privileges here if needed
- found = True
- break
- if not found:
- flash('File not found.', 'error')
- referer = request.referrer or url_for('admin_panel')
- return redirect(referer)
- else:
- flash('File not found or you do not have permission.', 'error')
+ is_admin_request = request.referrer and 'admhosto' in request.referrer
+
+ # Permission check: User must own the file OR it's an admin request
+ if not is_admin_request:
+ if username not in data['users']:
+ session.pop('username', None)
+ flash('Пользователь не найден!', 'error')
+ return redirect(url_for('login'))
+ user_files = data['users'][username].get('files', [])
+ if not any(file.get('hf_path') == hf_path for file in user_files):
+ flash('У вас нет доступа к этому файлу или файл не найден!', 'error')
return redirect(url_for('dashboard'))
+ else: # Admin request check if user exists
+ owner_username = hf_path.split('/')[1] if '/' in hf_path else None
+ if owner_username not in data['users']:
+ flash(f'Владелец файла ({owner_username}) не найден!', 'error')
+ return redirect(url_for('admin_panel'))
- if not item_to_download.get('hf_path'):
- flash('File metadata is incomplete (missing HF path). Cannot download.', 'error')
- referer = request.referrer or url_for('dashboard')
- return redirect(referer)
-
-
- hf_path = item_to_download['hf_path']
- original_filename = item_to_download.get('original_filename', item_to_download['name'])
- download_url = get_hf_item_url(hf_path, is_download=True)
-
- logging.info(f"Attempting download for {username if not is_admin_request else 'admin'} - Item: {item_path_normalized}, HF Path: {hf_path}, URL: {download_url}")
-
+ # Generate the download URL (usually includes ?download=true)
+ # Using hf_hub_url might be safer if direct URL structure changes
try:
+ # Get a URL that forces download
+ file_url = hf_hub_url(repo_id=REPO_ID, filename=hf_path, repo_type="dataset", revision="main")
+ # Append download=true if not already present (hf_hub_url might not add it)
+ file_url += "?download=true"
+
+ api = HfApi()
headers = {}
if HF_TOKEN_READ:
- headers["Authorization"] = f"Bearer {HF_TOKEN_READ}"
+ headers["authorization"] = f"Bearer {HF_TOKEN_READ}"
- response = requests.get(download_url, headers=headers, stream=True, timeout=60) # Added timeout
- response.raise_for_status()
-
- # Stream the download
- file_content = BytesIO(response.content) # Read into memory (for moderate files)
- # For very large files, stream response directly:
- # return Response(stream_with_context(response.iter_content(chunk_size=8192)), content_type=response.headers['Content-Type'])
- # But send_file is simpler for most cases
+ # Use requests to stream the download
+ response = requests.get(file_url, headers=headers, stream=True)
+ response.raise_for_status() # Check for HTTP errors (like 404 Not Found)
+ # Stream the content using send_file
return send_file(
- file_content,
+ BytesIO(response.content), # Read content into memory (consider streaming for large files)
as_attachment=True,
- download_name=original_filename,
- mimetype=mimetypes.guess_type(original_filename)[0] or 'application/octet-stream'
+ download_name=original_filename, # Use the original filename for the user
+ mimetype='application/octet-stream' # Generic mimetype
)
except requests.exceptions.RequestException as e:
logging.error(f"Error downloading file from HF ({hf_path}): {e}")
- flash(f'Error downloading file: {e}', 'error')
+ flash('Ошибка скачивания файла с сервера хранилища!', 'error')
except Exception as e:
- logging.error(f"Unexpected error during download ({hf_path}): {e}", exc_info=True)
- flash('An unexpected error occurred during download.', 'error')
+ logging.error(f"Unexpected error during download ({hf_path}): {e}")
+ flash('Произошла непредвиденная ошибка при скачивании файла.', 'error')
+
+ # Redirect back based on where the request came from
+ if is_admin_request:
+ owner_username = hf_path.split('/')[1] if '/' in hf_path else None
+ if owner_username:
+ return redirect(url_for('admin_user_files', username=owner_username))
+ else:
+ return redirect(url_for('admin_panel')) # Fallback
+ else:
+ return redirect(url_for('dashboard'))
- referer = request.referrer or url_for('dashboard')
- return redirect(referer)
-@app.route('/delete/', methods=['POST'])
-def delete_item(item_path):
+@app.route('/delete/')
+def delete_file(hf_path):
if 'username' not in session:
- flash('Please log in.', 'info')
+ flash('Пожалуйста, войдите в систему!', 'error')
return redirect(url_for('login'))
+ if not HF_TOKEN_WRITE:
+ flash('Удаление невозможна: токен для записи на Hugging Face не настроен.', 'error')
+ return redirect(url_for('dashboard'))
+
username = session['username']
data = load_data()
- if username not in data.get('users', {}):
+ if username not in data['users'] or 'files' not in data['users'][username]:
+ session.pop('username', None)
+ flash('Пользователь или его файлы не найдены!', 'error')
return redirect(url_for('login'))
- item_path_normalized = normalize_path(item_path)
- user_items = data['users'][username].get('items', [])
- item_to_delete = next((item for item in user_items if item['path'] == item_path_normalized), None)
-
- if not item_to_delete:
- flash('Item not found.', 'error')
- # Try to guess the previous path if possible
- parent = '/'.join(item_path_normalized.split('/')[:-1]) or '/'
- return redirect(url_for('dashboard', path=parent))
-
- current_view_path = '/'.join(item_path_normalized.split('/')[:-1]) or '/' # Path user was viewing
-
- api = get_hf_api()
- errors = []
- deleted_hf_paths = []
- items_to_remove_from_db = []
-
- if item_to_delete['type'] == 'file':
- items_to_remove_from_db.append(item_to_delete)
- hf_path = item_to_delete.get('hf_path')
- if hf_path:
- deleted_hf_paths.append(hf_path)
- else:
- logging.warning(f"File item {item_path_normalized} missing hf_path, only removing from DB.")
-
- elif item_to_delete['type'] == 'folder':
- items_to_remove_from_db.append(item_to_delete)
- # Find all children (files and subfolders) recursively
- folder_prefix = item_path_normalized + ('/' if item_path_normalized != '/' else '')
- children_to_delete = [item for item in user_items if item['path'].startswith(folder_prefix) and item['path'] != item_path_normalized]
-
- for child in children_to_delete:
- items_to_remove_from_db.append(child)
- if child['type'] == 'file' and child.get('hf_path'):
- deleted_hf_paths.append(child['hf_path'])
-
- # Try deleting the folder on HF Hub (might contain files not tracked or delete empty structure)
- # Construct the HF folder path correctly
- folder_hf_base = f"cloud_files/{username}"
- relative_folder_path = item_path_normalized.lstrip('/')
- hf_folder_path_to_delete = f"{folder_hf_base}/{relative_folder_path}" if relative_folder_path else folder_hf_base
-
- try:
- if HF_TOKEN_WRITE:
- logging.info(f"Attempting to delete HF folder: {hf_folder_path_to_delete}")
- api.delete_folder(
- folder_path=hf_folder_path_to_delete,
- repo_id=REPO_ID,
- repo_type="dataset",
- token=HF_TOKEN_WRITE,
- commit_message=f"User {username} deleted folder {item_path_normalized} and contents"
- )
- logging.info(f"Successfully deleted HF folder: {hf_folder_path_to_delete}")
- # If folder deletion worked, we don't need to delete individual files listed in deleted_hf_paths
- # But let's keep the individual deletion logic just in case delete_folder fails partially or has different semantics.
- # For safety, we'll still attempt individual deletes if delete_folder seems to fail.
- # Clear the list if delete_folder succeeds reliably? For now, keep both attempts.
- else:
- logging.warning(f"HF_TOKEN_WRITE not set. Cannot delete folder {hf_folder_path_to_delete} from HF.")
-
-
- except Exception as e:
- logging.error(f"Error deleting folder {hf_folder_path_to_delete} from HF Hub: {e}. Proceeding with individual file deletions if any.")
- errors.append(f"Could not fully remove folder '{item_to_delete['name']}' from storage. Some files might remain.")
-
-
- # Delete individual files from HF if listed (covers single file delete and folder contents)
- if HF_TOKEN_WRITE:
- for hf_path_to_delete in deleted_hf_paths:
- try:
- logging.info(f"Deleting HF file: {hf_path_to_delete}")
- api.delete_file(
- path_in_repo=hf_path_to_delete,
- repo_id=REPO_ID,
- repo_type="dataset",
- token=HF_TOKEN_WRITE,
- commit_message=f"User {username} deleted item associated with {hf_path_to_delete}"
- )
- except Exception as e:
- # Log error but continue trying to remove from DB
- logging.error(f"Error deleting file {hf_path_to_delete} from HF Hub: {e}")
- errors.append(f"Failed to delete file '{hf_path_to_delete.split('/')[-1]}' from storage.")
- elif deleted_hf_paths:
- logging.warning(f"HF_TOKEN_WRITE not set. Cannot delete {len(deleted_hf_paths)} associated files from HF.")
- errors.append("Could not delete files from storage (token missing). Removed from listing.")
-
-
- # Update the database: remove all marked items
- paths_to_remove = {item['path'] for item in items_to_remove_from_db}
- data['users'][username]['items'] = [item for item in user_items if item['path'] not in paths_to_remove]
+ user_files = data['users'][username]['files']
+ original_filename = "Неизвестный файл"
+ file_found_in_db = False
+ for file_info in user_files:
+ if file_info.get('hf_path') == hf_path:
+ original_filename = file_info.get('original_filename', original_filename)
+ file_found_in_db = True
+ break
+
+ if not file_found_in_db:
+ flash(f'Файл ({original_filename}) не найден в вашей базе данных!', 'warning')
+ # Optionally try deleting from HF anyway if out of sync? For now, just redirect.
+ return redirect(url_for('dashboard'))
try:
+ api = HfApi()
+ api.delete_file(
+ path_in_repo=hf_path,
+ repo_id=REPO_ID,
+ repo_type="dataset",
+ token=HF_TOKEN_WRITE,
+ commit_message=f"User {username} deleted file {original_filename}"
+ )
+ # Remove from database only after successful HF deletion
+ data['users'][username]['files'] = [f for f in user_files if f.get('hf_path') != hf_path]
save_data(data)
- if not errors:
- flash(f"'{item_to_delete.get('original_filename') or item_to_delete['name']}' deleted successfully.", 'success')
- else:
- flash(f"'{item_to_delete.get('original_filename') or item_to_delete['name']}' removed from listing, but some errors occurred.", 'warning')
- for error in errors:
- flash(error, 'error')
+ flash(f'Файл "{original_filename}" успешно удален!', 'success')
+ logging.info(f"User {username} deleted file {hf_path}")
+
except Exception as e:
- flash('Error saving changes after deletion.', 'error')
- logging.error(f"Failed to save data after deleting item for {username}: {e}")
- # Potentially revert DB changes in memory if save fails? Complex.
+ # Check if file not found error means it was already deleted
+ if "404" in str(e) or "not found" in str(e).lower():
+ logging.warning(f"File {hf_path} not found on HF during delete attempt by {username}, possibly already deleted. Removing from DB.")
+ data['users'][username]['files'] = [f for f in user_files if f.get('hf_path') != hf_path]
+ try:
+ save_data(data)
+ flash(f'Файл "{original_filename}" не найден в хранилище, удален из списка.', 'warning')
+ except Exception as save_e:
+ flash('Ошибка при обновлении базы данных после обнаружения отсутствующего файла.', 'error')
+ logging.error(f"Error saving data after failed delete (file not found): {save_e}")
- return redirect(url_for('dashboard', path=current_view_path))
+ else:
+ logging.error(f"Error deleting file {hf_path} for user {username}: {e}")
+ flash(f'Ошибка при удалении файла "{original_filename}"!', 'error')
+ return redirect(url_for('dashboard'))
@app.route('/logout')
def logout():
- username = session.pop('username', None)
- if username:
- logging.info(f"User {username} logged out.")
- flash('You have been logged out.', 'info')
- # Optional: Clear client-side storage via JS if needed, but session pop is key
+ session.pop('username', None)
+ flash('Вы успешно вышли из системы.', 'success')
+ # JS on login page handles localStorage clearing
return redirect(url_for('login'))
# --- Admin Routes ---
-ADMIN_USERNAME = os.getenv("ADMIN_USER", "admin")
-ADMIN_PASSWORD = os.getenv("ADMIN_PASS", "password") # Use env vars for real passwords!
+# Simple password protection for admin routes (replace with proper auth)
+ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "adminpass123")
-def is_admin():
- # Simple session check for admin status
+def check_admin():
+ """Checks if admin is logged in via session."""
return session.get('is_admin')
-@app.route('/admin/login', methods=['GET', 'POST'])
+@app.route('/admhosto/login', methods=['GET', 'POST'])
def admin_login():
if request.method == 'POST':
- username = request.form.get('username')
password = request.form.get('password')
- if username == ADMIN_USERNAME and password == ADMIN_PASSWORD:
+ if password == ADMIN_PASSWORD:
session['is_admin'] = True
- flash('Admin login successful.', 'success')
+ flash('Успешный вход в админ-панель.', 'success')
return redirect(url_for('admin_panel'))
else:
- flash('Invalid admin credentials.', 'error')
+ flash('Неверный пароль администратора.', 'error')
return redirect(url_for('admin_login'))
- # Simple login form for admin
+ if check_admin():
+ return redirect(url_for('admin_panel'))
+
html = f'''
-Admin Login
-'''
+
+
+
+ Вход в Админ-панель
+
+
+
+
+Admin Panel - Zeus Cloud
-'''
+ html = '''
+
+
+
+
+
+ Админ-панель - Zeus Cloud
+
+
+
+
+ ')
def admin_user_files(username):
- if not is_admin():
+ if not check_admin():
return redirect(url_for('admin_login'))
data = load_data()
if username not in data.get('users', {}):
- flash(f'User "{username}" not found.', 'error')
+ flash(f'Пользователь "{username}" не найден!', 'error')
return redirect(url_for('admin_panel'))
- user_data = data['users'][username]
- user_items = user_data.get('items', [])
- current_path = normalize_path(request.args.get('path', '/'))
-
- items_in_current_path, parent_path = get_items_in_path(user_items, current_path)
- breadcrumbs = get_breadcrumbs(current_path)
-
- html = f'''
-
-Files for {{ username }} - Admin
-
-
-
-
-
- {% if item.type == 'folder' %}
-
-
-
- {% elif item.file_type == 'image' %}
-  }})
- {% elif item.file_type == 'video' %}
-
- {% elif item.file_type == 'pdf' %}
-
- {% elif item.file_type == 'text' %}
-
- {% else %}
-
- {% endif %}
-
-
-
- - {% if item.type == 'folder' %} - {{ item.name }} - {% else %} - {{ item.original_filename }} - {% endif %} -
-{% if item.type == 'file' %} {{ item.upload_date }} {% else %} Folder {% endif %}
-
- {% if item.type == 'file' %}
- Download
- {% endif %}
-
+
-
-Logout
-
+
- {% endfor %}
- {% if not items_in_current_path %}
- Добавить на главный экран
+Для быстрого доступа к Zeus Cloud, вы можете добавить это приложение на главный экран вашего телефона:
+
+
+
+ Android (Chrome):
+-
+
- Откройте Zeus Cloud в браузере Chrome. +
- Нажмите на меню (три точки). +
- Выберите "Установить приложение" или "Добавить на главный экран". +
- Подтвердите добавление. +
+
iOS (Safari):
+-
+
- Откройте Zeus Cloud в браузере Safari. +
- Нажмите кнопку "Поделиться" (квадрат со стрелкой). +
- Прокрутите вниз и выберите "На экран «Домой»". +
- Нажмите "Добавить". +
This folder is empty.
- {% endif %} -
-
-
-
-
-'''
- return render_template_string(html, username=username, items_in_current_path=items_in_current_path,
- current_path=current_path, parent_path=parent_path, breadcrumbs=breadcrumbs,
- get_hf_item_url=get_hf_item_url, HF_TOKEN_READ=HF_TOKEN_READ) # Pass function and token to template
-
-@app.route('/create_folder', methods=['POST'])
-def create_folder():
- if 'username' not in session:
- return redirect(url_for('login'))
-
- username = session['username']
- data = load_data()
- if username not in data.get('users', {}):
- return redirect(url_for('login'))
-
- current_path = normalize_path(request.args.get('path', '/'))
- folder_name = request.form.get('folder_name', '').strip()
-
- if not folder_name:
- flash('Folder name cannot be empty.', 'error')
- return redirect(url_for('dashboard', path=current_path))
-
- # Basic validation for folder name
- if not folder_name.replace(' ','').isalnum() and '_' not in folder_name and '-' not in folder_name:
- flash('Folder name can only contain letters, numbers, spaces, underscores, and hyphens.', 'error')
- return redirect(url_for('dashboard', path=current_path))
-
- folder_name = secure_filename(folder_name) # Clean it up
- if not folder_name:
- flash('Invalid folder name.', 'error')
- return redirect(url_for('dashboard', path=current_path))
-
-
- new_folder_path = normalize_path(f"{current_path}/{folder_name}")
- user_items = data['users'][username].get('items', [])
-
- # Check if folder or file with the same path already exists
- if any(item['path'] == new_folder_path for item in user_items):
- flash(f"A folder or file named '{folder_name}' already exists here.", 'error')
- return redirect(url_for('dashboard', path=current_path))
+
+
+
- folder_info = {
- 'type': 'folder',
- 'path': new_folder_path,
- 'name': folder_name,
- 'created_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
- # 'hf_path' is not strictly needed for folders managed in JSON only
- }
- user_items.append(folder_info)
- data['users'][username]['items'] = user_items
+
+
+
+'''
+ # Pass username, files, repo_id and token (if needed) to template
+ return render_template_string(html, username=username, user_files=user_files, repo_id=REPO_ID, HF_TOKEN_READ=HF_TOKEN_READ)
-@app.route('/download/Admin Login
-{% with messages = get_flashed_messages(with_categories=true) %} - {% if messages %}{% for category, message in messages %}{{ message }}
{% endfor %}{% endif %}
-{% endwith %}
-
+
+'''
return render_template_string(html)
-@app.route('/admin/logout')
+@app.route('/admhosto/logout')
def admin_logout():
session.pop('is_admin', None)
- flash('Admin logged out.', 'info')
- return redirect(url_for('login'))
+ flash('Вы вышли из админ-панели.', 'info')
+ return redirect(url_for('admin_login'))
@app.route('/admhosto')
def admin_panel():
- if not is_admin():
+ if not check_admin():
return redirect(url_for('admin_login'))
data = load_data()
users = data.get('users', {})
- html = f'''
-
-Вход в Админ-панель
+ {% with messages = get_flashed_messages(with_categories=true) %} + {% if messages %}{% for category, message in messages %} +{{ message }}
+ {% endfor %}{% endif %}
+ {% endwith %}
+
+Admin Panel
-Admin Logout -User List
-{% for username, user_data in users.items() %}
-
-{% with messages = get_flashed_messages(with_categories=true) %}
- {% if messages %}
- {{{{ username }}}}
-
-{% endfor %}
-{% if not users %}Registered: {{ user_data.get('created_at', 'N/A') }}
-Items: {{ user_data.get('items', []) | length }}
- -No users registered yet.
{% endif %}
- {% for category, message in messages %}
{% endif %}
-{% endwith %}{{ message }}
{% endfor %}
-
+
+
+
+'''
return render_template_string(html, users=users)
@app.route('/admhosto/user/Админ-панель Zeus Cloud
+ Выйти + {% with messages = get_flashed_messages(with_categories=true) %} + {% if messages %}
+ {% for category, message in messages %}
+
{% endif %}
+ {% endwith %}
+ {{ message }}
+ {% endfor %}
+ Список пользователей ({{ users | length }})
+
+ {% for username, user_data in users.items() | sort %}
+
+
+ {{ username }}
+
+ {% endfor %}
+ {% if not users %}
+ Дата регистрации: {{ user_data.get('created_at', 'N/A') }}
+Количество файлов: {{ user_data.get('files', []) | length }}
+
+
+
+ Пользователей пока нет.
+ {% endif %} +
-
Files for User: {{ username }}
-Back to User List - -{% with messages = get_flashed_messages(with_categories=true) %} - {% if messages %}{% for category, message in messages %}{{ message }}
{% endfor %}{% endif %}
-{% endwith %}
-
-{% if parent_path is not none %}
-⬆️ Go Up
-{% endif %}
-
-
- {% for item in items_in_current_path %}
- ', methods=['POST'])
def admin_delete_user(username):
- if not is_admin():
- flash('Admin privileges required.', 'error')
- return redirect(url_for('admin_login'))
+ if not check_admin():
+ flash('Требуется авторизация администратора.', 'error')
+ return redirect(url_for('admin_login'))
+ if not HF_TOKEN_WRITE:
+ flash('Удаление невозможно: токен для записи на Hugging Face не настроен.', 'error')
+ return redirect(url_for('admin_panel'))
data = load_data()
if username not in data.get('users', {}):
- flash(f'User "{username}" not found.', 'error')
+ flash(f'Пользователь "{username}" не найден!', 'error')
return redirect(url_for('admin_panel'))
- logging.warning(f"ADMIN ACTION: Attempting to delete user {username} and all their files.")
- api = get_hf_api()
- hf_user_folder = f"cloud_files/{username}"
- delete_folder_error = False
+ user_files_to_delete = data['users'][username].get('files', [])
+ hf_paths_to_delete = [file['hf_path'] for file in user_files_to_delete if 'hf_path' in file]
+ folder_path = f"cloud_files/{username}"
- # Step 1: Delete user's folder from Hugging Face Hub
- if HF_TOKEN_WRITE:
+ try:
+ api = HfApi()
+ deleted_hf = False
+ # Try deleting the whole folder first (more efficient)
try:
- logging.info(f"Admin deleting HF folder: {hf_user_folder}")
+ logging.info(f"Admin attempting to delete folder: {folder_path}")
api.delete_folder(
- folder_path=hf_user_folder,
- repo_id=REPO_ID,
- repo_type="dataset",
- token=HF_TOKEN_WRITE,
- commit_message=f"Admin deleted user {username} and all files"
+ folder_path=folder_path,
+ repo_id=REPO_ID,
+ repo_type="dataset",
+ token=HF_TOKEN_WRITE,
+ commit_message=f"Admin deleted folder for user {username}"
)
- logging.info(f"Successfully deleted HF folder for user {username}.")
- except Exception as e:
- # Log error but continue to delete user from DB
- logging.error(f"Error deleting HF folder {hf_user_folder} for user {username}: {e}")
- # Check if it was "not found" error, which is okay if user had no uploads
- if "404" not in str(e) and "not found" not in str(e).lower():
- delete_folder_error = True
- flash(f"Warning: Could not completely delete storage folder for {username}. Check HF repo.", 'warning')
- else:
- logging.info(f"HF folder {hf_user_folder} likely did not exist or was already empty.")
-
- else:
- logging.warning("HF_TOKEN_WRITE not set. Cannot delete user's folder from HF.")
- flash("Warning: Cannot delete user files from storage (admin token missing).", 'warning')
- # Set flag to indicate potential orphaned files
- if data['users'][username].get('items'):
- delete_folder_error = True
-
- # Step 2: Delete user from the database
- del data['users'][username]
-
- try:
+ logging.info(f"Admin successfully deleted folder {folder_path}")
+ deleted_hf = True
+ except Exception as folder_del_err:
+ logging.warning(f"Admin failed to delete folder {folder_path} ({folder_del_err}). Attempting individual file deletion.")
+ # Fallback: delete files individually if folder deletion fails or isn't supported well
+ if hf_paths_to_delete:
+ delete_results = api.delete_files(
+ paths_in_repo=hf_paths_to_delete,
+ repo_id=REPO_ID,
+ repo_type="dataset",
+ token=HF_TOKEN_WRITE,
+ commit_message=f"Admin deleted files for user {username}",
+ # Consider adding ignore_patterns=None, ignore_regex=None if needed
+ )
+ # Note: delete_files might not raise errors for non-existent files, just logs them.
+ # We assume success if no exception is raised here.
+ logging.info(f"Admin deleted individual files for user {username}. Results (may vary): {delete_results}")
+ deleted_hf = True # Assume deletion attempted/succeeded for DB removal
+
+
+ # Delete user from database AFTER attempting HF deletion
+ del data['users'][username]
save_data(data)
- log_msg = f"Admin successfully deleted user {username}."
- if delete_folder_error:
- log_msg += " (Potential errors deleting storage folder)"
- logging.warning(log_msg) # Log as warning because it's a destructive admin action
- flash(f'User "{username}" has been deleted.', 'success' if not delete_folder_error else 'warning')
+ flash(f'Пользователь "{username}" и его файлы (попытка удаления из хранилища) успешно удалены из базы данных!', 'success')
+ logging.info(f"Admin deleted user {username} entry from database.")
+
except Exception as e:
- flash('Error saving changes after deleting user.', 'error')
- logging.error(f"Failed to save data after deleting user {username}: {e}")
- # If save fails, user is deleted in memory but not persisted - critical error
+ logging.error(f"Error during admin deletion of user {username}: {e}")
+ flash(f'Произошла ошибка при удалении пользователя "{username}"!', 'error')
return redirect(url_for('admin_panel'))
-@app.route('/admhosto/delete_item//', methods=['POST'])
-def admin_delete_item(username, item_path):
- if not is_admin():
- flash('Admin privileges required.', 'error')
- return redirect(url_for('admin_login'))
+@app.route('/admhosto/delete_file//', methods=['POST'])
+def admin_delete_file(username, hf_path):
+ if not check_admin():
+ flash('Требуется авторизация администратора.', 'error')
+ return redirect(url_for('admin_login'))
+ if not HF_TOKEN_WRITE:
+ flash('Удаление невозможно: токен для записи на Hugging Face не настроен.', 'error')
+ return redirect(url_for('admin_user_files', username=username))
data = load_data()
if username not in data.get('users', {}):
- flash(f'User "{username}" not found.', 'error')
+ flash(f'Пользователь "{username}" не найден!', 'error')
return redirect(url_for('admin_panel'))
- item_path_normalized = normalize_path(item_path)
- user_items = data['users'][username].get('items', [])
- item_to_delete = next((item for item in user_items if item['path'] == item_path_normalized), None)
-
- if not item_to_delete:
- flash('Item not found for this user.', 'error')
- referer = request.referrer or url_for('admin_user_files', username=username)
- return redirect(referer)
-
- current_view_path = '/'.join(item_path_normalized.split('/')[:-1]) or '/'
- logging.warning(f"ADMIN ACTION: Attempting deletion of item '{item_path_normalized}' for user {username}.")
+ user_files = data['users'][username].get('files', [])
+ original_filename = "Неизвестный файл"
+ file_found_in_db = False
+ for file_info in user_files:
+ if file_info.get('hf_path') == hf_path:
+ original_filename = file_info.get('original_filename', original_filename)
+ file_found_in_db = True
+ break
- api = get_hf_api()
- errors = []
- deleted_hf_paths = []
- items_to_remove_from_db = []
-
- if item_to_delete['type'] == 'file':
- items_to_remove_from_db.append(item_to_delete)
- hf_path = item_to_delete.get('hf_path')
- if hf_path:
- deleted_hf_paths.append(hf_path)
- else:
- logging.warning(f"Admin Delete: File item {item_path_normalized} missing hf_path.")
-
- elif item_to_delete['type'] == 'folder':
- items_to_remove_from_db.append(item_to_delete)
- folder_prefix = item_path_normalized + ('/' if item_path_normalized != '/' else '')
- children_to_delete = [item for item in user_items if item['path'].startswith(folder_prefix) and item['path'] != item_path_normalized]
-
- for child in children_to_delete:
- items_to_remove_from_db.append(child)
- if child['type'] == 'file' and child.get('hf_path'):
- deleted_hf_paths.append(child['hf_path'])
-
- folder_hf_base = f"cloud_files/{username}"
- relative_folder_path = item_path_normalized.lstrip('/')
- hf_folder_path_to_delete = f"{folder_hf_base}/{relative_folder_path}" if relative_folder_path else folder_hf_base
-
- if HF_TOKEN_WRITE:
- try:
- logging.info(f"Admin deleting HF folder: {hf_folder_path_to_delete}")
- api.delete_folder(folder_path=hf_folder_path_to_delete, repo_id=REPO_ID, repo_type="dataset", token=HF_TOKEN_WRITE, commit_message=f"Admin deleted folder {item_path_normalized} for {username}")
- logging.info(f"Admin successfully deleted HF folder: {hf_folder_path_to_delete}")
- except Exception as e:
- logging.error(f"Admin error deleting folder {hf_folder_path_to_delete} from HF: {e}.")
- if "404" not in str(e) and "not found" not in str(e).lower():
- errors.append(f"Could not fully remove storage folder '{item_to_delete['name']}'.")
- else:
- logging.warning("HF_TOKEN_WRITE not set. Cannot delete folder from HF.")
- if any(item['type'] == 'file' for item in children_to_delete):
- errors.append("Cannot delete storage folder (token missing).")
-
- # Delete individual files (if needed)
- if HF_TOKEN_WRITE:
- for hf_path_to_delete in deleted_hf_paths:
- try:
- logging.info(f"Admin deleting HF file: {hf_path_to_delete}")
- api.delete_file(path_in_repo=hf_path_to_delete, repo_id=REPO_ID, repo_type="dataset", token=HF_TOKEN_WRITE, commit_message=f"Admin deleted item {hf_path_to_delete} for {username}")
- except Exception as e:
- logging.error(f"Admin error deleting file {hf_path_to_delete} from HF: {e}")
- if "404" not in str(e) and "not found" not in str(e).lower():
- errors.append(f"Failed to delete file '{hf_path_to_delete.split('/')[-1]}' from storage.")
- elif deleted_hf_paths:
- logging.warning(f"HF_TOKEN_WRITE not set. Cannot delete {len(deleted_hf_paths)} associated files from HF.")
- errors.append("Cannot delete files from storage (token missing).")
-
- # Update DB
- paths_to_remove = {item['path'] for item in items_to_remove_from_db}
- data['users'][username]['items'] = [item for item in user_items if item['path'] not in paths_to_remove]
+ if not file_found_in_db:
+ flash(f'Файл ({original_filename}) не найден в базе данных пользователя "{username}"!', 'warning')
+ # Proceed with HF deletion attempt anyway in case DB is out of sync
try:
+ api = HfApi()
+ api.delete_file(
+ path_in_repo=hf_path,
+ repo_id=REPO_ID,
+ repo_type="dataset",
+ token=HF_TOKEN_WRITE,
+ commit_message=f"Admin deleted file {original_filename} for user {username}"
+ )
+
+ # Remove from DB after successful HF deletion (or if it wasn't in DB initially)
+ data['users'][username]['files'] = [f for f in user_files if f.get('hf_path') != hf_path]
save_data(data)
- item_display_name = item_to_delete.get('original_filename') or item_to_delete['name']
- if not errors:
- flash(f"Admin successfully deleted '{item_display_name}' for user {username}.", 'success')
- else:
- flash(f"Admin deleted '{item_display_name}' from listing, but errors occurred during storage cleanup.", 'warning')
- for error in errors:
- flash(error, 'error')
- logging.warning(f"Admin deleted item '{item_path_normalized}' for user {username}. Errors: {len(errors)}")
- except Exception as e:
- flash('Error saving changes after admin deletion.', 'error')
- logging.error(f"Failed to save data after admin deleting item for {username}: {e}")
+ flash(f'Файл "{original_filename}" пользователя "{username}" успешно удален!', 'success')
+ logging.info(f"Admin deleted file {hf_path} for user {username}")
- return redirect(url_for('admin_user_files', username=username, path=current_view_path))
+ except Exception as e:
+ # Check if file not found error means it was already deleted
+ if "404" in str(e) or "not found" in str(e).lower():
+ logging.warning(f"Admin: File {hf_path} not found on HF during delete attempt for {username}. Removing from DB.")
+ data['users'][username]['files'] = [f for f in user_files if f.get('hf_path') != hf_path]
+ try:
+ save_data(data)
+ flash(f'Файл "{original_filename}" не найден в хранилище, удален из списка пользователя "{username}".', 'warning')
+ except Exception as save_e:
+ flash('Ошибка при обновлении базы данных после обнаружения отсутствующего файла.', 'error')
+ logging.error(f"Admin: Error saving data after failed delete (file not found): {save_e}")
+ else:
+ logging.error(f"Admin error deleting file {hf_path} for user {username}: {e}")
+ flash(f'Ошибка при удалении файла "{original_filename}" пользователя "{username}"!', 'error')
+ return redirect(url_for('admin_user_files', username=username))
if __name__ == '__main__':
if not HF_TOKEN_WRITE:
- logging.warning("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
- logging.warning("! HF_TOKEN (write access) is NOT SET.")
- logging.warning("! File uploads, deletions, folder creations, and DB backups to HF Hub WILL FAIL.")
- logging.warning("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
+ logging.warning("HF_TOKEN (write access) is not set. File uploads and deletions will fail.")
if not HF_TOKEN_READ:
- logging.warning("HF_TOKEN_READ is not set. Using HF_TOKEN (write) for reads.")
- logging.warning("Downloads/previews MIGHT fail if HF_TOKEN is not set or repo is private.")
+ logging.warning("HF_TOKEN_READ is not set. Falling back to HF_TOKEN (if set). File downloads/previews might fail for private repos if HF_TOKEN is also not set.")
+ if not os.getenv("ADMIN_PASSWORD"):
+ logging.warning("ADMIN_PASSWORD environment variable not set. Using default weak password 'adminpass123'.")
- # Initial data load/creation on startup
- if not os.path.exists(DATA_FILE):
- logging.info(f"{DATA_FILE} not found, attempting initial download or creating empty.")
- try:
- load_data() # Try to download/initialize
- except Exception as e:
- logging.error(f"Critical error during initial data load: {e}. Starting with empty.", exc_info=True)
- with open(DATA_FILE, 'w', encoding='utf-8') as f:
- json.dump({'users': {}}, f)
+
+ # Initial data load and potential download from HF
+ logging.info("Performing initial data load...")
+ load_data()
+ logging.info("Initial data load complete.")
+ # Start periodic backup only if write token is available
if HF_TOKEN_WRITE:
backup_thread = threading.Thread(target=periodic_backup, daemon=True)
backup_thread.start()
logging.info("Periodic backup thread started.")
else:
- logging.warning("Periodic backup thread NOT started (HF_TOKEN_WRITE missing).")
+ logging.warning("Periodic backup disabled because HF_TOKEN (write access) is not set.")
+
+ # Use waitress or gunicorn in production instead of Flask's built-in server
+ # Example using Flask's server for development:
+ app.run(debug=False, host='0.0.0.0', port=int(os.getenv("PORT", 7860)))
- # Use Gunicorn or Waitress in production instead of app.run(debug=True)
- # Example: gunicorn --bind 0.0.0.0:7860 app:app
- app.run(debug=False, host='0.0.0.0', port=7860) # Debug=False is important for production
\ No newline at end of file
+# --- END OF FILE app (8).py ---
\ No newline at end of file
- Файлы пользователя {{ username }} - Zeus Cloud
+
+
+
+
+
-
-
- {% if item.type == 'folder' %}
-
-
-
- {% elif item.file_type == 'image' %}
-
- {% elif item.file_type == 'video' %}
-
- {% elif item.file_type == 'pdf' %}
-
- {% elif item.file_type == 'text' %}
-
- {% else %}
-
- {% endif %}
-
-
-
-
- {% if item.type == 'folder' %}
- {{ item.name }}
- {% else %}
- {{ item.original_filename }}
({{ item.name }})
- {% endif %}
-
Path: {{ item.path }}
-Created/Uploaded: {{ item.upload_date or item.created_at }}
-
- {% if item.type == 'file' %}
- Download
- {% endif %}
-
-
+ user_files = sorted(data['users'][username].get('files', []), key=lambda x: x.get('upload_date', ''), reverse=True)
+
+ # Generate HF URLs for previews
+ for file_info in user_files:
+ try:
+ file_info['preview_url'] = hf_hub_url(repo_id=REPO_ID, filename=file_info['hf_path'], repo_type='dataset')
+ file_info['modal_url'] = file_info['preview_url'] # Use same base URL
+ file_info['download_url'] = url_for('download_file', hf_path=file_info['hf_path'], original_filename=file_info['original_filename'])
+ file_info['admin_delete_url'] = url_for('admin_delete_file', username=username, hf_path=file_info['hf_path'])
+ except Exception as e:
+ logging.error(f"Admin: Error generating HF URL for {file_info.get('hf_path')}: {e}")
+ file_info['preview_url'] = '#'
+ file_info['modal_url'] = '#'
+ file_info['download_url'] = '#'
+ file_info['admin_delete_url'] = '#'
+
+
+ html = '''
+
+
+
+
+
+
+ ← Назад
+ Выйти
+
-Файлы пользователя: {{ username }}
+ {% with messages = get_flashed_messages(with_categories=true) %} + {% if messages %} + {% for category, message in messages %} +{{ message }}
+ {% endfor %}
+ {% endif %}
+ {% endwith %}
+
+ {% for file in user_files %}
+
- {% endfor %}
- {% if not items_in_current_path %}
-
+
+ {% endfor %}
+ {% if not user_files %}
+
+ {% if file['type'] == 'video' %}
+
+ {% elif file['type'] == 'image' %}
+
+ {% elif file['type'] == 'pdf' %}
+
+ {% elif file['type'] == 'text' %}
+
+ {% else %}
+
+ Нет предпросмотра для {{ file.type }}
+ {% endif %} +
+
+ {{ file.original_filename | truncate(30, True) }}
+{{ file.get('upload_date', 'N/A') }}
+ Скачать + +У этого пользователя пока нет файлов.
+ {% endif %}This folder is empty for user {{ username }}.
- {% endif %} -
-
-
-
-'''
- return render_template_string(html, username=username, items_in_current_path=items_in_current_path,
- current_path=current_path, parent_path=parent_path, breadcrumbs=breadcrumbs,
- get_hf_item_url=get_hf_item_url, HF_TOKEN_READ=HF_TOKEN_READ)
+
+
+
+
+
+
+'''
+ return render_template_string(html, username=username, user_files=user_files, repo_id=REPO_ID)
@app.route('/admhosto/delete_user/