diff --git a/db/migrations.py b/db/migrations.py deleted file mode 100644 index b963862..0000000 --- a/db/migrations.py +++ /dev/null @@ -1,281 +0,0 @@ -import asyncio -import logging -from datetime import datetime - -from PTT import parse_title -from pymongo import ASCENDING - -from db import database -from db.crud import update_meta_stream, update_metadata -from db.models import ( - TorrentStreams, - MediaFusionSeriesMetaData, - TVStreams, - MediaFusionMetaData, -) - -# Configure logging -logging.basicConfig( - level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" -) -logger = logging.getLogger(__name__) - - -async def migrate_series_streams(): - """ - Fetch series metadata IDs, retrieve TorrentStreams with empty episode lists, - parse titles using PTT.parse_title, and create episodes based on this information. - """ - logger.info("Starting Series Streams Migration...") - - meta_collection = MediaFusionSeriesMetaData.get_motor_collection() - stream_collection = TorrentStreams.get_motor_collection() - - try: - # Fetch all series metadata IDs - series_ids = await meta_collection.find({"type": "series"}).distinct("_id") - if not series_ids: - logger.info("No series metadata IDs found.") - return - count = await stream_collection.count_documents( - {"meta_id": {"$in": series_ids}, "episode_files": {"$size": 0}} - ) - logger.info(f"Found {count} TorrentStreams with empty episode lists.") - # Find TorrentStreams with empty episode lists - async for stream in stream_collection.find( - {"meta_id": {"$in": series_ids}, "episode_files": {"$size": 0}} - ): - stream_id = stream["_id"] - title = stream.get("torrent_name", "") - - # Parse the torrent title - parsed_data = parse_title(title) - if not parsed_data: - logger.warning(f"Failed to parse title for stream ID: {stream_id}") - continue - - # Create episode details based on parsed data - seasons = parsed_data.get("seasons") - episodes = parsed_data.get("episodes") - - if len(seasons) == 1 and episodes: - episode_details = [ - { - "season_number": seasons[0], - "episode_number": episode, - } - for episode in episodes - ] - elif len(seasons) == 1 and not episodes: - episode_details = [ - { - "season_number": seasons[0], - "episode_number": 1, - } - ] - elif len(seasons) > 1: - episode_details = [ - { - "season_number": season, - "episode_number": 1, - } - for season in seasons - ] - elif episodes: - episode_details = [ - { - "season_number": 1, - "episode_number": episode, - } - for episode in episodes - ] - elif parsed_data.get("date") and stream["meta_id"].startswith("tt"): - episode_date = datetime.strptime(parsed_data["date"], "%Y-%m-%d") - metadata = await MediaFusionSeriesMetaData.get(stream["meta_id"]) - episode = [ - episode - for episode in metadata.episodes - if episode.released - and episode.released.date() == episode_date.date() - ] - if not metadata.episodes or ( - not episode and metadata.episodes[0].thumbnail - ): - await update_metadata([metadata.id], "series") - metadata = await MediaFusionSeriesMetaData.get(stream["meta_id"]) - episode = [ - episode - for episode in metadata.episodes - if episode.released - and episode.released.date() == episode_date.date() - ] - if not episode: - logger.warning( - f"No episode found by {episode_date.date()} date for {parsed_data.get('title')} ({metadata.id}), {stream_id}" - ) - continue - imdb_episode = episode[0] - - logger.info( - f"Episode found by {episode_date} date for {parsed_data.get('title')} ({metadata.id})" - ) - episode_details = [ - { - "season_number": imdb_episode.season_number, - "episode_number": imdb_episode.episode_number, - "released": imdb_episode.released, - "title": imdb_episode.title, - } - ] - elif stream["meta_id"].startswith("tt"): - episode_details = [ - { - "season_number": 1, - "episode_number": 1, - } - ] - else: - logger.warning( - f"Failed to identify episode details for stream ID: {stream_id}. Deleting stream '{title}'" - ) - await TorrentStreams.find({"_id": stream_id}).delete() - continue - - # Update the TorrentStream with new episode details - await stream_collection.update_one( - {"_id": stream_id}, - {"$set": {"episode_files": [episode_details]}}, - ) - - logger.info(f"Updated stream ID: {stream_id} with parsed episode details.") - - logger.info("Series Streams Migration completed successfully.") - except Exception as e: - logger.error(f"Error during Series Streams Migration: {e}") - raise - - -async def cleanup_duplicate_metadata(): - """ - Migration script to clean up duplicate metadata and update related stream records. - Modified version for standalone MongoDB (no transactions). - """ - logging.info("Starting duplicate metadata cleanup migration") - - # Find all duplicate metadata with mf prefix - pipeline = [ - {"$match": {"_id": {"$regex": "^mf"}}}, - { - "$group": { - "_id": {"title": "$title", "year": "$year", "type": "$type"}, - "count": {"$sum": 1}, - "docs": { - "$push": { - "id": "$_id", - "total_streams": {"$ifNull": ["$total_streams", 0]}, - "last_updated_at": "$last_updated_at", - } - }, - } - }, - {"$match": {"count": {"$gt": 1}}}, - ] - - duplicate_groups = ( - await MediaFusionMetaData.get_motor_collection() - .aggregate(pipeline) - .to_list(None) - ) - - logging.info(f"Found {len(duplicate_groups)} groups of duplicates") - - for group in duplicate_groups: - group_key = group["_id"] - docs = group["docs"] - - # Sort by total_streams (desc) and last_updated_at (desc) - sorted_docs = sorted( - docs, - key=lambda x: ( - x.get("total_streams", 0), - x.get("last_updated_at", datetime.min), - ), - reverse=True, - ) - - # Keep the first one (most streams/latest update) - keep_id = sorted_docs[0]["id"] - remove_ids = [doc["id"] for doc in sorted_docs[1:]] - - logging.info(f"Processing group: {group_key}") - logging.info(f"Keeping: {keep_id}") - logging.info(f"Removing: {remove_ids}") - - # Update streams based on content type - content_type = group_key["type"] - - try: - # Update TorrentStreams for movies and series - if content_type in ["movie", "series"]: - result = await TorrentStreams.get_motor_collection().update_many( - {"meta_id": {"$in": remove_ids}}, {"$set": {"meta_id": keep_id}} - ) - logging.info(f"Updated {result.modified_count} torrent streams") - - # Update TVStreams for tv content - elif content_type == "tv": - result = await TVStreams.get_motor_collection().update_many( - {"meta_id": {"$in": remove_ids}}, {"$set": {"meta_id": keep_id}} - ) - logging.info(f"Updated {result.modified_count} TV streams") - - # Update streams count for the kept document - update_data = await update_meta_stream(keep_id) - - # Remove duplicate metadata last - result = await MediaFusionMetaData.get_motor_collection().delete_many( - {"_id": {"$in": remove_ids}} - ) - logging.info(f"Removed {result.deleted_count} duplicate metadata records") - - logging.info( - f"Updated stream count for {keep_id}: {update_data['total_streams']}" - ) - - except Exception as e: - logging.error(f"Error processing group {group_key}: {str(e)}") - continue - - -async def migrate_custom_flag(): - """Migration script to set is_custom field for existing documents.""" - # Update all documents with mf prefix to have is_custom = True - await MediaFusionMetaData.get_motor_collection().update_many( - {"_id": {"$regex": "^mf"}}, {"$set": {"is_custom": True}} - ) - - # Update all other documents to have is_custom = False - await MediaFusionMetaData.get_motor_collection().update_many( - {"_id": {"$not": {"$regex": "^mf"}}}, {"$set": {"is_custom": False}} - ) - await MediaFusionMetaData.get_motor_collection().create_index( - [("title", ASCENDING), ("year", ASCENDING), ("type", ASCENDING)], - unique=True, - partialFilterExpression={"is_custom": True}, - name="unique_title_year_type_for_mf_id", - ) - logging.info("Migration completed: is_custom field has been set for all documents") - - -async def main(): - logger.info("Starting migration") - await database.init(allow_index_dropping=True) - await cleanup_duplicate_metadata() - await migrate_custom_flag() - - await migrate_series_streams() - logger.info("Migration completed successfully") - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/utils/cleanup_torrents.py b/utils/cleanup_torrents.py deleted file mode 100644 index 59057af..0000000 --- a/utils/cleanup_torrents.py +++ /dev/null @@ -1,267 +0,0 @@ -import asyncio -import logging -import sys -from datetime import datetime, timezone, timedelta -from typing import Dict - -import PTT -from beanie import BulkWriter -from tqdm import tqdm - -from db import database -from db.models import ( - TorrentStreams, - MediaFusionMetaData, -) -from utils.parser import calculate_max_similarity_ratio, is_contain_18_plus_keywords - -BATCH_SIZE = 1000 # Adjust the batch size as needed -SKIP_TORRENTS = 0 # Skip the first n torrents -LAST_UPDATED = datetime.now(tz=timezone.utc) - timedelta(days=1) - -# Configure logging -logging.basicConfig( - level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" -) -logger = logging.getLogger(__name__) - - -async def cleanup_torrents(dry_run: bool = True) -> Dict[str, int]: - metrics = { - "total": 0, - "removed": 0, - "no_metadata": 0, - "valid": 0, - "ratio": 0, - "adult": 0, - "year_mismatch": 0, - "removed_non_imdb": 0, - "non_movie": 0, - } - lock = asyncio.Lock() - - # Create a cursor for efficient pagination - cursor = TorrentStreams.find(TorrentStreams.updated_at < LAST_UPDATED).sort( - -TorrentStreams.updated_at - ) - total_torrents = await cursor.count() - total_torrents = max(0, total_torrents - SKIP_TORRENTS) - - progress_bar = tqdm(total=total_torrents, desc="Processing torrents") - torrent_bulk_writer = BulkWriter() - movies_bulk_writer = BulkWriter() - series_bulk_writer = BulkWriter() - - cursor = cursor.skip(SKIP_TORRENTS) - - processed_count = 0 - async for torrent in cursor: - await process_torrent( - torrent, - dry_run, - metrics, - lock, - torrent_bulk_writer, - movies_bulk_writer, - series_bulk_writer, - ) - - processed_count += 1 - if processed_count >= BATCH_SIZE: - await torrent_bulk_writer.commit() - await movies_bulk_writer.commit() - await series_bulk_writer.commit() - torrent_bulk_writer.operations.clear() - movies_bulk_writer.operations.clear() - series_bulk_writer.operations.clear() - processed_count = 0 - - progress_bar.update(1) - progress_bar.set_postfix(metrics) - - # Commit any remaining operations - await torrent_bulk_writer.commit() - await movies_bulk_writer.commit() - await series_bulk_writer.commit() - - progress_bar.close() - return metrics - - -async def process_torrent( - torrent: TorrentStreams, - dry_run: bool, - metrics: Dict[str, int], - lock: asyncio.Lock, - torrent_bulk_writer: BulkWriter, - movies_bulk_writer: BulkWriter, - series_bulk_writer: BulkWriter, -) -> str: - async with lock: - metrics["total"] += 1 - - meta_data = await MediaFusionMetaData.find_one({"_id": torrent.meta_id}) - - if not meta_data: - await torrent.delete(bulk_writer=torrent_bulk_writer) - async with lock: - metrics["no_metadata"] += 1 - metrics["removed"] += 1 - logger.debug( - f"No metadata found for torrent: {torrent.torrent_name}, {torrent.meta_id}" - ) - return "removed" - - if meta_data.type == "movie": - meta_data_bulk_writer = movies_bulk_writer - else: - meta_data_bulk_writer = series_bulk_writer - - if ( - not dry_run - and torrent.meta_id.startswith("mf") - and torrent.source - in [ - "TamilBlasters", - "TamilMV", - ] - ): - await torrent.delete(bulk_writer=torrent_bulk_writer) - await meta_data.delete(bulk_writer=meta_data_bulk_writer) - async with lock: - metrics["removed"] += 1 - metrics["removed_non_imdb"] += 1 - logger.debug(f"Removed non-IMDb torrent: {torrent.torrent_name}") - return "removed" - - if is_contain_18_plus_keywords(torrent.torrent_name): - if not dry_run: - await torrent.delete(bulk_writer=torrent_bulk_writer) - async with lock: - metrics["removed"] += 1 - metrics["adult"] += 1 - logger.debug(f"Removed torrent due to adult content: {torrent.torrent_name}") - return "removed" - - parsed_data = PTT.parse_title(torrent.torrent_name, True) - - max_ratio = calculate_max_similarity_ratio( - parsed_data.get("title", ""), meta_data.title, meta_data.aka_titles - ) - expected_ratio = 70 if torrent.source in ["TamilMV", "TamilBlasters"] else 87 - if ( - max_ratio < expected_ratio - and torrent.meta_id.startswith("tt") - and not any(x in torrent.catalog for x in ["wwe_tgx", "ufc_tgx"]) - ): - if not dry_run: - await torrent.delete(bulk_writer=torrent_bulk_writer) - async with lock: - metrics["removed"] += 1 - metrics["ratio"] += 1 - logger.debug( - f"Removed torrent due to low similarity ratio: {parsed_data.get('title')} != {meta_data.title} (ratio: {max_ratio}%) (full title: {torrent.torrent_name})" - ) - return "removed" - - if torrent.meta_id.startswith("tt") and meta_data.type == "movie": - if parsed_data.get("seasons"): - if not dry_run: - await torrent.delete(bulk_writer=torrent_bulk_writer) - async with lock: - metrics["removed"] += 1 - metrics["non_movie"] += 1 - logger.debug( - f"Removed non-movie torrent: {torrent.torrent_name}, found season: {parsed_data['seasons']}. Expected movie." - ) - return "removed" - - if meta_data.type == "movie" and parsed_data.get("year") != meta_data.year: - if not dry_run: - await torrent.delete(bulk_writer=torrent_bulk_writer) - async with lock: - metrics["removed"] += 1 - metrics["year_mismatch"] += 1 - logger.debug( - f"Removed torrent due to year mismatch: {torrent.torrent_name} ({parsed_data.get('year')} != {meta_data.year}) ({meta_data.id}) ({meta_data.type})" - ) - return "removed" - - if ( - not dry_run - and torrent.meta_id.startswith("tt") - and "torrentgalaxy" not in torrent.source.lower() - ): - torrent.languages = parsed_data.get("languages") - torrent.resolution = parsed_data.get("resolution") - torrent.codec = parsed_data.get("codec") - torrent.quality = parsed_data.get("quality") - torrent.audio = ( - parsed_data.get("audio")[0] if parsed_data.get("audio") else None - ) - torrent.updated_at = datetime.now(tz=timezone.utc) - await torrent.save(bulk_writer=torrent_bulk_writer) - return "updated" - - torrent.updated_at = datetime.now(tz=timezone.utc) - await torrent.save(bulk_writer=torrent_bulk_writer) - return "updated" - - -async def fix_torrent_catalogs(): - torrents = TorrentStreams.find( - { - "source": {"$in": ["TamilMV", "TamilBlasters"]}, - } - ) - supported_categories = ["hdrip", "tcrip", "series", "dubbed", "old"] - async for torrent in torrents: - existing_catalogs = torrent.catalog - parsed_data = PTT.parse_title(torrent.torrent_name, True) - category = existing_catalogs[0].split("_")[1] if existing_catalogs else None - if category not in supported_categories: - if parsed_data.get("seasons"): - category = "series" - elif "dubbed" in torrent.torrent_name.lower(): - category = "dubbed" - elif ( - "pre" in torrent.torrent_name.lower() - or "cam" in torrent.torrent_name.lower() - ): - category = "tcrip" - else: - category = "hdrip" - new_catalogs = [] - for language in parsed_data.get("languages"): - if language == "English" and "eng" not in torrent.torrent_name.lower(): - continue - if category == "dubbed" and language == "English": - continue - new_catalogs.append(f"{language.lower()}_{category}") - if not new_catalogs: - continue - if "user_upload" in existing_catalogs: - new_catalogs.append("user_upload") - if set(existing_catalogs) == set(new_catalogs): - continue - logger.info( - f"Updating catalog for {torrent.id} Existing: {existing_catalogs} -> New: {new_catalogs}, {torrent.torrent_name}" - ) - torrent.catalog = new_catalogs - await torrent.save() - - -# Initialize the database connection and start the cleanup process -async def main(is_dry_run: bool = True, log_level: str = "INFO"): - logger.setLevel(log_level) - logger.info(f"Running cleanup with dry_run={is_dry_run}, log_level={log_level}") - await database.init() - - summary = await cleanup_torrents(is_dry_run) - logger.info(f"Summary: {summary}") - - -if __name__ == "__main__": - is_perform_cleanup = "--cleanup" in sys.argv - log_level = "DEBUG" if "--debug" in sys.argv else "INFO" - asyncio.run(main(not is_perform_cleanup, log_level))