Skip to content

clean up key/bpm repairer #9077

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
michellebrier merged 2 commits into from
Jul 15, 2024
Merged

clean up key/bpm repairer #9077

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
04b7683
clean up repairer
michellebrier Jul 9, 2024
1de857f
log
michellebrier Jul 9, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 4 additions & 85 deletions packages/discovery-provider/src/tasks/repair_audio_analyses.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,15 +1,12 @@
import json
import random
import re
from datetime import datetime
from typing import List, Optional
from typing import List

import requests
from redis import Redis
from sqlalchemy import or_
from sqlalchemy.orm.session import Session

from src.api_helpers import generate_signature
from src.models.tracks.track import Track
from src.tasks.celery_app import celery
from src.utils import get_all_nodes
Expand All @@ -20,7 +17,7 @@

REPAIR_AUDIO_ANALYSES_LOCK = "repair_audio_analyses_lock"
DEFAULT_LOCK_TIMEOUT_SECONDS = 30 * 60 # 30 minutes
BATCH_SIZE = 1000
BATCH_SIZE = 5000


def query_tracks(session: Session) -> List[Track]:
Expand All @@ -35,7 +32,7 @@ def query_tracks(session: Session) -> List[Track]:
Track.genre != "Podcast",
Track.genre != "Audiobooks",
)
.order_by(Track.track_id.asc())
.order_by(Track.track_id.desc())
.limit(BATCH_SIZE)
.all()
)
Expand All @@ -49,69 +46,16 @@ def select_content_nodes(redis: Redis):
return random.sample(endpoints, min(5, len(endpoints)))


def retrigger_audio_analysis(
nodes: List[str],
track_id: int,
track_cid: str,
upload_id: Optional[str],
legacy_track: bool,
):
cid = ""
if legacy_track:
cid = track_cid
upload_id = ""
data = {
"trackId": track_id,
"cid": cid,
"timestamp": int(datetime.utcnow().timestamp() * 1000),
"upload_id": upload_id,
}
signature = generate_signature(data)
params = {
"signature": json.dumps(
{
"data": json.dumps(data),
"signature": signature,
}
)
}
for node in nodes:
try:
endpoint = (
f"{node}/tracks/legacy/{track_cid}/analyze"
if legacy_track
else f"{node}/uploads/{upload_id}/analyze"
)
resp = requests.post(endpoint, params=params, timeout=5)
if resp.status_code == 400:
# Should only return a 400 if content found the track_cid to not be an audio file, therefore unable to be analyzed.
# Quit early because all CNs will 400 for this cid.
logger.warning(
f"repair_audio_analyses.py | attempt to trigger audio analysis for track {track_id} (track_cid: {track_cid}, audio_upload_id: {upload_id}) failed with status code {resp.status_code}: {resp.text}"
)
return
resp.raise_for_status()
return
except Exception:
# Fallback to the next node
continue
logger.warning(
f"repair_audio_analyses.py | failed to trigger audio analysis for track {track_id} (track_cid: {track_cid}, audio_upload_id: {upload_id}). tried {nodes}"
)


def repair(session: Session, redis: Redis):
# Query batch of tracks that are missing key or bpm and have err counts < 3 from db
tracks = query_tracks(session)
nodes = select_content_nodes(redis)
num_tracks_updated = 0
num_analyses_retriggered = 0
for track in tracks:
if not track.track_cid:
# Only analyze streamable tracks
continue
legacy_track = not track.audio_upload_id
found = False
for node in nodes:
try:
# Query random content node for the audio upload id
Expand Down Expand Up @@ -156,39 +100,14 @@ def repair(session: Session, redis: Redis):
if track_updated:
num_tracks_updated += 1

# Failures get retried up to 3 times
# if (not key or not bpm) and error_count < 3:
# # Trigger another audio analysis but don't bother polling for result. Will read it in next batch.
# num_analyses_retriggered += 1
# retrigger_audio_analysis(
# nodes,
# track.track_id,
# track.track_cid,
# track.audio_upload_id,
# legacy_track,
# )
if error_count >= 3:
logger.warning(
f"repair_audio_analyses.py | Track ID {track.track_id} (track_cid: {track.track_cid}, audio_upload_id: {track.audio_upload_id}) failed audio analysis >= 3 times"
)
found = True
break

if not found:
logger.warning(
f"repair_audio_analyses.py | failed to query audio analysis for track {track.track_id} (track_cid: {track.track_cid}, audio_upload_id: {track.audio_upload_id}). tried {nodes}. triggering analysis..."
)
num_analyses_retriggered += 1
retrigger_audio_analysis(
nodes,
track.track_id,
track.track_cid,
track.audio_upload_id,
legacy_track,
)

logger.info(
f"repair_audio_analyses.py | updated {num_tracks_updated} tracks, retriggered analyses for {num_analyses_retriggered} tracks. last track ID processed: {tracks[-1].track_id}"
f"repair_audio_analyses.py | updated {num_tracks_updated} tracks. last track ID processed: {tracks[-1].track_id}"
)


Expand Down