I have developed a Python script that allows users to upload videos from YouTube to LBRY. The script also performs video post-processing and database operations. I would like to request a code review to get feedback on the code quality and potential improvements. [The Github Repo - v0.1.0]
I have only used sqlite and have not yet tested with PostgreSQL or MySQL. My tests are not as comprehensive as I would like them to be, yet. I would like general feedback on how I could improve the code and any shortcomings of the current code. I especially would like feedback on the database design & database operations and how they could be better. I ultimately would like to use this as part of a webapp, maybe using flask, where users can put in a link to their videos and it will be uploaded to LBRY. Eventually I would like to have entire YouTube channels transfer and playlist transfers as features, and even support for other video platforms.
Directory Tree:
vid2lbry/
├── database
│ ├── database_operations.py
│ ├── __init__.py
│ ├── models.py
│ └── youtube_lbry.db
├── download_archive
├── downloader
│ ├── __init__.py
│ └── youtube.py
├── pybry
├── tests
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_config.ini
│ └── test_database_operations.py
├── uploader
│ ├── __init__.py
│ └── lbry.py
├── app.log
├── config.ini
├── copy-config.ini
├── __init__.py
├── LICENSE
├── main.py
├── README.md
└── requirements.txt
config.ini:
[Database]
dialect = sqlite
[sqlite]
db_path = database/youtube_lbry.db
[mysql]
host = localhost
port = 3306
user = your_mysql_user
password = your_mysql_password
name = youtube_lbry
[postgresql]
host = localhost
port = 5432
user = your_postgresql_user
password = your_postgresql_password
name = youtube_lbry
[LBRY]
channel_id = {Put your ID here}
[Downloader]
simulate = False
download_archive_dir = download_archive
download_archive = download_archive.txt
format = be
stvideo[ext=mp4]+bestaudio
outtmpl = download_archive/%%(id)s.%%(ext)s
postprocessors_key = FFmpegVideoConvertor
preferedformat = mp4
database/models.py:
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import UniqueConstraint
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
class Video(Base):
__tablename__ = 'videos'
id = Column(Integer, primary_key=True)
video_source = Column(String)
video_id = Column(String, unique=True)
title = Column(String)
description = Column(String)
video_url = Column(String)
thumbnail_url = Column(String)
downloads = relationship('Download', primaryjoin="Video.video_id==Download.video_id")
uploads = relationship('LBRYUpload', primaryjoin="Video.video_id==LBRYUpload.video_id")
class Download(Base):
__tablename__ = 'downloads'
id = Column(Integer, primary_key=True)
video_source = Column(String)
video_id = Column(String, ForeignKey('videos.video_id'))
download_path = Column(String)
video = relationship(
'Video',
back_populates='downloads'
)
class LBRYUpload(Base):
__tablename__ = 'lbryuploads'
id = Column(Integer, primary_key=True)
video_source = Column(String)
video_id = Column(String, ForeignKey('videos.video_id'))
lbry_status = Column(Integer)
lbry_url = Column(String)
video = relationship(
'Video',
back_populates='uploads'
)
database/database_operations.py:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from database.models import Base
from database.models import Video
from database.models import Download
from database.models import LBRYUpload
Session = sessionmaker()
def get_database_engine(config):
"""
Get the database engine based on the configuration.
Args:
config (ConfigParser): The configuration object.
Returns:
sqlalchemy.engine.Engine: The SQLAlchemy database engine.
"""
db_type = config.get('Database', 'dialect')
if db_type == 'sqlite':
db_path = config.get('sqlite', 'db_path')
engine = create_engine(f'sqlite:///{db_path}', echo=True)
elif db_type == 'mysql':
# untested
db_host = config.get('mysql', 'host')
db_port = config.getint('mysql', 'port')
db_user = config.get('mysql', 'user')
db_password = config.get('mysql', 'password')
db_name = config.get('mysql', 'name')
engine = create_engine(f'mysql:///{db_user}:{db_password}@{db_host}:{db_port}/{db_name}', echo=True)
elif db_type == 'postgresql':
# untested
db_host = config.get('postgresql', 'host')
db_port = config.getint('postgresql', 'port')
db_user = config.get('postgresql', 'user')
db_password = config.get('postgresql', 'password')
db_name = config.get('postgresql', 'name')
engine = create_engine(f'postgresql:///{db_user}:{db_password}@{db_host}:{db_port}/{db_name}', echo=True)
else:
raise ValueError(f"Invalid database type: {db_type}")
return engine
def initialize_database(engine):
"""
Create all database tables if they don't exist.
Args:
engine (sqlalchemy.engine.Engine): The SQLAlchemy database engine.
"""
Session.configure(bind=engine)
Base.metadata.create_all(engine)
def create_video_record(
session,
video_info: dict,
video_url: str
):
"""
Create a new Video record in the database.
Args:
session (sqlalchemy.orm.Session): The SQLAlchemy database session.
video_info (dict): The video information dictionary.
youtube_url (str): The URL of the YouTube video.
"""
try:
video = Video(
video_source=video_info['video_source'],
video_id=video_info['id'],
title=video_info['title'],
description=video_info['description'],
video_url=video_url,
thumbnail_url=video_info['thumbnail']
)
session.merge(video)
except Exception as e:
session.rollback()
raise e
def create_download_record(session, video_info: dict):
"""
Create a new Download record in the database.
Args:
session (sqlalchemy.orm.Session): The SQLAlchemy database session.
video_info (dict): The video information dictionary.
"""
try:
download = Download(
video_source=video_info['video_source'],
video_id=video_info['id'],
download_path=video_info['download_path']
)
session.add(download)
except Exception as e:
session.rollback()
raise e
def create_lbry_upload_record(session, video_info: dict):
"""
Create a new LBRYUpload record in the database.
Args:
session (sqlalchemy.orm.Session): The SQLAlchemy database session.
video_info (dict): The video information dictionary.
"""
try:
lbry_upload = LBRYUpload(
video_source=video_info['video_source'],
video_id=video_info['id'],
lbry_status=0,
lbry_url=None
)
session.add(lbry_upload)
except Exception as e:
session.rollback()
raise e
def update_lbry_upload_record(
session,
video_id: str,
lbry_url: str,
lbry_status: int
):
"""
Update the LBRYUpload record in the database.
Args:
session (sqlalchemy.orm.Session): The SQLAlchemy database session.
video_id (str): The ID of the video.
lbry_url (str): The LBRY URL of the uploaded video.
lbry_status (int): The status of the LBRY upload (-1: Failed, 0: Pending, 1: Success).
"""
try:
lbry_upload = session.query(LBRYUpload).filter(LBRYUpload.video_id == video_id).one()
lbry_upload.lbry_url = lbry_url
lbry_upload.lbry_status = lbry_status
except Exception as e:
session.rollback()
raise e
downloader/youtube.py:
from __future__ import unicode_literals
import os
import re
import subprocess
from youtube_dl import YoutubeDL
class YouTubeProcessor:
def __init__(self, download_archive, youtube_url, ydl_opts):
"""
Initialize the YouTubeProcessor.
Args:
download_archive (str): Path to the download archive directory.
youtube_url (str): The URL of the YouTube video to process.
ydl_opts (dict): Options for youtube-dl.
"""
self.download_archive = download_archive
self.video_source = 'youtube'
self.youtube_url = youtube_url
self.ydl_opts = ydl_opts
def is_valid_youtube_url(self):
"""
Check if the provided URL is a valid YouTube URL.
Returns:
bool: True if the URL is valid, False otherwise.
"""
pattern = r'^(https?\:\/\/)?(www\.)?(youtube\.com|youtu\.?be)\/.+$'
return bool(re.match(pattern, self.youtube_url))
def process_video(self):
"""
Process the YouTube video by downloading and post-processing.
Returns:
dict: The video information dictionary.
"""
if not self.is_valid_youtube_url():
raise ValueError("Invalid YouTube URL: Please provide a valid YouTube URL.")
with YoutubeDL(self.ydl_opts) as ydl:
video_info = ydl.extract_info(
self.youtube_url, download=True
)
video_info['video_source'] = self.video_source
# Check if the downloaded file is already in MP4 format
mp4_filename = os.path.join(
self.download_archive,
f"{video_info['id']}.mp4"
)
if os.path.exists(mp4_filename):
# Set the downloaded file path to the MP4 file directly
video_info['download_path'] = mp4_filename
else:
# Perform post-processing using ffmpeg to convert to MP4 format
original_filename = os.path.join(
self.download_archive,
f"{video_info['id']}.{video_info['ext']}"
)
mp4_filename = os.path.join(
self.download_archive,
f"{video_info['id']}.mp4"
)
subprocess.run(
[
'ffmpeg',
'-i',
original_filename,
'-c',
'copy',
mp4_filename
],
check=True
)
# Set the downloaded file path to the MP4 file
video_info['download_path'] = mp4_filename
return video_info
uploader/lbry.py:
from pybry import LbrydApi
class LBRYUploader:
def __init__(self):
self.lbry = LbrydApi()
def upload(self, video_info):
"""
Upload the video to LBRY.
Args:
video_info (dict): The video information dictionary.
Returns:
str: The LBRY URL of the uploaded video, or None if the upload failed.
"""
try:
lbry_results = self.lbry.publish(
name=video_info['id'], preview=False, blocking=True, bid='0.001',
file_path=video_info['download_path'],
channel_id="68dad212fddb3d6cb294278e11a303a6a98f2905",
thumbnail_url=video_info['thumbnail'],
title=video_info['title'], description=video_info['description']
)
return lbry_results[0]['outputs'][0]['permanent_url']
except Exception as e:
print(f"Error uploading video to LBRY: {e}")
if hasattr(e, 'response'):
print(f"LBRY API Response: {e.response}")
return None
main.py:
from __future__ import unicode_literals
import os
import logging
import configparser
import argparse
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from database.database_operations import get_database_engine
from database.database_operations import initialize_database
from database.database_operations import create_video_record
from database.database_operations import create_download_record
from database.database_operations import create_lbry_upload_record
from database.database_operations import update_lbry_upload_record
from downloader.youtube import YouTubeProcessor
from uploader.lbry import LBRYUploader
# Set up logging
logging.basicConfig(
filename='app.log',
level=logging.ERROR,
format='%(asctime)s - %(levelname)s - %(message)s'
)
def parse_arguments():
parser = argparse.ArgumentParser(description='Process a YouTube video and upload it to LBRY.')
parser.add_argument('youtube_url', type=str, help='URL of the YouTube video to process and upload to LBRY.')
return parser.parse_args()
def read_config(config_file):
"""
Read and parse the configuration file.
Args:
config_file (str): Path to the configuration file.
Returns:
configparser.ConfigParser: The parsed configuration object.
"""
config = configparser.ConfigParser()
config.read(config_file)
return config
def main():
# Read the configuration file
config = read_config('config.ini')
download_archive = config.get('Downloader', 'download_archive')
download_archive_dir = os.path.abspath(
config.get('Downloader', 'download_archive_dir')
)
# Read settings from the configuration file
channel_id = config.get('LBRY', 'channel_id')
ydl_opts = {
'simulate': config.getboolean('Downloader', 'simulate'),
'download_archive': os.path.join(download_archive_dir, download_archive),
'format': config.get('Downloader', 'format'),
'outtmpl': config.get('Downloader', 'outtmpl'),
}
args = parse_arguments()
url = args.youtube_url
try:
# Video processing
youtube_processor = YouTubeProcessor(download_archive_dir, url, ydl_opts)
video_info = youtube_processor.process_video()
# Database operations
engine = get_database_engine(config)
Session = sessionmaker(bind=engine)
with Session() as session:
# Create tables if they don't exist
initialize_database(engine)
# Perform database operations within a transaction block
session.begin()
create_video_record(session, video_info, url)
create_download_record(session, video_info)
create_lbry_upload_record(session, video_info)
session.commit()
# LBRY upload
lbry_uploader = LBRYUploader()
lbry_url = lbry_uploader.upload(video_info)
# Update LBRYUpload record with the LBRY URL and status
if lbry_url:
print(f"LBRY URL: {lbry_url}")
# Set status to success
update_lbry_upload_record(
session,
video_info['id'],
lbry_url,
lbry_status=1
)
session.commit()
else:
print("LBRY upload failed.")
# Set status to failed
update_lbry_upload_record(
session,
video_info['id'],
lbry_url=None,
lbry_status=-1
)
session.commit()
except Exception as e:
logging.exception(f"An unexpected error occurred: {e}")
print(f"An unexpected error occured: {e}")
if __name__ == "__main__":
main()
So far, I only have tests for the database operations functions.
tests/conftests.py:
import pytest
from configparser import ConfigParser
def load_test_config():
config = ConfigParser()
config.read("tests/test_config.ini")
return config
def pytest_addoption(parser):
config = load_test_config()
db_choices = list(config.sections())[1:]
db_choices_str = ", ".join(db_choices)
parser.addoption(
"--db-type",
action="store",
default="sqlite",
choices=db_choices,
help=f"Specify the database type to use for testing. Available choices: {db_choices_str} (default: sqlite).",
)
tests/test_database_operations.py:
from configparser import ConfigParser
import pytest
import os
import sys
# Get the parent directory of the current directory (vid2lbry directory)
parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# Add the parent directory to the Python module search path
sys.path.insert(0, parent_dir)
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from database.models import Base
from database.models import Video
from database.models import Download
from database.models import LBRYUpload
from database.database_operations import get_database_engine
from database.database_operations import initialize_database
from database.database_operations import create_video_record
from database.database_operations import create_download_record
from database.database_operations import create_lbry_upload_record
from database.database_operations import update_lbry_upload_record
def load_test_config():
config = ConfigParser()
config.read("tests/test_config.ini")
return config
@pytest.fixture(scope="function")
def db_session(request):
db_type = request.config.getoption("--db-type")
# Load the test configuration from the config file
config = load_test_config()
engine = get_database_engine(config)
Session = sessionmaker(bind=engine)
session = Session()
# Initialize the database with tables
Base.metadata.create_all(engine)
return session
# Test cases for database operations
def test_create_video_record(db_session):
# Test creating a video record in the database
# Sample video information
video_info = {
'video_source': 'youtube',
'id': 'PXTsVwd_j5A',
'title': 'THE JAMIL AL-AMIN CONSPIRACY',
'description': 'The Jamil Al-Amin (H. Rap Brown) Conspiracy. A true leader who walked with Martin Luther King, propagated the ideas of Malcolm X and became the Minister of Defense of the Black Panther Party is still alive and jailed under suspicious circumstances. Free Jamil Al Amin!',
'thumbnail': 'https://example.com/sample_thumbnail.jpg',
}
# Perform the operation
create_video_record(db_session, video_info, 'https://youtu.be/PXTsVwd_j5A')
# Check if the video record is created
video = db_session.query(Video).filter(Video.video_id == 'PXTsVwd_j5A').first()
assert video is not None
assert video.video_source == 'youtube'
assert video.title == 'THE JAMIL AL-AMIN CONSPIRACY'
assert video.description == 'The Jamil Al-Amin (H. Rap Brown) Conspiracy. A true leader who walked with Martin Luther King, propagated the ideas of Malcolm X and became the Minister of Defense of the Black Panther Party is still alive and jailed under suspicious circumstances. Free Jamil Al Amin!'
assert video.video_url == 'https://youtu.be/PXTsVwd_j5A'
assert video.thumbnail_url == 'https://example.com/sample_thumbnail.jpg'
def test_create_download_record(db_session):
# Test creating a download record in the database
# Sample video information
video_info = {
'video_source': 'youtube',
'id': 'PXTsVwd_j5A',
'download_path': 'download_archive/PXTsVwd_j5A.mp4',
}
# Perform the operation
create_download_record(db_session, video_info)
# Check if the download record is created
download = db_session.query(Download).filter(Download.video_id == 'PXTsVwd_j5A').first()
assert download is not None
assert download.video_source == 'youtube'
assert download.video_id == 'PXTsVwd_j5A'
assert download.download_path == 'download_archive/PXTsVwd_j5A.mp4'
def test_create_lbry_upload_record(db_session):
# Test creating an LBRYUpload record in the database
# Sample video information
video_info = {
'video_source': 'youtube',
'id': 'PXTsVwd_j5A',
}
# Perform the operation
create_lbry_upload_record(db_session, video_info)
# Check if the LBRYUpload record is created
lbry_upload = db_session.query(LBRYUpload).filter(LBRYUpload.video_id == 'PXTsVwd_j5A').first()
assert lbry_upload is not None
assert lbry_upload.video_source == 'youtube'
assert lbry_upload.video_id == 'PXTsVwd_j5A'
assert lbry_upload.lbry_status == 0
assert lbry_upload.lbry_url is None
