mirror of
https://github.com/Viren070/MediaFusion.git
synced 2025-12-01 23:21:11 +01:00
remove deprecated migration scripts
This commit is contained in:
mhdzumair
@@ -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())
|
||||
@@ -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))
|
||||
Reference in New Issue
Block a user