Authenticate with Schedules Direct using stored credentials. Returns (token, None) on success or (None, Response) on failure.
(self, source)
| 119 | |
| 120 | |
| 121 | def _sd_authenticate(self, source): |
| 122 | """ |
| 123 | Authenticate with Schedules Direct using stored credentials. |
| 124 | Returns (token, None) on success or (None, Response) on failure. |
| 125 | """ |
| 126 | import hashlib |
| 127 | import requests as http_requests |
| 128 | from apps.epg.tasks import SD_BASE_URL |
| 129 | from core.utils import dispatcharr_http_headers |
| 130 | |
| 131 | username = (source.username or '').strip() |
| 132 | password = (source.password or '').strip() |
| 133 | if not username or not password: |
| 134 | return None, Response( |
| 135 | {"error": "Username and password are required."}, |
| 136 | status=status.HTTP_400_BAD_REQUEST |
| 137 | ) |
| 138 | |
| 139 | sha1_password = hashlib.sha1(password.encode('utf-8')).hexdigest() |
| 140 | try: |
| 141 | auth_response = http_requests.post( |
| 142 | f"{SD_BASE_URL}/token", |
| 143 | json={'username': username, 'password': sha1_password}, |
| 144 | headers=dispatcharr_http_headers(), |
| 145 | timeout=15, |
| 146 | ) |
| 147 | auth_response.raise_for_status() |
| 148 | token = auth_response.json().get('token') |
| 149 | if not token: |
| 150 | return None, Response( |
| 151 | {"error": "Authentication failed. Check your credentials."}, |
| 152 | status=status.HTTP_401_UNAUTHORIZED |
| 153 | ) |
| 154 | return token, None |
| 155 | except http_requests.exceptions.RequestException as e: |
| 156 | return None, Response( |
| 157 | {"error": f"Authentication failed: {str(e)}"}, |
| 158 | status=status.HTTP_502_BAD_GATEWAY |
| 159 | ) |
| 160 | |
| 161 | def _get_sd_reset_at(self, source): |
| 162 | """Retrieve stored reset timestamp from EPGSource model field.""" |
no test coverage detected