USER
from flask_socketio import SocketIO
import os
import json
from passlib.context import CryptContext
from PIL import Image
from flask import Flask, render_template, url_for, redirect, request, send_from_directory, flash, session
from flask_login import LoginManager, login_user, current_user, logout_user, UserMixin
from flask_restful import Api
from flask_sqlalchemy import SQLAlchemy
from data import db_session
from data.videos import Video
from data.users import User
from sqlalchemy.orm import joinedload
from data.login_form import LoginForm
from data.register_form import RegisterForm
from data.video_form import VideoForm
from data.users_resource import UsersResource, UserListResource
from data.videos_resource import VideosResource, VideoListResource
from data.likes_resource import LikeResource, NotLikeResource
from data.subscriptions_resource import FollowResource, NotFollowResource
from werkzeug.utils import secure_filename
from data.avatar_form import AvatarForm
from data.comments import Comment
from data.comment_form import CommentForm
from data.music import Music
from werkzeug.utils import secure_filename
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy import func
from flask_migrate import Migrate
import uuid
import time
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime, timedelta
from datetime import datetime
app = Flask(__name__)
app.config['SECRET_KEY'] = 'yandexlyceum_secret_key'
api = Api(app)
socketio = SocketIO(app)
login_manager = LoginManager()
login_manager.init_app(app)
pwd_context = CryptContext(
schemes=["pbkdf2_sha256"],
default="pbkdf2_sha256",
pbkdf2_sha256__default_rounds=30000
)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///music.db'
db = SQLAlchemy(app)
migrate = Migrate(app, db)
# Set the folder to store avatars
AVATAR_UPLOAD_FOLDER = os.path.join(app.root_path, 'static', 'avatars')
# Allowed extensions
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'}
app.config['AVATAR_UPLOAD_FOLDER'] = AVATAR_UPLOAD_FOLDER
app.config['MAX_CONTENT_LENGTH'] = 300 * 1024 * 1024 # 2MB max upload sizes
app.config['UPLOAD_FOLDER'] = 'static/uploads'
app.config['COVER_FOLDER'] = 'static/covers'
app.config['ALLOWED_AUDIO_EXTENSIONS'] = {'mp3', 'wav', 'flac', 'ogg'}
app.config['ALLOWED_IMAGE_EXTENSIONS'] = {'png', 'jpg', 'jpeg', 'gif'}
# Создаем папки, если их нет
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
os.makedirs(app.config['COVER_FOLDER'], exist_ok=True)
import tkinter as tk
import psutil
import time
import threading
# Модель для треков
from functools import wraps
from flask import abort
import random
from flask import send_file
import magic
import subprocess
from werkzeug.utils import secure_filename
UPLOAD_FOLDERA = 'uploadsvids'
PROCESSED_FOLDERA = 'processedvids'
ALLOWED_EXTENSIONS = {'mp4', 'mkv', 'avi', 'mov', 'flv'}
app.config['UPLOAD_FOLDERA'] = UPLOAD_FOLDERA
app.config['PROCESSED_FOLDERA'] = PROCESSED_FOLDERA
# Ensure that directories exist
os.makedirs(UPLOAD_FOLDERA, exist_ok=True)
os.makedirs(PROCESSED_FOLDERA, exist_ok=True)
# Define allowed extensions
# Define allowed file extensions as a global variable
ALLOWED_EXTENSIONS = {'mp4', 'mkv', 'avi', 'mov', 'flv'}
def convert_to_mp4(input_path, output_path):
"""
Convert the input video file to MP4 format using FFmpeg.
"""
try:
subprocess.check_output([
'ffmpeg', '-i', input_path, '-c:v', 'libx264',
'-preset', 'fast', '-crf', '22',
'-c:a', 'aac', '-b:a', '192k', output_path
])
except subprocess.CalledProcessError as e:
print(f"Failed to convert video. Error: {e.output}")
raise
def allowed_file(filename):
"""
Check if the file extension is in the allowed list.
"""
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
@app.route('/mp4', methods=['GET', 'POST'])
def upload_mp4():
if request.method == 'POST':
if 'file' not in request.files:
return "No file part"
file = request.files['file']
if file.filename == '':
return "No selected file"
# Call the allowed_file function with the filename only
if file:
filename = secure_filename(file.filename)
file_path = os.path.join(app.config['UPLOAD_FOLDERA'], filename)
file.save(file_path)
mime = magic.Magic(mime=True)
mime_type = mime.from_file(file_path)
output_path = os.path.join(app.config['PROCESSED_FOLDERA'], f'{os.path.splitext(filename)[0]}.mp4')
if mime_type != 'video/mp4':
convert_to_mp4(file_path, output_path)
else:
convert_to_mp4(file_path, output_path)
return redirect(url_for('uploaded_mp4', filename=os.path.basename(output_path)))
return render_template('mp4.html')
@app.route('/tools', methods=['GET', 'POST'])
def tools():
params = {
'db_sess': db_sess,
'User': User,
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
return render_template('tools.html', **params)
@app.route('/mp4s/<filename>')
def uploaded_mp4(filename):
return send_from_directory(app.config['PROCESSED_FOLDERA'], filename)
UPLOAD_FOLDERI = 'static/uploadsimg'
CONVERTED_FOLDER = 'static/convertedimg'
os.makedirs(UPLOAD_FOLDERI, exist_ok=True)
os.makedirs(CONVERTED_FOLDER, exist_ok=True)
app.config['UPLOAD_FOLDERI'] = UPLOAD_FOLDERI
@app.route('/img2jpg', methods=['GET', 'POST'])
def convert_image():
converted_file_url = None
if request.method == 'POST':
if 'image' not in request.files:
return 'No file part', 400
file = request.files['image']
if file.filename == '':
return 'No selected file', 400
# Save the original file
filename = secure_filename(file.filename)
upload_path = os.path.join(app.config['UPLOAD_FOLDERI'], filename)
file.save(upload_path)
# Convert to JPG
base_filename = os.path.splitext(filename)[0]
jpg_filename = f"{base_filename}.jpg"
converted_path = os.path.join(CONVERTED_FOLDER, jpg_filename)
with Image.open(upload_path) as img:
rgb_im = img.convert('RGB') # Convert to RGB
rgb_im.save(converted_path, 'JPEG')
# Remove the original uploaded file
os.remove(upload_path)
# Get URL for downloading
converted_file_url = url_for('static', filename=f'convertedimg/{jpg_filename}')
return render_template('convert.html', converted_file_url=converted_file_url)
def admin_required(f):
@wraps(f)
def decorated_function(*args, **kwargs):
if not current_user.is_authenticated or not current_user.is_admin:
abort(403) # Запрещено
return f(*args, **kwargs)
return decorated_function
def owner_or_admin_required(f):
@wraps(f)
def decorated_function(*args, **kwargs):
track_id = kwargs.get('track_id')
track = Track.query.get_or_404(track_id)
if not current_user.is_authenticated:
abort(403)
return f(*args, **kwargs)
return decorated_function
@app.route('/delete_track/<int:track_id>', methods=['POST'])
def delete_track(track_id):
track = Track.query.get_or_404(track_id)
# Удаление аудио файла
audio_path = os.path.join(app.config['UPLOAD_FOLDER'], track.audio_filename)
if os.path.exists(audio_path):
os.remove(audio_path)
# Список файлов обложек по умолчанию
default_covers = ['default.png']
# Удаление обложки трека, если она не является обложкой по умолчанию
if track.cover_filename and track.cover_filename not in default_covers:
cover_path = os.path.join(app.config['COVER_FOLDER'], track.cover_filename)
if os.path.exists(cover_path):
os.remove(cover_path)
# Удаление записи трека из базы данных
db.session.delete(track)
db.session.commit()
flash('Трек успешно удалён.', 'success')
return redirect(url_for('tracks'))
class User(db.Model, UserMixin):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
login = db.Column(db.String(150), unique=True, nullable=False)
password = db.Column(db.String(150), nullable=False)
name = db.Column(db.String(150), nullable=False)
surname = db.Column(db.String(150), nullable=False)
is_admin = db.Column(db.Boolean, default=False)
is_tester = db.Column(db.Boolean, default=False) # Добавьте это поле
tracks = relationship('Track', backref='owner', lazy=True)
videos = relationship('Video', backref="author", lazy=True)
class Album(db.Model):
__tablename__ = 'albums'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(150), nullable=False)
artist = db.Column(db.String(150), nullable=False)
cover_filename = db.Column(db.String(150), nullable=False, default='default_album.png')
upload_date = db.Column(db.DateTime, default=datetime.utcnow)
tracks = relationship('Track', backref='album', lazy=True, cascade="all, delete-orphan")
class Track(db.Model):
__tablename__ = 'track'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(150), nullable=False)
album_id = db.Column(db.Integer, ForeignKey('albums.id'), nullable=True)
cover_filename = db.Column(db.String(150), nullable=False, default='default_track.png')
audio_filename = db.Column(db.String(150), nullable=False)
upload_date = db.Column(db.DateTime, default=datetime.utcnow)
likes = db.Column(db.Integer, default=0)
owner_id = db.Column(db.Integer, ForeignKey('users.id'), nullable=False)
# Связь с лайками с каскадным удалением
likes_details = relationship('Like', backref='track', lazy=True, cascade="all, delete, delete-orphan")
class Like(db.Model):
__tablename__ = 'likes'
id = db.Column(db.Integer, primary_key=True)
session_id = db.Column(db.String(150), nullable=False)
track_id = db.Column(db.Integer, ForeignKey('track.id'), nullable=False)
# Проверка разрешенных расширений файлов
def allowed_file(filename, allowed_extensions):
return '.' in filename and \
filename.rsplit('.', 1)[1].lower() in allowed_extensions
@app.route('/audiosearch')
def audiosearch():
query = request.args.get('search', '').strip()
if query:
search_pattern = f"%{query.lower()}%"
results = Track.query.join(Album).filter(
db.or_(
func.lower(Track.title).like(search_pattern),
func.lower(Album.title).like(search_pattern),
func.lower(Album.artist).like(search_pattern)
)
).all()
else:
results = []
return render_template('search_results.html', query=query, results=results)
# Генерация уникального идентификатора сессии
@app.before_request
def make_session_permanent():
if 'session_id' not in session:
session['session_id'] = str(uuid.uuid4())
# Страница загрузки трека
@app.route('/upload_audio', methods=['GET', 'POST'])
def upload():
if request.method == 'POST':
# Получаем данные формы
title = request.form.get('title')
audio_file = request.files.get('audio_file')
cover_file = request.files.get('cover_file')
album_id = request.form.get('album_id') # ID выбранного альбома
new_album_title = request.form.get('new_album_title') # Название нового альбома (если создается)
new_album_artist = request.form.get('new_album_artist') # Исполнитель нового альбома
new_album_cover = request.files.get('new_album_cover') # Обложка нового альбома
# Проверяем обязательные поля
if not title or not audio_file:
flash('Название и аудио файл обязательны для заполнения.', 'danger')
return redirect(url_for('upload'))
# Проверяем тип аудио файла
if not allowed_file(audio_file.filename, app.config['ALLOWED_AUDIO_EXTENSIONS']):
flash('Неверный тип аудио файла.', 'danger')
return redirect(url_for('upload'))
# Если создается новый альбом
if new_album_title and new_album_artist:
# Обработка обложки нового альбома
if new_album_cover and allowed_file(new_album_cover.filename, app.config['ALLOWED_IMAGE_EXTENSIONS']):
new_album_cover_filename = str(uuid.uuid4()) + '_' + secure_filename(new_album_cover.filename)
new_album_cover_path = os.path.join(app.config['COVER_FOLDER'], new_album_cover_filename)
new_album_cover.save(new_album_cover_path)
else:
new_album_cover_filename = 'default_album.png'
# Создание нового альбома
new_album = Album(title=new_album_title, artist=new_album_artist, cover_filename=new_album_cover_filename)
db.session.add(new_album)
db.session.commit()
album_id = new_album.id
flash('Новый альбом успешно создан!', 'success')
elif album_id:
album_id = int(album_id)
album = Album.query.get(album_id)
if not album:
flash('Выбранный альбом не существует.', 'danger')
return redirect(url_for('upload'))
else:
album_id = None # Трек не привязан к альбому
# Сохраняем аудио файл
audio_filename = str(uuid.uuid4()) + '_' + secure_filename(audio_file.filename)
audio_path = os.path.join(app.config['UPLOAD_FOLDER'], audio_filename)
audio_file.save(audio_path)
# Сохраняем обложку трека или используем стандартную
if cover_file and allowed_file(cover_file.filename, app.config['ALLOWED_IMAGE_EXTENSIONS']):
cover_filename = str(uuid.uuid4()) + '_' + secure_filename(cover_file.filename)
cover_path = os.path.join(app.config['COVER_FOLDER'], cover_filename)
cover_file.save(cover_path)
else:
cover_filename = 'default.png'
# Создаем новый трек в базе данных
new_track = Track(title=title, album_id=album_id, cover_filename=cover_filename, audio_filename=audio_filename, owner_id=current_user.id)
db.session.add(new_track)
db.session.commit()
flash('Трек успешно загружен!', 'success')
return redirect(url_for('track', track_id=new_track.id))
# Получаем все альбомы для отображения в выпадающем списке
albums = Album.query.order_by(Album.title).all()
return render_template('upload.html', albums=albums)
from flask import flash
# Маршрут для отображения всех альбомов
@app.route('/albums')
def albums():
all_albums = Album.query.order_by(Album.upload_date.desc()).all()
return render_template('albums.html', albums=all_albums)
# Маршрут для создания нового альбома
@app.route('/create_album', methods=['GET', 'POST'])
def create_album():
# Только авторизованные пользователи могут создавать альбомы
if request.method == 'POST':
title = request.form.get('title')
artist = request.form.get('artist')
cover_file = request.files.get('cover_file')
if not title or not artist:
flash('Название и исполнитель обязательны для заполнения.', 'danger')
return redirect(url_for('create_album'))
# Обработка обложки альбома
if cover_file and allowed_file(cover_file.filename, app.config['ALLOWED_IMAGE_EXTENSIONS']):
cover_filename = str(uuid.uuid4()) + '_' + secure_filename(cover_file.filename)
cover_path = os.path.join(app.config['COVER_FOLDER'], cover_filename)
cover_file.save(cover_path)
else:
cover_filename = 'default_album.png' # Используйте стандартную обложку
# Создание нового альбома
new_album = Album(title=title, artist=artist, cover_filename=cover_filename)
db.session.add(new_album)
db.session.commit()
flash('Альбом успешно создан!', 'success')
return redirect(url_for('albums'))
return render_template('create_album.html')
# Маршрут для просмотра отдельного альбома с его треками
@app.route('/album/<int:album_id>')
def view_album(album_id):
album = Album.query.get_or_404(album_id)
return render_template('view_album.html', album=album)
# Страница со списком всех треков
@app.route('/tracks')
def tracks():
all_albums = Album.query.order_by(Album.upload_date.desc()).limit(5).all()
all_tracks = Track.query.all()
params = {
'user': user,
'all_albums': all_albums
}
# Define a list of colors for default.png covers
color_map = ["#FFB6C1", "#FFD700", "#ADFF2F", "#87CEFA", "#FF69B4", "#98FB98", "#FF6347", "#8A2BE2"]
return render_template('tracks.html', tracks=all_tracks, color_map=color_map)
# Страница отдельного трека
@app.route('/track/<int:track_id>')
def track(track_id):
track = Track.query.get_or_404(track_id)
# Проверяем, лайкнул ли пользователь этот трек
session_id = session.get('session_id')
liked = Like.query.filter_by(session_id=session_id, track_id=track_id).first()
return render_template('track.html', track=track, liked=liked is not None)
# Обработка лайков
@app.route('/like_music/<int:track_id>', methods=['POST'])
def like(track_id):
track = Track.query.get_or_404(track_id)
session_id = session.get('session_id')
existing_like = Like.query.filter_by(session_id=session_id, track_id=track_id).first()
if not existing_like:
new_like = Like(session_id=session_id, track_id=track_id)
track.likes += 1
db.session.add(new_like)
db.session.commit()
return redirect(url_for('track', track_id=track_id))
# Страница с рекомендациями
@app.route('/recommendations')
def recommendations():
top_liked = Track.query.order_by(Track.likes.desc()).limit(5).all()
newest = Track.query.order_by(Track.upload_date.desc()).limit(5).all()
return render_template('recommendations.html', top_liked=top_liked, newest=newest)
# Отдача обложек и аудио файлов
#@app.route('/covers/<filename>')
#def cover(filename):
# return send_from_directory(app.config['COVER_FOLDER'], filename)
@app.route('/uploads/<filename>')
def uploaded_file(filename):
return send_from_directory(app.config['UPLOAD_FOLDER'], filename)
@app.route('/favicon.ico')
def favicon():
"""Возвращает кастомную иконку страницы в браузере"""
return send_from_directory(os.path.join(app.root_path, 'static/img'),
'vhs.ico', mimetype='image/vnd.microsoft.icon')
@socketio.on('comment')
def handle_comment_event(username, video_title):
message = f"{username} commented on your video '{video_title}'"
socketio.emit('notify', {'message': message})
def like_video(video_id):
# Assuming this function is called during a like action
video = db_sess.query(Video).get(video_id)
if video:
message = f"Someone liked your video '{video.title}'"
socketio.emit('notify', {'message': message}, room=video.author_id) # send to video author
def subscribe_to_user(user_id):
# Assuming this function is called during a subscribe action
user = db_sess.query(User).get(user_id)
if user:
message = "Someone subscribed to you!"
socketio.emit('notify', {'message': message}, room=user.id)
def load_user(user_id):
"""Loads the user by ID."""
return db_sess.query(User).get(user_id)
@app.route('/upload_avatar', methods=['GET', 'POST'])
@app.route('/upload_avatar', methods=['POST'])
def upload_avatar():
"""Allows authenticated users to upload and customize their avatar."""
form = AvatarForm()
if request.method == 'POST':
# Handling form submission and image saving
if 'avatar' in request.files:
avatar = request.files['avatar']
filename = secure_filename(avatar.filename)
file_ext = os.path.splitext(filename)[1].lower()
avatar_filename = f"user_{current_user.id}{file_ext}"
avatar_path = os.path.join(app.config['AVATAR_UPLOAD_FOLDER'], avatar_filename)
# Save and process the avatar
try:
with Image.open(avatar) as img:
img = img.convert('RGB')
img.thumbnail((300, 300))
img.save(avatar_path, optimize=True, quality=85)
# Update user's avatar in the database
current_user.avatar = avatar_filename
db_sess.commit()
flash("Аватар успешно обновлен!", "success")
return redirect(url_for('user', user_id=current_user.id))
except Exception as e:
flash('Загрузка не удалась. Пожалуйста, попробуйте снова.', 'danger')
return redirect(url_for('upload_avatar'))
return render_template('upload_avatar.html', form=form)
@app.errorhandler(403)
def forbidden(e):
"""Возвращает кастомную страницу ошибки 403"""
params = {
'title': 'Доступ запрещен',
'message': 'Похоже, у вас нет доступа к этой странице ¯\\_(ツ)_/¯',
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
return render_template('error1.html', **params), 403
def to_int(value):
try:
return int(value)
except ValueError:
return 0 # or handle as needed
app.jinja_env.filters['to_int'] = to_int
@app.errorhandler(404)
def page_not_found(e):
"""Возвращает кастомную страницу ошибки 404"""
params = {
'title': 'Страница не найдена',
'message': 'Похоже, мы не можем найти нужную вам страницу ¯\\_(ツ)_/¯',
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
return render_template('error2.html', **params), 404
@app.errorhandler(405)
def method_not_allowed(e):
"""Возвращает кастомную страницу ошибки 405"""
params = {
'title': 'Метод не разрешен',
'message': 'Похоже, этот метод не разрешен ¯\\_(ツ)_/¯',
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
return render_template('error3.html', **params), 405
@app.errorhandler(500)
def internal_server_error(e):
"""Возвращает кастомную страницу ошибки 500"""
params = {
'title': 'Ошибка сервера',
'message': 'Похоже, на сервере произошла непредвиденная ошибка ¯\\_(ツ)_/¯',
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
return render_template('error4.html', **params), 500
@app.route('/')
@app.route('/index')
def index():
"""Возвращает домашнюю страницу"""
all_videos = db_sess.query(Video).all()
if len(all_videos) > 10:
last_videos = all_videos[len(all_videos) - 10:][::-1]
else:
last_videos = all_videos[::-1]
best_videos = sorted(all_videos, key=lambda x: -x.likes)[:3]
underrated_videos = list(filter(lambda x: x.likes == 0, all_videos))
if len(underrated_videos) > 9:
underrated_videos = underrated_videos[len(underrated_videos) - 9:][::-1]
else:
underrated_videos = underrated_videos[::-1]
params = {
'title': 'Video Hosting Service',
'last_videos': last_videos,
'best_videos': best_videos,
'underrated_videos': underrated_videos,
'user': user,
'db_sess': db_sess,
'User': User,
'Video': Video,
'authenticated': current_user.is_authenticated,
'all_videos': all_videos,
'current_user': current_user
}
return render_template('index.html', **params)
@login_manager.user_loader
def load_user(user_id):
"""Возвращает пользователя"""
return db_sess.query(User).get(user_id)
@app.route('/login', methods=['GET', 'POST'])
def login():
"""Возвращает страницу авторизации и авторизует пользователя"""
if current_user.is_authenticated:
return forbidden('')
form = LoginForm()
params = {
'title': 'Авторизация',
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
if form.validate_on_submit():
user = db_sess.query(User).filter(User.login == form.login.data).first()
if user and pwd_context.verify(form.password.data, user.password):
login_user(user, remember=form.remember_me.data)
return redirect('/')
return render_template('login.html',
message='Неправильный логин или пароль',
form=form, **params)
return render_template('login.html', form=form, **params)
from flask import render_template, request, redirect, url_for, flash
from flask_login import login_required, current_user
@app.route('/edit_comment/<int:comment_id>', methods=['GET', 'POST'])
@login_required
def edit_comment(comment_id):
# Fetch the comment from the database using comment_id
comment = db_sess.query(Comment).filter(Comment.id == comment_id).first()
# Ensure the comment exists and the current_user is the author
if comment is None or comment.user_id != current_user.id:
flash('Comment not found or you do not have permission to edit it.', 'danger')
return redirect(url_for('video', video_id=comment.video_id)) # Redirect to the video's page
if request.method == 'POST':
new_content = request.form.get('content')
if new_content:
comment.content = new_content
db_sess.commit()
flash('Comment updated successfully!', 'success')
else:
flash('No content provided for update.', 'danger')
return redirect(url_for('video', video_id=comment.video_id))
return render_template('edit_comment.html', comment=comment)
@app.route('/register', methods=['GET', 'POST'])
def register():
"""Возвращает страницу регистрации и регистрирует пользователя"""
if current_user.is_authenticated:
return forbidden('')
form = RegisterForm()
params = {
'title': 'Регистрация',
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
if form.validate_on_submit():
user = db_sess.query(User).filter(User.login == form.login.data).all()
if user:
return render_template('register.html',
message='Пользователь с таким логином уже существует',
form=form, **params)
if form.password.data != form.repeat_password.data:
return render_template('register.html',
message='Пароли не совпадают',
form=form, **params)
user = User()
user.login = form.login.data
user.password = pwd_context.hash(form.password.data)
user.name = form.name.data
user.surname = form.surname.data
db_sess.add(user)
db_sess.commit()
login_user(user, remember=form.remember_me.data)
return redirect('/')
return render_template('register.html', form=form, **params)
from flask import jsonify, request, current_app
@app.route('/api/like_comment/<int:comment_id>', methods=['POST']) # Ensure the user is logged in
def like_comment(comment_id):
# Logic to handle liking a comment
comment = db_sess.query(Comment).filter_by(id=comment_id).first()
if not comment:
return jsonify({'success': False, 'message': 'Comment not found.'}), 404
# Example logic: increment the like count
comment.likes += 1
db_sess.commit()
return jsonify({'success': True, 'new_like_count': comment.likes})
@app.route('/logout')
def logout():
"""
Производит выход пользователя из системы, возвращает домашнюю страницу.
Если пользователь не авторизован, возвращает страницу ошибки 403
"""
if current_user.is_authenticated:
logout_user()
return redirect('/')
else:
return forbidden('')
@app.route('/api/pin_comment/<int:comment_id>', methods=['POST'])
def pin_comment(comment_id):
comment = db_sess.query(Comment).get(comment_id)
if not comment:
return {'error': 'Comment not found'}, 404
video = db_sess.query(Video).get(comment.video_id)
# Сбрасываем все закрепленные комментарии для данного видео
db_sess.query(Comment).filter(Comment.video_id == video.id).update({'pinned': False})
# Закрепляем выбранный комментарий
comment.pinned = True
db_sess.commit()
return {'success': True}
@app.route('/delete_comment/<int:comment_id>', methods=['POST'])
def delete_comment(comment_id):
comment = db_sess.query(Comment).get(comment_id)
if not comment:
flash('Комментарий не найден.', 'danger')
return redirect(request.referrer)
if not current_user.is_authenticated or current_user.id != comment.user_id:
flash('Вы не можете удалить этот комментарий.', 'danger')
return redirect(request.referrer)
db_sess.delete(comment)
db_sess.commit()
flash('Комментарий удален.', 'success')
return redirect(request.referrer)
from flask import abort
from data.video_views import VideoView
@app.route('/video/<int:video_id>', methods=['GET', 'POST'])
def video(video_id):
current_video = db_sess.query(Video).get(video_id)
if current_video is None:
return page_not_found('')
# Проверяем, разрешены ли комментарии для этого видео
comments_enabled = current_video.comments_enabled
# Получаем автора видео
author = db_sess.query(User).get(current_video.author)
all_videos = db_sess.query(Video).all()
# Проверяем, поставил ли текущий пользователь лайк и подписан ли
like = False
subscription = False
# Check if the user is logged in
if current_user.is_authenticated:
# Check if the user has already viewed the video
view_exists = db_sess.query(VideoView).filter_by(
user_id=current_user.id, video_id=video_id).first()
# If not, increment the view count and add a record to video_views
if not view_exists:
current_video.views += 1
new_view = VideoView(user_id=current_user.id, video_id=video_id)
db_sess.add(new_view)
db_sess.commit()
if current_user.is_authenticated:
likes = json.loads(current_user.likes)
subscriptions = json.loads(current_user.subscriptions)
if video_id in likes:
like = True
if current_video.author in subscriptions:
subscription = True
# Обрабатываем форму комментария
# Обрабатываем форму комментария только если комментарии включены
form = CommentForm()
if comments_enabled and form.validate_on_submit():
if current_user.is_authenticated and not current_user.is_blocked:
# Добавление комментария
comment = Comment(
content=form.content.data,
user_id=current_user.id,
video_id=video_id,
parent_id=form.parent_id.data or None
)
db_sess.add(comment)
db_sess.commit()
flash('Комментарий добавлен!', 'success')
return redirect(url_for('video', video_id=video_id))
else:
flash('Вы не можете оставить комментарий.', 'danger')
return redirect
# Получаем комментарии к видео
comments = db_sess.query(Comment).filter(Comment.video_id == video_id).options(
joinedload(Comment.user),
joinedload(Comment.parent).joinedload(Comment.user),
joinedload(Comment.replies).joinedload(Comment.user)
).order_by(Comment.datetime.desc()).all() # Изменено на desc() для сортировки в обратном порядке
# Создаем словарь комментариев по id для удобства доступа
comments_by_id = {comment.id: comment for comment in comments}
# Строим дерево комментариев
root_comments = []
for comment in comments:
if comment.parent_id is None:
root_comments.append(comment)
else:
parent = comments_by_id.get(comment.parent_id)
if parent:
if not hasattr(parent, 'replies_sorted'):
parent.replies_sorted = []
parent.replies_sorted.append(comment)
last_videos = all_videos[len(all_videos) - 10:][::-1]
params = {
'title': f"{current_video.title} - {author.name} {author.surname}",
'current_video': current_video,
'video_src': url_for("static", filename=f'vid/{video_id}.mp4'),
'author': author,
'like': like,
'subscription': subscription,
'db_sess': db_sess,
'comments': root_comments,
'User': User,
'authenticated': current_user.is_authenticated,
'current_user': current_user,
'avatar_url': url_for('static', filename=f'avatars/{author.avatar}'),
'comments': comments,
'last_videos': last_videos,
'form': form
}
return render_template('video.html', **params)
@app.route('/add_video', methods=['GET', 'POST'])
def add_video():
if not current_user.is_authenticated:
return forbidden('')
form = VideoForm()
params = {
'title': 'Добавление видео',
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
# Добавляем проверку на блокировку
if current_user.is_blocked:
flash('Ваш аккаунт заблокирован и вы не можете загружать видео.', 'danger')
return render_template('add_video.html', form=form, **params)
if form.validate_on_submit():
video_file = request.files['video']
if not video_file.filename.lower().endswith('.mp4'):
return render_template('add_video.html',
message='Данный формат видео не поддерживается. '
'Пожалуйста, загрузите видео в формате mp4.',
form=form, **params)
preview_file = request.files['preview']
if not preview_file.filename.lower().endswith('.jpg'):
return render_template('add_video.html',
message='Данный формат обложки не поддерживается. '
'Пожалуйста, загрузите изображение в формате jpg.',
form=form, **params)
video = Video()
video.title = form.title.data
video.description = '<br>'.join(form.description.data.split('\n'))
video.author = current_user.id
video.comments_enabled = form.comments_enabled.data # Сохраняем значение флажка
db_sess.add(video)
db_sess.commit()
video_file.save(os.path.join('static/vid', f'{video.id}.mp4'))
preview_file.save(os.path.join('static/pre', f'{video.id}.jpg'))
return redirect('/')
return render_template('add_video.html', form=form, **params)
@app.route('/edit_video/<int:video_id>', methods=['GET', 'POST'])
def edit_video(video_id):
"""
Возвращает страницу редактирования видео
если пользователь авторизован и является автором,
иначе доступ запрщен
"""
form = VideoForm()
video = db_sess.query(Video).get(video_id)
if not current_user.is_authenticated:
return forbidden('')
if not current_user.id != video.author:
return forbidden('')
params = {
'title': 'Редактирование видео',
'video': video,
'description': '\n'.join(video.description.split('<br>')),
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
if form.validate_on_submit():
video_file = request.files['video']
if not video_file.filename.lower().endswith('.mp4'):
return render_template('edit_video.html',
message='Данный формат видео не поддерживается',
form=form, **params)
preview_file = request.files['preview']
if not preview_file.filename.lower().endswith('.jpg'):
return render_template('edit_video.html',
message='Данный формат обложки не поддерживается',
form=form, **params)
video.title = form.title.data
video.description = '<br>'.join(form.description.data.split('\n'))
db_sess.commit()
video_file.save(os.path.join('static/vid', f'{video.id}.mp4'))
preview_file.save(os.path.join('static/pre', f'{video.id}.jpg'))
with Image.open(f'static/pre/{video.id}.jpg') as preview_file:
width, height = preview_file.size
if width >= height:
new_width = height * 16 // 9
delta = (width - new_width) // 2
preview_file = preview_file.crop((delta, 0, new_width + delta, height))
else:
new_height = width * 9 // 16
delta = (height - new_height) // 2
preview_file = preview_file.crop((0, delta, width, new_height + delta))
preview_file.save(f'static/pre/{video.id}.jpg')
return redirect('/')
return render_template('edit_video.html', form=form, **params)
@app.route('/delete_video/<int:video_id>')
def delete_video(video_id):
video = db_sess.query(Video).get(video_id)
if not video:
return page_not_found('Video not found.')
# No need for explicit comment deletion; should cascade
db_sess.delete(video)
db_sess.commit()
# Remove video file and preview
try:
os.remove(os.path.join('static/vid', f'{video.id}.mp4'))
os.remove(os.path.join('static/pre', f'{video.id}.jpg'))
except FileNotFoundError:
pass # Handle if the files do not exist
return redirect(f'/user/{current_user.id}')
@app.route('/user/<int:user_id>')
def user(user_id):
"""Возвращает страницу выбранного пользователя с его видео"""
user = db_sess.query(User).get(user_id)
if user is None:
return page_not_found('')
# Получаем все видео пользователя
videos = db_sess.query(Video).filter(Video.author == user_id).all()[::-1]
# Вычисляем количество видео
video_count = len(videos)
# Альтернативно, для оптимизации можно использовать:
# video_count = db_sess.query(Video).filter(Video.author == user_id).count()
subscription = False
if current_user.is_authenticated:
subscriptions = json.loads(current_user.subscriptions)
if user_id in subscriptions:
subscription = True
print(f"Is user {user_id} subscribed by current user? {'Yes' if subscription else 'No'}")
# Пагинация видео (разбивка на группы по 3 видео)
paginated_videos = [videos[i * 3:i * 3 + 3] for i in range(len(videos) // 3)] + [videos[(len(videos) // 3) * 3:]]
# Добавляем все необходимые параметры, включая video_count
params = {
'title': f'{user.name} {user.surname}',
'user': user,
'videos': paginated_videos,
'subscription': subscription,
'empty': True if len(videos) == 0 else False,
'authenticated': current_user.is_authenticated,
'current_user': current_user,
'video_count': video_count # Передаём количество видео
}
return render_template('user.html', **params)
@app.route('/delete_user', methods=['GET', 'POST'])
def delete_user():
"""Handles user deletion with confirmation."""
if not current_user.is_authenticated:
return forbidden('')
if request.method == 'POST':
# Check for confirmation
if 'confirm_delete' in request.form:
# Delete user's videos (assuming you want to cascade deletion)
# Delete user's comments
for comment in current_user.comments:
db_sess.delete(comment)
# Delete the user
db_sess.delete(current_user)
db_sess.commit()
flash('Your account has been deleted.', 'success')
return redirect(url_for('index')) # Redirect to homepage
else:
flash('Deletion cancelled.', 'info')
return redirect(url_for('settings')) # Redirect back to settings
# GET request - show confirmation page
return render_template('delete_user.html', title='Delete Account')
@app.route('/settings')
def settings():
if not current_user.is_authenticated:
return forbidden('')
params = {
'title': f'Настройки',
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
return render_template('settings.html', **params)
def decline_subscription(number):
"""
Возвращает правильную форму слова "подписчик" в зависимости от числа.
Например:
1 -> подписчик
2 -> подписчика
5 -> подписчиков
"""
number = abs(number) # Убираем знак, если число отрицательное
last_two = number % 100
last = number % 10
if 11 <= last_two <= 14:
return "подписчиков"
if last == 1:
return "подписчик"
if 2 <= last <= 4:
return "подписчика"
return "подписчиков"
def decline_views(number):
"""
Возвращает правильную форму слова "подписчик" в зависимости от числа.
Например:
1 -> подписчик
2 -> подписчика
5 -> подписчиков
"""
number = abs(number) # Убираем знак, если число отрицательное
last_two = number % 100
last = number % 10
if 11 <= last_two <= 14:
return "просмотров"
if last == 1:
return "просмотр"
if 2 <= last <= 4:
return "просмотра"
return "посмотров"
# Регистрация фильтра в Jinja2
app.jinja_env.filters['decline_subscription'] = decline_subscription
app.jinja_env.filters['decline_views'] = decline_views
from data.users import User
from sqlalchemy import update
@app.route('/admin/users', methods=['GET', 'POST'])
def admin_users():
"""Административная страница для управления флажками, блокировками и статусом админа пользователей"""
if not current_user.is_authenticated or not current_user.is_admin:
return forbidden('') # Только администратор имеет доступ
users = db_sess.query(User).all()
if request.method == 'POST':
# Обработка изменений флажков, блокировок и статуса админа
for user in users:
if user.id == 1:
continue # Не изменять статус администратора
# Обработка флажков is_checked
is_checked = request.form.get(f'checked_{user.id}', 'off') == 'on'
if user.is_checked != is_checked:
user.is_checked = is_checked
db_sess.add(user)
# Обработка блокировок is_blocked
is_blocked = request.form.get(f'blocked_{user.id}', 'off') == 'on'
if user.is_blocked != is_blocked:
user.is_blocked = is_blocked
db_sess.add(user)
is_tester = request.form.get(f'is_tester_{user.id}', 'off') == 'on'
if user.is_tester != is_tester:
user.is_tester = is_tester
db_sess.add(user)
# Обработка статуса администратора is_admin
is_admin = request.form.get(f'is_admin_{user.id}', 'off') == 'on'
if user.is_admin != is_admin:
user.is_admin = is_admin
db_sess.add(user)
db_sess.commit()
flash('Изменения успешно сохранены.', 'success')
return redirect(url_for('admin_users'))
params = {
'title': 'Администрирование пользователей',
'users': users,
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
return render_template('admin_users.html', **params)
@app.route('/feed')
def feed():
"""
Возвращает страницу с лентой видео от людей,
на которых подписан пользователь.
Если пользователь не авторизован, доступ запрещен
"""
if not current_user.is_authenticated:
return forbidden('')
subscriptions = json.loads(current_user.subscriptions)
videos = []
for i in subscriptions:
videos += db_sess.query(Video).filter(Video.author == i).all()[::-1][:12]
new_videos = sorted(videos, key=lambda x: x.datetime)[::-1]
new_videos = [
{
'video': i,
'author': db_sess.query(User).get(i.author)
}
for i in new_videos
]
params = {
'title': f'Лента',
'new_videos': [new_videos[i * 3:i * 3 + 3] for i in range(len(new_videos) // 3)] +
[new_videos[(len(new_videos) // 3) * 3:]],
'empty': True if len(new_videos) == 0 else False,
'subscripted': True if len(subscriptions) != 0 else False,
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
return render_template('feed.html', **params)
@app.route('/favorite')
def favorite():
"""
Возвращает страницу с любимыми видео пользователя.
Если пользователь не авторизован, доступ запрещен
"""
if not current_user.is_authenticated:
return forbidden('')
likes = json.loads(current_user.likes)[::-1]
liked_videos = [
{
'video': video,
'author': db_sess.query(User).get(video.author) if video.author else None
}
for i in likes
if (video := db_sess.query(Video).get(i)) is not None
]
params = {
'title': f'Любимое',
'liked_videos': [liked_videos[i * 3:i * 3 + 3] for i in range(len(liked_videos) // 3)] +
[liked_videos[(len(liked_videos) // 3) * 3:]],
'empty': True if len(liked_videos) == 0 else False,
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
return render_template('favorite.html', **params)
@app.route('/people')
def people():
"""
Возвращает страницу с пользователями, на которых подписан пользователь.
Если пользователь не авторизован, доступ запрещен
"""
if not current_user.is_authenticated:
return forbidden('')
subscriptions = json.loads(current_user.subscriptions)
subscripted_users = [
{
'user': db_sess.query(User).get(i)
}
for i in subscriptions
]
subscripted_users.sort(key=lambda x: x['user'].name + x['user'].surname + x['user'].login)
params = {
'title': f'Люди',
'subscripted_users': subscripted_users,
'empty': True if len(subscripted_users) == 0 else False,
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
return render_template('people.html', **params)
from data.edit_profile_form import EditProfileForm # Импортируем форму
@app.route('/edit_profile', methods=['GET', 'POST'])
def edit_profile():
if not current_user.is_authenticated:
return forbidden('')
form = EditProfileForm(obj=current_user)
params = {
'title': 'Редактирование профиля',
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
if form.validate_on_submit():
if form.login.data and form.login.data != current_user.login:
existing_user = db_sess.query(User).filter(User.login == form.login.data).first()
if existing_user:
flash('Пользователь с таким логином уже существует.', 'danger')
return render_template('edit_profile.html', form=form, **params)
if form.name.data:
current_user.name = form.name.data
if form.surname.data:
current_user.surname = form.surname.data
if form.password.data:
current_user.password = pwd_context.hash(form.password.data)
current_user.channel_description = form.channel_description.data
db_sess.commit()
flash('Профиль обновлён успешно!', 'success')
return redirect(url_for('user', user_id=current_user.id))
return render_template('edit_profile.html', form=form, **params)
@app.route('/api/comment_like/<int:comment_id>', methods=['POST'])
def comment_like(comment_id):
if not current_user.is_authenticated:
return {'error': 'Unauthorized'}, 401
comment = db_sess.query(Comment).get(comment_id)
if not comment:
return {'error': 'Comment not found'}, 404
likes = json.loads(current_user.comment_likes)
if comment_id in likes:
return {'error': 'Already liked'}, 400
likes.append(comment_id)
current_user.comment_likes = json.dumps(likes)
comment.likes += 1
db_sess.commit()
return {'success': True, 'likes': comment.likes}
@app.route('/api/comment_unlike/<int:comment_id>', methods=['POST'])
def comment_unlike(comment_id):
if not current_user.is_authenticated:
return {'error': 'Unauthorized'}, 401
comment = db_sess.query(Comment).get(comment_id)
if not comment:
return {'error': 'Comment not found'}, 404
likes = json.loads(current_user.comment_likes)
if comment_id not in likes:
return {'error': 'Not liked'}, 400
likes.remove(comment_id)
current_user.comment_likes = json.dumps(likes)
comment.likes -= 1
db_sess.commit()
return {'success': True, 'likes': comment.likes}
@app.route('/about')
def about():
params = {
'title': f'Про сайт'
}
return render_template('about.html', **params)
from flask import request, render_template
from sqlalchemy import or_
from flask import request, render_template
from sqlalchemy import or_
@app.route('/search')
def search():
query = request.args.get('search', '').strip()
if query:
# Поиск по названию трека или альбома
search_pattern = f"%{query}%"
# Используем JOIN для связи треков с альбомами
results = Track.query.join(Album).filter(
db.or_(
Track.title.ilike(search_pattern),
Album.title.ilike(search_pattern),
Album.artist.ilike(search_pattern)
)
).all()
else:
results = []
return render_template('search_results.html', tracks=results, query=query)
@app.route('/users')
def users():
# Query all users from the database
all_users = db_sess.query(User).all()
# Set up template parameters
params = {
'title': 'All Users',
'users': all_users,
'authenticated': current_user.is_authenticated,
'current_user': current_user
}
# Render the template with the list of users
return render_template('users.html', **params)
from data.db_session import SqlAlchemyBase
from flask_socketio import SocketIO
db_session.global_init("db/hosting.sql")
from data import __all_models
db_sess = db_session.create_session()
SqlAlchemyBase.metadata.create_all(bind=db_sess.bind)
db_sess.rollback
SqlAlchemyBase.metadata.create_all(bind=db_sess.bind)
api.add_resource(UsersResource, '/api/users/<int:user_id>')
api.add_resource(UserListResource, '/api/users')
api.add_resource(VideosResource, '/api/videos/<int:video_id>')
api.add_resource(VideoListResource, '/api/videos')
api.add_resource(LikeResource, '/api/like/<int:video_id>')
api.add_resource(NotLikeResource, '/api/not_like/<int:video_id>')
api.add_resource(FollowResource, '/api/follow/<int:user_id>')
api.add_resource(NotFollowResource, '/api/not_follow/<int:user_id>')
if __name__ == '__main__':
from waitress import serve
app.run(port=8080, host='0.0.0.0')
это мой код добавь эту функцию на всех рутах где ее нет:
params = {
'db_sess': db_sess,
'User': User,
'authenticated': current_user.is_authenticated,
'current_user': current_user
}