quaik8/sdb2-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: Lint fixes of services
All checks were successful
Backend CI / test (push) Successful in 3m59s
Details

Browse Source
This commit is contained in:
JSC
2025-08-01 01:27:47 +02:00
parent 95ccb76233
commit a10111793c
12 changed files with 237 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
app/services/sound_normalizer.py
View File

@@ -153,7 +153,9 @@ class SoundNormalizerService:
"""Normalize audio using two-pass loudnorm for better quality."""
try:
logger.info(
"Starting two-pass normalization: %s -> %s", input_path, output_path,
"Starting two-pass normalization: %s -> %s",
input_path,
output_path,
)
# First pass: analyze
@@ -177,7 +179,7 @@ class SoundNormalizerService:
result = ffmpeg.run(stream, capture_stderr=True, quiet=True)
analysis_output = result[1].decode("utf-8")
except ffmpeg.Error as e:
logger.error(
logger.exception(
"FFmpeg first pass failed for %s. Stdout: %s, Stderr: %s",
input_path,
e.stdout.decode() if e.stdout else "None",
@@ -193,9 +195,11 @@ class SoundNormalizerService:
json_match = re.search(r'\{[^{}]*"input_i"[^{}]*\}', analysis_output)
if not json_match:
logger.error(
"Could not find JSON in loudnorm output: %s", analysis_output,
"Could not find JSON in loudnorm output: %s",
analysis_output,
)
raise ValueError("Could not extract loudnorm analysis data")
msg = "Could not find JSON in loudnorm output"
raise ValueError(msg)
logger.debug("Found JSON match: %s", json_match.group())
analysis_data = json.loads(json_match.group())
@@ -211,7 +215,10 @@ class SoundNormalizerService:
]:
if str(analysis_data.get(key, "")).lower() in invalid_values:
logger.warning(
"Invalid analysis value for %s: %s. Falling back to one-pass normalization.",
(
"Invalid analysis value for %s: %s. "
"Falling back to one-pass normalization."
),
key,
analysis_data.get(key),
)
@@ -252,7 +259,7 @@ class SoundNormalizerService:
ffmpeg.run(stream, quiet=True, overwrite_output=True)
logger.info("Two-pass normalization completed: %s", output_path)
except ffmpeg.Error as e:
logger.error(
logger.exception(
"FFmpeg second pass failed for %s. Stdout: %s, Stderr: %s",
input_path,
e.stdout.decode() if e.stdout else "None",
@@ -267,12 +274,14 @@ class SoundNormalizerService:
async def normalize_sound(
self,
sound: Sound,
*,
force: bool = False,
one_pass: bool | None = None,
sound_data: dict | None = None,
) -> NormalizationInfo:
"""Normalize a single sound."""
# Use provided sound_data to avoid detached instance issues, or capture from sound
# Use provided sound_data to avoid detached instance issues,
# or capture from sound
if sound_data:
filename = sound_data["filename"]
sound_id = sound_data["id"]
@@ -391,6 +400,7 @@ class SoundNormalizerService:
async def normalize_all_sounds(
self,
*,
force: bool = False,
one_pass: bool | None = None,
) -> NormalizationResults:
@@ -409,7 +419,7 @@ class SoundNormalizerService:
if force:
# Get all sounds if forcing
sounds = []
for sound_type in self.type_directories.keys():
for sound_type in self.type_directories:
type_sounds = await self.sound_repo.get_by_type(sound_type)
sounds.extend(type_sounds)
else:
@@ -419,17 +429,16 @@ class SoundNormalizerService:
logger.info("Found %d sounds to process", len(sounds))
# Capture all sound data upfront to avoid session detachment issues
sound_data_list = []
for sound in sounds:
sound_data_list.append(
{
"id": sound.id,
"filename": sound.filename,
"type": sound.type,
"is_normalized": sound.is_normalized,
"name": sound.name,
},
)
sound_data_list = [
{
"id": sound.id,
"filename": sound.filename,
"type": sound.type,
"is_normalized": sound.is_normalized,
"name": sound.name,
}
for sound in sounds
]
# Process each sound using captured data
for i, sound in enumerate(sounds):
@@ -485,6 +494,7 @@ class SoundNormalizerService:
async def normalize_sounds_by_type(
self,
sound_type: str,
*,
force: bool = False,
one_pass: bool | None = None,
) -> NormalizationResults:
@@ -508,17 +518,16 @@ class SoundNormalizerService:
logger.info("Found %d %s sounds to process", len(sounds), sound_type)
# Capture all sound data upfront to avoid session detachment issues
sound_data_list = []
for sound in sounds:
sound_data_list.append(
{
"id": sound.id,
"filename": sound.filename,
"type": sound.type,
"is_normalized": sound.is_normalized,
"name": sound.name,
},
)
sound_data_list = [
{
"id": sound.id,
"filename": sound.filename,
"type": sound.type,
"is_normalized": sound.is_normalized,
"name": sound.name,
}
for sound in sounds
]
# Process each sound using captured data
for i, sound in enumerate(sounds):