-
Notifications
You must be signed in to change notification settings - Fork 2
/
scan_plex.py
116 lines (96 loc) · 4.51 KB
/
scan_plex.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
import os
import subprocess
import requests
import xml.etree.ElementTree as ET
import json
import argparse, asyncio, aioconsole
def get_plex_config():
"""Retrieve Plex configuration from plex.json."""
try:
with open('plex.json', 'r') as file:
config = json.load(file)
return config
except FileNotFoundError:
return None
except json.JSONDecodeError:
raise Exception("Error decoding plex.json.")
async def prompt_for_config():
"""Prompt user for Plex configuration details."""
plex_host = await aioconsole.ainput("Enter Plex host (e.g., localhost or IP of server): ")
plex_port = await aioconsole.ainput("Enter Plex port (e.g., 32400): ")
plex_token = await aioconsole.ainput("Enter Plex token: ")
return plex_host.strip(), plex_port.strip(), plex_token.strip()
def save_plex_config(plex_host, plex_port, plex_token):
config = {
'plex_url': f'http://{plex_host}:{plex_port}',
'plex_token': plex_token
}
with open('plex.json', 'w', encoding='utf-8') as file:
json.dump(config, file, indent=4)
async def ensure_plex_config():
"""Ensure plex.json exists and is properly configured."""
config = get_plex_config()
if config is None:
print("Configuration file not found or empty.")
plex_host, plex_port, plex_token = await prompt_for_config()
save_plex_config(plex_host, plex_port, plex_token)
else:
plex_host = config.get('plex_url', '').replace('http://', '').split(':')[0]
plex_port = config.get('plex_url', '').replace('http://', '').split(':')[1] if ':' in config.get('plex_url', '') else ''
plex_token = config.get('plex_token', '')
if not plex_host or not plex_port or not plex_token:
print("Some configuration details are missing or incomplete.")
plex_host, plex_port, plex_token = await prompt_for_config()
save_plex_config(plex_host, plex_port, plex_token)
return f'http://{plex_host}:{plex_port}', plex_token
def get_plex_library_sections(plex_url, plex_token):
"""Retrieve the list of library sections from Plex."""
url = f"{plex_url}/library/sections?X-Plex-Token={plex_token}"
response = requests.get(url)
if response.status_code != 200:
raise Exception(f"Failed to retrieve library sections: {response.status_code}")
try:
root = ET.fromstring(response.content)
sections = {}
for directory in root.findall('.//Directory'):
title = directory.get('title')
key = directory.get('key')
locations = [loc.get('path') for loc in directory.findall('./Location')]
sections[key] = {'title': title, 'locations': locations, "key": key}
return sections
except ET.ParseError as e:
raise Exception(f"Failed to parse library sections response: {e}")
def scan_plex_library_sections(src_dir, plex_url, plex_token):
if not os.path.isdir(src_dir):
raise ValueError(f"Source directory '{src_dir}' does not exist or is not a directory.")
subdirs = [os.path.join(src_dir, d) for d in os.listdir(src_dir) if os.path.isdir(os.path.join(src_dir, d))]
try:
sections = get_plex_library_sections(plex_url, plex_token)
except Exception as e:
print(f"Failed to retrieve library sections from Plex: {e}")
return
for subdir in subdirs:
section_id = None
for key, info in sections.items():
if 'locations' in info and subdir in info['locations']:
section_id = key
if not section_id:
print(f"No matching library section found in Plex for: {subdir}, please ensure directory exists and is mapped to a Plex library")
continue
curl_command = (
f'curl '
f'"{plex_url}/library/sections/{section_id}/refresh?X-Plex-Token={plex_token}"'
)
try:
subprocess.run(curl_command, shell=True, check=True)
print(f"Successfully scanned library section: {subdir}")
except subprocess.CalledProcessError as e:
print(f"Failed to scan library section: {subdir}. Error: {e}")
async def main():
parser = argparse.ArgumentParser(description='Scan Plex library sections.')
parser.add_argument('src_dir', type=str, help='Source directory to scan.')
args = parser.parse_args()
plex_url, plex_token = await ensure_plex_config()
scan_plex_library_sections(args.src_dir, plex_url, plex_token)
if __name__ == '__main__':
asyncio.run(main())