Archives
/
calibre-web
mirror of https://github.com/janeczku/calibre-web
2
0
Fork 0
Code Issues Releases Wiki Activity

Remove deprecated utcnow method

Browse Source 
Bugfix mp3 cover metadata extraction
pull/3085/merge
Ozzie Isaacs 1 month ago
parent 074aed6997
commit a56d1c80ae
9 changed files with 107 additions and 58 deletions
Show Stats Download Patch File Download Diff File

3
cps/__init__.py
Unescape Escape View File

@ -72,6 +72,9 @@ mimetypes.add_type('application/mpeg', '.mpeg')
mimetypes.add_type('audio/mpeg', '.mp3')
mimetypes.add_type('audio/x-m4a', '.m4a')
mimetypes.add_type('audio/x-m4a', '.m4b')
mimetypes.add_type('audio/x-hx-aac-adts', '.aac')
mimetypes.add_type('audio/vnd.dolby.dd-raw', '.ac3')
mimetypes.add_type('video/x-ms-asf', '.asf')
mimetypes.add_type('audio/ogg', '.ogg')
mimetypes.add_type('application/ogg', '.oga')
mimetypes.add_type('text/css', '.css')

56
cps/audio.py
Unescape Escape View File

@ -20,6 +20,7 @@ import os
import mutagen
import base64
from . import cover
from cps.constants import BookMeta
@ -27,22 +28,35 @@ from cps.constants import BookMeta
def get_audio_file_info(tmp_file_path, original_file_extension, original_file_name):
tmp_cover_name = None
audio_file = mutagen.File(tmp_file_path)
if original_file_extension in [".mp3", ".wav"]:
comments = None
if original_file_extension in [".mp3", ".wav", ".aiff"]:
cover_data = list()
for key, val in audio_file.tags.items():
if key.startswith("APIC:"):
cover_data.append(val)
if key.startswith("COMM:"):
comments = val.text[0]
title = audio_file.tags.get('TIT2').text[0] if "TIT2" in audio_file.tags else None
author = audio_file.tags.get('TPE1').text[0] if "TPE1" in audio_file.tags else None
if author is None:
author = audio_file.tags.get('TPE2').text[0] if "TPE2" in audio_file.tags else None
comments = audio_file.tags.get('COMM').text[0] if "COMM" in audio_file.tags else None
tags = audio_file.tags.get('TCON').text[0] if "TCON" in audio_file.tags else None # Genre
series = audio_file.tags.get('TALB').text[0] if "TALB" in audio_file.tags else None# Album
series_id = audio_file.tags.get('TRCK').text[0] if "TRCK" in audio_file.tags else None # track no.
publisher = audio_file.tags.get('TPUB').text[0] if "TPUB" in audio_file.tags else None
pubdate = audio_file.tags.get('XDOR').text[0] if "XDOR" in audio_file.tags else None
cover_data = audio_file.tags.get('APIC:')
pubdate = str(audio_file.tags.get('TDRL').text[0]) if "TDRL" in audio_file.tags else None
if not pubdate:
pubdate = str(audio_file.tags.get('TDRC').text[0]) if "TDRC" in audio_file.tags else None
if not pubdate:
pubdate = str(audio_file.tags.get('TDOR').text[0]) if "TDOR" in audio_file.tags else None
if cover_data:
tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg')
with open(tmp_cover_name, "wb") as cover_file:
cover_file.write(cover_data.data)
cover_info = cover_data[0]
for dat in cover_data:
if dat.type == mutagen.id3.PictureType.COVER_FRONT:
cover_info = dat
break
cover.cover_processing(tmp_file_path, cover_info.data, "." + cover_info.mime[-3:])
elif original_file_extension in [".ogg", ".flac"]:
title = audio_file.tags.get('TITLE')[0] if "TITLE" in audio_file else None
author = audio_file.tags.get('ARTIST')[0] if "ARTIST" in audio_file else None
@ -61,6 +75,36 @@ def get_audio_file_info(tmp_file_path, original_file_extension, original_file_na
tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg')
with open(tmp_cover_name, "wb") as cover_file:
cover_file.write(audio_file.pictures[0].data)
elif original_file_extension in [".aac"]:
title = audio_file.tags.get('Title').value if "title" in audio_file else None
author = audio_file.tags.get('Artist').value if "artist" in audio_file else None
comments = None # audio_file.tags.get('COMM', None)
tags = ""
series = audio_file.tags.get('Album').value if "Album" in audio_file else None
series_id = audio_file.tags.get('Track').value if "Track" in audio_file else None
publisher = audio_file.tags.get('Label').value if "Label" in audio_file else None
pubdate = audio_file.tags.get('Year').value if "Year" in audio_file else None
cover_data = audio_file.tags['Cover Art (Front)']
if cover_data:
tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg')
with open(tmp_cover_name, "wb") as cover_file:
cover_file.write(cover_data.value.split(b"\x00",1)[1])
elif original_file_extension in [".asf"]:
title = audio_file.tags.get('Title')[0].value if "title" in audio_file else None
author = audio_file.tags.get('Artist')[0].value if "artist" in audio_file else None
comments = None # audio_file.tags.get('COMM', None)
tags = ""
series = audio_file.tags.get('Album')[0].value if "Album" in audio_file else None
series_id = audio_file.tags.get('Track')[0].value if "Track" in audio_file else None
publisher = audio_file.tags.get('Label')[0].value if "Label" in audio_file else None
pubdate = audio_file.tags.get('Year')[0].value if "Year" in audio_file else None
cover_data = audio_file.tags['WM/Picture']
if cover_data:
tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg')
with open(tmp_cover_name, "wb") as cover_file:
cover_file.write(cover_data[0].value)
return BookMeta(
file_path=tmp_file_path,

4
cps/db.py
Unescape Escape View File

@ -1029,10 +1029,10 @@ class CalibreDB:
return title.strip()
try:
# sqlalchemy <1.4.24
# sqlalchemy <1.4.24 and sqlalchemy 2.0
conn = conn or self.session.connection().connection.driver_connection
except AttributeError:
# sqlalchemy >1.4.24 and sqlalchemy 2.0
# sqlalchemy >1.4.24
conn = conn or self.session.connection().connection.connection
try:
conn.create_function("title_sort", 1, _title_sort)

8
cps/editbooks.py
Unescape Escape View File

@ -246,8 +246,12 @@ def upload():
modify_date = False
# create the function for sorting...
calibre_db.update_title_sort(config)
calibre_db.session.connection().connection.connection.create_function('uuid4', 0, lambda: str(uuid4()))
try:
# sqlalchemy 2.0
uuid_func = calibre_db.session.connection().connection.driver_connection
except AttributeError:
uuid_func = calibre_db.session.connection().connection.connection
uuid_func.create_function('uuid4', 0,lambda: str(uuid4()))
meta, error = file_handling_on_upload(requested_file)
if error:
return error

43
cps/helper.py
Unescape Escape View File

@ -788,24 +788,23 @@ def get_book_cover_internal(book, resolution=None):
def get_book_cover_thumbnail(book, resolution):
if book and book.has_cover:
return ub.session \
.query(ub.Thumbnail) \
.filter(ub.Thumbnail.type == THUMBNAIL_TYPE_COVER) \
.filter(ub.Thumbnail.entity_id == book.id) \
.filter(ub.Thumbnail.resolution == resolution) \
.filter(or_(ub.Thumbnail.expiration.is_(None), ub.Thumbnail.expiration > datetime.now(UTC)) \
.first()
return (ub.session
.query(ub.Thumbnail)
.filter(ub.Thumbnail.type == THUMBNAIL_TYPE_COVER)
.filter(ub.Thumbnail.entity_id == book.id)
.filter(ub.Thumbnail.resolution == resolution)
.filter(or_(ub.Thumbnail.expiration.is_(None), ub.Thumbnail.expiration > datetime.now(UTC)))
.first())
def get_series_thumbnail_on_failure(series_id, resolution):
book = calibre_db.session \
.query(db.Books) \
.join(db.books_series_link) \
.join(db.Series) \
.filter(db.Series.id == series_id) \
.filter(db.Books.has_cover == 1) \
.first()
book = (calibre_db.session
.query(db.Books)
.join(db.books_series_link)
.join(db.Series)
.filter(db.Series.id == series_id)
.filter(db.Books.has_cover == 1)
.first())
return get_book_cover_internal(book, resolution=resolution)
@ -827,13 +826,13 @@ def get_series_cover_internal(series_id, resolution=None):
def get_series_thumbnail(series_id, resolution):
return ub.session \
.query(ub.Thumbnail) \
.filter(ub.Thumbnail.type == THUMBNAIL_TYPE_SERIES) \
.filter(ub.Thumbnail.entity_id == series_id) \
.filter(ub.Thumbnail.resolution == resolution) \
.filter(or_(ub.Thumbnail.expiration.is_(None), ub.Thumbnail.expiration > datetime.now(UTC))) \
.first()
return (ub.session
.query(ub.Thumbnail)
.filter(ub.Thumbnail.type == THUMBNAIL_TYPE_SERIES)
.filter(ub.Thumbnail.entity_id == series_id)
.filter(ub.Thumbnail.resolution == resolution)
.filter(or_(ub.Thumbnail.expiration.is_(None), ub.Thumbnail.expiration > datetime.now(UTC)))
.first())
# saves book cover from url

2
cps/kobo.py
Unescape Escape View File

@ -795,7 +795,7 @@ def HandleStateRequest(book_uuid):
if new_book_read_status == ub.ReadBook.STATUS_IN_PROGRESS \
and new_book_read_status != book_read.read_status:
book_read.times_started_reading += 1
book_read.last_time_started_reading = datetime.datetime.now(UTC)
book_read.last_time_started_reading = datetime.now(UTC)
book_read.read_status = new_book_read_status
update_results_response["StatusInfoResult"] = {"Result": "Success"}
except (KeyError, TypeError, ValueError, StatementError):

15
cps/tasks/thumbnail.py
Unescape Escape View File

@ -20,14 +20,13 @@ import os
from shutil import copyfile, copyfileobj
from urllib.request import urlopen
from io import BytesIO
from datetime import datetime, UTC
from .. import constants
from cps import config, db, fs, gdriveutils, logger, ub
from cps.services.worker import CalibreTask, STAT_CANCELLED, STAT_ENDED
from datetime import datetime
from sqlalchemy import func, text, or_
from flask_babel import lazy_gettext as N_
try:
from wand.image import Image
use_IM = True
@ -322,12 +321,12 @@ class TaskGenerateSeriesThumbnails(CalibreTask):
.all()
def get_series_thumbnails(self, series_id):
return self.app_db_session \
.query(ub.Thumbnail) \
.filter(ub.Thumbnail.type == constants.THUMBNAIL_TYPE_SERIES) \
.filter(ub.Thumbnail.entity_id == series_id) \
.filter(or_(ub.Thumbnail.expiration.is_(None), ub.Thumbnail.expiration > datetime.now(UTC))) \
.all()
return (self.app_db_session
.query(ub.Thumbnail)
.filter(ub.Thumbnail.type == constants.THUMBNAIL_TYPE_SERIES)
.filter(ub.Thumbnail.entity_id == series_id)
.filter(or_(ub.Thumbnail.expiration.is_(None), ub.Thumbnail.expiration > datetime.now(UTC)))
.all())
def create_series_thumbnail(self, series, series_books, resolution):
thumbnail = ub.Thumbnail()

32
cps/ub.py
Unescape Escape View File

@ -20,7 +20,7 @@
import atexit
import os
import sys
from datetime import datetime, UTC
from datetime import datetime, UTC, timedelta
import itertools
import uuid
from flask import session as flask_session
@ -77,7 +77,7 @@ def store_user_session():
if flask_session.get('_user_id', ""):
try:
if not check_user_session(_user, _id, _random):
expiry = int((datetime.datetime.now() + datetime.timedelta(days=31)).timestamp())
expiry = int((datetime.now() + timedelta(days=31)).timestamp())
user_session = User_Sessions(_user, _id, _random, expiry)
session.add(user_session)
session.commit()
@ -109,7 +109,7 @@ def check_user_session(user_id, session_key, random):
User_Sessions.random == random,
).one_or_none()
if found is not None:
new_expiry = int((datetime.datetime.now() + datetime.timedelta(days=31)).timestamp())
new_expiry = int((datetime.now() + timedelta(days=31)).timestamp())
if new_expiry - found.expiry > 86400:
found.expiry = new_expiry
session.merge(found)
@ -370,8 +370,8 @@ class Shelf(Base):
user_id = Column(Integer, ForeignKey('user.id'))
kobo_sync = Column(Boolean, default=False)
books = relationship("BookShelf", backref="ub_shelf", cascade="all, delete-orphan", lazy="dynamic")
created = Column(DateTime, default=datetime.datetime.utcnow)
last_modified = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
created = Column(DateTime, default=lambda: datetime.now(UTC))
last_modified = Column(DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
def __repr__(self):
return '<Shelf %d:%r>' % (self.id, self.name)
@ -385,7 +385,7 @@ class BookShelf(Base):
book_id = Column(Integer)
order = Column(Integer)
shelf = Column(Integer, ForeignKey('shelf.id'))
date_added = Column(DateTime, default=datetime.datetime.utcnow)
date_added = Column(DateTime, default=lambda: datetime.now(UTC))
def __repr__(self):
return '<Book %r>' % self.id
@ -398,7 +398,7 @@ class ShelfArchive(Base):
id = Column(Integer, primary_key=True)
uuid = Column(String)
user_id = Column(Integer, ForeignKey('user.id'))
last_modified = Column(DateTime, default=datetime.datetime.utcnow)
last_modified = Column(DateTime, default=lambda: datetime.now(UTC))
class ReadBook(Base):
@ -418,7 +418,7 @@ class ReadBook(Base):
cascade="all",
backref=backref("book_read_link",
uselist=False))
last_modified = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
last_modified = Column(DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
last_time_started_reading = Column(DateTime, nullable=True)
times_started_reading = Column(Integer, default=0, nullable=False)
@ -441,7 +441,7 @@ class ArchivedBook(Base):
user_id = Column(Integer, ForeignKey('user.id'))
book_id = Column(Integer)
is_archived = Column(Boolean, unique=False)
last_modified = Column(DateTime, default=datetime.datetime.utcnow)
last_modified = Column(DateTime, default=lambda: datetime.now(UTC))
class KoboSyncedBooks(Base):
@ -460,8 +460,8 @@ class KoboReadingState(Base):
id = Column(Integer, primary_key=True, autoincrement=True)
user_id = Column(Integer, ForeignKey('user.id'))
book_id = Column(Integer)
last_modified = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
priority_timestamp = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
last_modified = Column(DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
priority_timestamp = Column(DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
current_bookmark = relationship("KoboBookmark", uselist=False, backref="kobo_reading_state", cascade="all, delete")
statistics = relationship("KoboStatistics", uselist=False, backref="kobo_reading_state", cascade="all, delete")
@ -471,7 +471,7 @@ class KoboBookmark(Base):
id = Column(Integer, primary_key=True)
kobo_reading_state_id = Column(Integer, ForeignKey('kobo_reading_state.id'))
last_modified = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
last_modified = Column(DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
location_source = Column(String)
location_type = Column(String)
location_value = Column(String)
@ -484,7 +484,7 @@ class KoboStatistics(Base):
id = Column(Integer, primary_key=True)
kobo_reading_state_id = Column(Integer, ForeignKey('kobo_reading_state.id'))
last_modified = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
last_modified = Column(DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
remaining_time_minutes = Column(Integer)
spent_reading_minutes = Column(Integer)
@ -496,7 +496,7 @@ def receive_before_flush(session, flush_context, instances):
if isinstance(change, (ReadBook, KoboStatistics, KoboBookmark)):
if change.kobo_reading_state:
change.kobo_reading_state.last_modified = datetime.now(UTC)
# Maintain the last_modified bit for the Shelf table.
# Maintain the last_modified_bit for the Shelf table.
for change in itertools.chain(session.new, session.deleted):
if isinstance(change, BookShelf):
change.ub_shelf.last_modified = datetime.now(UTC)
@ -539,7 +539,7 @@ class RemoteAuthToken(Base):
def __init__(self):
super().__init__()
self.auth_token = (hexlify(os.urandom(4))).decode('utf-8')
self.expiration = datetime.datetime.now() + datetime.timedelta(minutes=10) # 10 min from now
self.expiration = datetime.now() + timedelta(minutes=10) # 10 min from now
def __repr__(self):
return '<Token %r>' % self.id
@ -614,7 +614,7 @@ def migrate_Database(_session):
def clean_database(_session):
# Remove expired remote login tokens
now = datetime.datetime.now()
now = datetime.now()
try:
_session.query(RemoteAuthToken).filter(now > RemoteAuthToken.expiration).\
filter(RemoteAuthToken.token_type != 1).delete()

2
cps/uploader.py
Unescape Escape View File

@ -91,7 +91,7 @@ def process(tmp_file_path, original_file_name, original_file_extension, rar_exec
original_file_name,
original_file_extension,
rar_executable)
elif extension_upper in [".MP3", ".OGG", ".FLAC", ".WAV"] and use_audio_meta:
elif extension_upper in [".MP3", ".OGG", ".FLAC", ".WAV", ".AAC", ".AIFF", ".ASF", ".MP4"] and use_audio_meta:
meta = audio.get_audio_file_info(tmp_file_path, original_file_extension, original_file_name)
except Exception as ex:
log.warning('cannot parse metadata, using default: %s', ex)

Write Preview
Loading…
Cancel
Save