mirror of
https://github.com/Viren070/MediaFusion.git
synced 2025-12-01 23:21:11 +01:00
Refactor RPDB poster update logic for batch processing.
Updated the RPDB poster update flow to handle multiple IMDB IDs in batches, improving efficiency. Replaced single poster checks with a batch-based approach and optimized Redis operations. Enhanced error handling and reduced redundant asynchronous calls.
This commit is contained in:
mhdzumair
+135
-62
@@ -13,91 +13,164 @@ RPDB_UNSUPPORTED_HASH = "rpdb_unsupported_ids"
|
||||
RPDB_UNSUPPORTED_EXPIRY = 60 * 60 * 24 * 7 # 7 days in seconds
|
||||
|
||||
|
||||
async def check_rpdb_poster_availability(rpdb_poster_url: str) -> bool:
|
||||
async def check_rpdb_posters_availability(urls: list[str]) -> dict[str, bool]:
|
||||
"""Batch check multiple poster URLs availability."""
|
||||
results = {}
|
||||
async with httpx.AsyncClient(
|
||||
timeout=10.0, proxy=settings.requests_proxy_url
|
||||
) as client:
|
||||
|
||||
async def check_url(url: str):
|
||||
try:
|
||||
response = await client.head(url)
|
||||
results[url] = response.status_code == 200
|
||||
except Exception as exc:
|
||||
logging.error(f"Exception for {url} - {exc}")
|
||||
results[url] = False
|
||||
|
||||
await asyncio.gather(*(check_url(url) for url in urls), return_exceptions=True)
|
||||
return results
|
||||
|
||||
|
||||
async def safe_redis_operation(pipe, operation: str, *args):
|
||||
"""Execute Redis operation safely, logging errors but not raising them."""
|
||||
try:
|
||||
async with httpx.AsyncClient(
|
||||
timeout=10.0, proxy=settings.requests_proxy_url
|
||||
) as client:
|
||||
response = await client.head(rpdb_poster_url)
|
||||
return response.status_code == 200
|
||||
except httpx.TimeoutException as exc:
|
||||
logging.error(f"Timeout for {rpdb_poster_url} - {exc}")
|
||||
return False
|
||||
except httpx.HTTPError as exc:
|
||||
logging.error(f"HTTP Exception for {exc.request.url} - {exc}")
|
||||
return False
|
||||
except Exception as exc:
|
||||
logging.error(f"Exception for {rpdb_poster_url} - {exc}")
|
||||
return False
|
||||
if operation == "sadd":
|
||||
await pipe.sadd(*args)
|
||||
elif operation == "hset":
|
||||
await pipe.hset(*args)
|
||||
except Exception as e:
|
||||
logging.error(f"Redis {operation} failed for {args}: {e}")
|
||||
|
||||
|
||||
async def update_single_rpdb_poster(imdb_id: str, rpdb_poster_base: str) -> str | None:
|
||||
if not imdb_id.startswith("tt"):
|
||||
return None
|
||||
|
||||
rpdb_poster_url = f"{rpdb_poster_base}{imdb_id}.jpg"
|
||||
async def batch_update_rpdb_posters(
|
||||
imdb_ids: list[str], rpdb_poster_base: str, batch_size: int = 50
|
||||
) -> dict[str, str]:
|
||||
"""Process multiple IMDB IDs in batches to update poster URLs."""
|
||||
current_time = int(time.time())
|
||||
result_urls = {}
|
||||
|
||||
# Check if the IMDB ID is in the supported set
|
||||
if await REDIS_ASYNC_CLIENT.sismember(RPDB_SUPPORTED_SET, imdb_id):
|
||||
return rpdb_poster_url
|
||||
# Filter valid IMDB IDs
|
||||
valid_imdb_ids = [id for id in imdb_ids if id.startswith("tt")]
|
||||
if not valid_imdb_ids:
|
||||
return result_urls
|
||||
|
||||
# Check if the IMDB ID is in the unsupported hash
|
||||
expiry_time = await REDIS_ASYNC_CLIENT.hget(RPDB_UNSUPPORTED_HASH, imdb_id)
|
||||
try:
|
||||
# Check supported set
|
||||
supported = await REDIS_ASYNC_CLIENT.smembers(RPDB_SUPPORTED_SET)
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to get supported IDs from Redis: {e}")
|
||||
supported = set()
|
||||
|
||||
if expiry_time:
|
||||
expiry_time = int(expiry_time)
|
||||
if expiry_time > current_time:
|
||||
# Still within expiry period, return None
|
||||
return None
|
||||
else:
|
||||
# Expired, remove from unsupported hash
|
||||
await REDIS_ASYNC_CLIENT.hdel(RPDB_UNSUPPORTED_HASH, imdb_id)
|
||||
try:
|
||||
# Check unsupported hash
|
||||
unsupported = await REDIS_ASYNC_CLIENT.hgetall(RPDB_UNSUPPORTED_HASH)
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to get unsupported IDs from Redis: {e}")
|
||||
unsupported = {}
|
||||
|
||||
# Check availability (either not in unsupported hash or expired)
|
||||
if await check_rpdb_poster_availability(rpdb_poster_url):
|
||||
await REDIS_ASYNC_CLIENT.sadd(RPDB_SUPPORTED_SET, imdb_id)
|
||||
return rpdb_poster_url
|
||||
else:
|
||||
new_expiry_time = current_time + RPDB_UNSUPPORTED_EXPIRY
|
||||
await REDIS_ASYNC_CLIENT.hset(
|
||||
RPDB_UNSUPPORTED_HASH, imdb_id, str(new_expiry_time)
|
||||
)
|
||||
return None
|
||||
# Process supported IDs
|
||||
for imdb_id in valid_imdb_ids:
|
||||
if imdb_id in supported:
|
||||
result_urls[imdb_id] = f"{rpdb_poster_base}{imdb_id}.jpg"
|
||||
|
||||
# Filter out IDs that need checking
|
||||
to_check = []
|
||||
for imdb_id in valid_imdb_ids:
|
||||
if imdb_id in result_urls:
|
||||
continue
|
||||
|
||||
async def update_rpdb_poster(
|
||||
meta_item: schemas.MetaItem, user_data: schemas.UserData, catalog_type: str
|
||||
) -> schemas.MetaItem:
|
||||
if not user_data.rpdb_config or catalog_type not in ["movie", "series"]:
|
||||
return meta_item
|
||||
expiry_time = unsupported.get(imdb_id)
|
||||
if expiry_time and int(expiry_time) > current_time:
|
||||
continue
|
||||
|
||||
rpdb_poster_base = f"https://api.ratingposterdb.com/{user_data.rpdb_config.api_key}/imdb/poster-default/"
|
||||
updated_poster = await update_single_rpdb_poster(
|
||||
meta_item.meta.id, rpdb_poster_base
|
||||
)
|
||||
to_check.append(imdb_id)
|
||||
|
||||
if updated_poster:
|
||||
meta_item.meta.poster = updated_poster
|
||||
# Process remaining IDs in batches
|
||||
for i in range(0, len(to_check), batch_size):
|
||||
batch = to_check[i : i + batch_size]
|
||||
urls_to_check = [f"{rpdb_poster_base}{id}.jpg" for id in batch]
|
||||
|
||||
return meta_item
|
||||
# Check availability in parallel
|
||||
availability = await check_rpdb_posters_availability(urls_to_check)
|
||||
|
||||
# Update Redis and results for each ID independently
|
||||
for imdb_id, url in zip(batch, urls_to_check):
|
||||
try:
|
||||
if availability.get(url, False):
|
||||
# Try to add to supported set
|
||||
try:
|
||||
await REDIS_ASYNC_CLIENT.sadd(RPDB_SUPPORTED_SET, imdb_id)
|
||||
result_urls[imdb_id] = url
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to add {imdb_id} to supported set: {e}")
|
||||
# Still include in results even if Redis fails
|
||||
result_urls[imdb_id] = url
|
||||
else:
|
||||
# Try to add to unsupported hash
|
||||
try:
|
||||
new_expiry_time = current_time + RPDB_UNSUPPORTED_EXPIRY
|
||||
await REDIS_ASYNC_CLIENT.hset(
|
||||
RPDB_UNSUPPORTED_HASH, imdb_id, str(new_expiry_time)
|
||||
)
|
||||
except Exception as e:
|
||||
logging.error(
|
||||
f"Failed to add {imdb_id} to unsupported hash: {e}"
|
||||
)
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to process {imdb_id}: {e}")
|
||||
continue
|
||||
|
||||
return result_urls
|
||||
|
||||
|
||||
async def update_rpdb_posters(
|
||||
metas: schemas.Metas, user_data: schemas.UserData, catalog_type: str
|
||||
) -> schemas.Metas:
|
||||
"""Update multiple meta items with RPDB posters in an optimized way."""
|
||||
if not user_data.rpdb_config or catalog_type not in ["movie", "series"]:
|
||||
return metas
|
||||
|
||||
rpdb_poster_base = f"https://api.ratingposterdb.com/{user_data.rpdb_config.api_key}/imdb/poster-default/"
|
||||
|
||||
async def update_poster(meta):
|
||||
updated_poster = await update_single_rpdb_poster(meta.id, rpdb_poster_base)
|
||||
if updated_poster:
|
||||
meta.poster = updated_poster
|
||||
try:
|
||||
# Get all IMDB IDs
|
||||
imdb_ids = [meta.id for meta in metas.metas]
|
||||
|
||||
await asyncio.gather(
|
||||
*(update_poster(meta) for meta in metas.metas), return_exceptions=True
|
||||
)
|
||||
# Batch process all posters
|
||||
poster_urls = await batch_update_rpdb_posters(imdb_ids, rpdb_poster_base)
|
||||
|
||||
# Update meta items with new poster URLs
|
||||
for meta in metas.metas:
|
||||
if meta.id in poster_urls:
|
||||
meta.poster = poster_urls[meta.id]
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to update RPDB posters: {e}")
|
||||
# Return original metas if update fails
|
||||
return metas
|
||||
|
||||
return metas
|
||||
|
||||
|
||||
async def update_rpdb_poster(
|
||||
meta_item: schemas.MetaItem, user_data: schemas.UserData, catalog_type: str
|
||||
) -> schemas.MetaItem:
|
||||
"""Update single meta item with RPDB poster."""
|
||||
if not user_data.rpdb_config or catalog_type not in ["movie", "series"]:
|
||||
return meta_item
|
||||
|
||||
rpdb_poster_base = f"https://api.ratingposterdb.com/{user_data.rpdb_config.api_key}/imdb/poster-default/"
|
||||
|
||||
try:
|
||||
poster_urls = await batch_update_rpdb_posters(
|
||||
[meta_item.meta.id], rpdb_poster_base
|
||||
)
|
||||
|
||||
if meta_item.meta.id in poster_urls:
|
||||
meta_item.meta.poster = poster_urls[meta_item.meta.id]
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to update RPDB poster for {meta_item.meta.id}: {e}")
|
||||
# Return original meta_item if update fails
|
||||
return meta_item
|
||||
|
||||
return meta_item
|
||||
|
||||
Reference in New Issue
Block a user