Spaces:
Building
Building
| import libtorrent as lt | |
| import os | |
| import time | |
| import requests | |
| def torrent_info(link, search, ext): | |
| ses = lt.session() | |
| ses.listen_on(6881, 6891) | |
| downloads = [] | |
| # Path folder Hasil Download | |
| download_folder = '/home/user/app/Torrent' | |
| # Membuat folder hasil download jika belum ada | |
| if not os.path.exists(download_folder): | |
| os.makedirs(download_folder) | |
| print("Folder 'Hasil Download' berhasil dibuat") | |
| else: | |
| print("Folder 'Hasil Download' sudah ada") | |
| params = { | |
| 'save_path': download_folder, | |
| 'storage_mode': lt.storage_mode_t.storage_mode_sparse, | |
| 'auto_managed': True, | |
| 'duplicate_is_error': True | |
| } | |
| handle = lt.add_magnet_uri(ses, link, params) | |
| ses.start_dht() | |
| print("Mendapatkan metadata dari magnet link...") | |
| while (not handle.has_metadata()): | |
| time.sleep(1) | |
| print("Metadata diterima.") | |
| info = handle.get_torrent_info() | |
| file_list = info.files() | |
| info_array = [] | |
| for i, file in enumerate(file_list): | |
| if search in file.path and file.path.endswith(ext): | |
| info_array.append(f"{i}. {file.path}") | |
| return info_array | |
| def torrent_download(link, selected_files): | |
| ses = lt.session() | |
| ses.listen_on(6881, 6891) | |
| downloads = [] | |
| # Path folder Hasil Download | |
| download_folder = '/home/user/app/Torrent' | |
| # Membuat folder hasil download jika belum ada | |
| if not os.path.exists(download_folder): | |
| os.makedirs(download_folder) | |
| print("Folder 'Hasil Download' berhasil dibuat") | |
| else: | |
| print("Folder 'Hasil Download' sudah ada") | |
| params = { | |
| 'save_path': download_folder, | |
| 'storage_mode': lt.storage_mode_t.storage_mode_sparse, | |
| 'auto_managed': True, | |
| 'duplicate_is_error': True | |
| } | |
| handle = lt.add_magnet_uri(ses, link, params) | |
| ses.start_dht() | |
| print("Mendapatkan metadata dari magnet link...") | |
| while (not handle.has_metadata()): | |
| time.sleep(1) | |
| print("Metadata diterima.") | |
| info = handle.get_torrent_info() | |
| file_list = info.files() | |
| file_priorities = [0] * len(file_list) | |
| for index in selected_files: | |
| file_priorities[index] = 1 | |
| handle.prioritize_files(file_priorities) | |
| while (not handle.is_seed()): | |
| s = handle.status() | |
| state_str = ['queued', 'checking', 'downloading metadata', | |
| 'downloading', 'finished', 'seeding', 'allocating', 'checking fastresume'] | |
| print('%.2f%% complete (down: %.1f kb/s up: %.1f kB/s peers: %d) %s' % | |
| (s.progress * 100, s.download_rate / 1000, s.upload_rate / 1000, | |
| s.num_peers, state_str[s.state])) | |
| # Status bar | |
| progress_bar = '#' * int(s.progress * 100) | |
| remaining_space = ' ' * (100 - len(progress_bar)) | |
| status_bar = '[%s%s] %.2f%%' % (progress_bar, remaining_space, s.progress * 100) | |
| print(status_bar) | |
| if state_str[s.state] == 'finished': | |
| break | |
| time.sleep(1) | |
| # Return the path of the downloaded files | |
| downloaded_files = [] | |
| for index in selected_files: | |
| downloaded_files.append(os.path.join(download_folder, file_list[index].path)) | |
| return downloaded_files | |
| def torrent_group(x, input): | |
| info = "" | |
| text = f"""{input}""" | |
| lines = text.split("\n") | |
| numbers = [line.split(".")[0] for line in lines] | |
| # Kelompokkan hasil menjadi kelompok-kelompok yang terdiri dari 50 angka | |
| groups = [numbers[i:i+x] for i in range(0, len(numbers), x)] | |
| if x > 999: | |
| # Tampilkan hasil kelompokan dan tulis ke file | |
| for i, group in enumerate(groups): | |
| result = ", ".join(group) | |
| info += f"{result}\n" | |
| else: | |
| # Tampilkan hasil kelompokan dan tulis ke file | |
| for i, group in enumerate(groups): | |
| result = ", ".join(group) | |
| info += f"Kelompok {i+1}: {result}\n" | |
| return info | |