system-manager-wakasu
1#!/usr/bin/env python3
2"""
3Shared library for *arr (Sonarr, Radarr, Lidarr) automation scripts.
4
5Provides common functionality for API interaction, user confirmation,
6and output formatting across all *arr stack scripts.
7"""
8
9import argparse
10import sys
11from typing import Any, Dict, List, Optional
12
13import requests
14
15
16class ArrClient:
17 """Base client for *arr API interactions."""
18
19 def __init__(self, base_url: str, api_key: str):
20 """
21 Initialize the *arr API client.
22
23 Args:
24 base_url: Base URL of the *arr service
25 (e.g., http://localhost:8989)
26 api_key: API key for authentication
27 """
28 self.base_url = base_url.rstrip("/")
29 self.api_key = api_key
30 self.headers = {"X-Api-Key": api_key}
31
32 def get(
33 self, endpoint: str, params: Optional[Dict[str, Any]] = None
34 ) -> List[Dict[str, Any]] | Dict[str, Any]:
35 """
36 Make a GET request to the *arr API.
37
38 Args:
39 endpoint: API endpoint path (e.g., /api/v3/series)
40 params: Optional query parameters
41
42 Returns:
43 JSON response data
44
45 Raises:
46 SystemExit: If the request fails
47 """
48 url = f"{self.base_url}{endpoint}"
49
50 try:
51 response = requests.get(url, headers=self.headers, params=params)
52 response.raise_for_status()
53 return response.json()
54 except requests.exceptions.RequestException as e:
55 print(f"Error fetching from {endpoint}: {e}", file=sys.stderr)
56 if params:
57 print(f" Params: {params}", file=sys.stderr)
58 sys.exit(1)
59
60 def post(
61 self, endpoint: str, payload: Dict[str, Any]
62 ) -> Dict[str, Any]:
63 """
64 Make a POST request to the *arr API.
65
66 Args:
67 endpoint: API endpoint path (e.g., /api/v3/command)
68 payload: JSON payload to send
69
70 Returns:
71 JSON response data (empty dict on failure)
72 """
73 url = f"{self.base_url}{endpoint}"
74 headers = {**self.headers, "Content-Type": "application/json"}
75
76 try:
77 response = requests.post(url, headers=headers, json=payload)
78 response.raise_for_status()
79 return response.json()
80 except requests.exceptions.RequestException as e:
81 print(f"Error posting to {endpoint}: {e}", file=sys.stderr)
82 return {}
83
84
85def ask_confirmation(prompt: str) -> bool:
86 """
87 Ask user for yes/no confirmation.
88
89 Args:
90 prompt: Question to ask the user
91
92 Returns:
93 True if user confirms (y/yes), False otherwise (n/no)
94 """
95 while True:
96 response = input(f"{prompt} (y/n): ").lower().strip()
97 if response in ["y", "yes"]:
98 return True
99 elif response in ["n", "no"]:
100 return False
101 else:
102 print("Please answer 'y' or 'n'")
103
104
105def create_arr_parser(
106 service_name: str, description: str, default_port: int
107) -> argparse.ArgumentParser:
108 """
109 Create a standard argument parser for *arr scripts.
110
111 Args:
112 service_name: Name of the service (e.g., "Sonarr", "Radarr")
113 description: Description for the script
114 default_port: Default port for the service
115
116 Returns:
117 Configured ArgumentParser instance
118 """
119 parser = argparse.ArgumentParser(description=description)
120 parser.add_argument(
121 f"{service_name.lower()}_url",
122 metavar="url",
123 help=(
124 f"{service_name} base URL "
125 f"(e.g., http://localhost:{default_port})"
126 ),
127 )
128 parser.add_argument("api_key", help=f"{service_name} API key")
129 parser.add_argument(
130 "--dry-run",
131 action="store_true",
132 help="Show what would be changed without making changes",
133 )
134 parser.add_argument(
135 "--no-confirm",
136 "--yolo",
137 action="store_true",
138 dest="no_confirm",
139 help="Skip interactive confirmation (use with caution)",
140 )
141 return parser
142
143
144def print_separator(char: str = "=", width: int = 80) -> None:
145 """Print a separator line."""
146 print(char * width)
147
148
149def print_section_header(title: str) -> None:
150 """Print a section header with separators."""
151 print("\n" + "=" * 80)
152 print(title)
153 print("=" * 80)
154
155
156def print_item_list(
157 items: List[str], prefix: str, max_display: int = 5
158) -> None:
159 """
160 Print a list of items with optional truncation.
161
162 Args:
163 items: List of item names to display
164 prefix: Prefix message to show before the list
165 max_display: Maximum number of items to show before truncating
166 """
167 if not items:
168 return
169
170 count = len(items)
171 print(f"\n{prefix} ({count} items):")
172 for item in items[:max_display]:
173 print(f" - {item}")
174 if len(items) > max_display:
175 remaining = len(items) - max_display
176 print(f" ... and {remaining} more")
177
178
179def get_confirmation_decision(
180 args: argparse.Namespace, prompt: str
181) -> bool:
182 """
183 Determine whether to proceed based on dry-run, no-confirm, or user input.
184
185 Args:
186 args: Parsed command-line arguments
187 prompt: Confirmation prompt to show user
188
189 Returns:
190 True if should proceed, False otherwise
191 """
192 if args.dry_run:
193 print("\n[DRY RUN] Skipping actual operation")
194 return False
195 elif args.no_confirm:
196 print("\n[NO CONFIRM] Proceeding with operation...")
197 return True
198 else:
199 return ask_confirmation(prompt)
200
201
202def print_final_summary(
203 total: int,
204 processed: int,
205 skipped: int,
206 operation: str,
207 queue_note: bool = True,
208) -> None:
209 """
210 Print final summary of operations.
211
212 Args:
213 total: Total items that needed processing
214 processed: Number of items successfully processed
215 skipped: Number of items skipped
216 operation: Name of the operation (e.g., "Renamed", "Retagged")
217 queue_note: Whether to show the queue check note
218 """
219 print_section_header("FINAL SUMMARY")
220 print(f"\nItems processed: {total}")
221 print(f" - {operation}: {processed}")
222 print(f" - Skipped: {skipped}")
223
224 if processed > 0 and queue_note:
225 print(
226 f"\nNote: {operation} operations are queued. "
227 "Check the service's queue for progress."
228 )