Search available headends/lineups by country and postal code. Body: {"country": "USA", "postalcode": "07030"} Returns a flat list of lineups across all matching headends.
(self, request, pk=None)
| 407 | |
| 408 | @action(detail=True, methods=["post"], url_path="sd-lineups/search") |
| 409 | def sd_lineups_search(self, request, pk=None): |
| 410 | """ |
| 411 | Search available headends/lineups by country and postal code. |
| 412 | Body: {"country": "USA", "postalcode": "07030"} |
| 413 | Returns a flat list of lineups across all matching headends. |
| 414 | """ |
| 415 | import requests as http_requests |
| 416 | from apps.epg.tasks import SD_BASE_URL |
| 417 | from core.utils import dispatcharr_http_headers |
| 418 | |
| 419 | source = self.get_object() |
| 420 | if source.source_type != 'schedules_direct': |
| 421 | return Response( |
| 422 | {"error": "This action is only available for Schedules Direct sources."}, |
| 423 | status=status.HTTP_400_BAD_REQUEST |
| 424 | ) |
| 425 | |
| 426 | country = request.data.get('country', '').strip() |
| 427 | postalcode = request.data.get('postalcode', '').strip() |
| 428 | if not country or not postalcode: |
| 429 | return Response( |
| 430 | {"error": "country and postalcode are required."}, |
| 431 | status=status.HTTP_400_BAD_REQUEST |
| 432 | ) |
| 433 | |
| 434 | token, error = self._sd_authenticate(source) |
| 435 | if error: |
| 436 | return error |
| 437 | |
| 438 | headers = dispatcharr_http_headers(token=token) |
| 439 | |
| 440 | try: |
| 441 | resp = http_requests.get( |
| 442 | f"{SD_BASE_URL}/headends", |
| 443 | params={'country': country, 'postalcode': postalcode}, |
| 444 | headers=headers, |
| 445 | timeout=15, |
| 446 | ) |
| 447 | resp.raise_for_status() |
| 448 | headends = resp.json() |
| 449 | lineups = [] |
| 450 | for headend in headends: |
| 451 | for lineup in headend.get('lineups', []): |
| 452 | lineups.append({ |
| 453 | 'lineup': lineup.get('lineup'), |
| 454 | 'name': lineup.get('name'), |
| 455 | 'transport': headend.get('transport'), |
| 456 | 'location': headend.get('location'), |
| 457 | 'headend': headend.get('headend'), |
| 458 | }) |
| 459 | return Response({"lineups": lineups}) |
| 460 | except http_requests.exceptions.RequestException as e: |
| 461 | return Response( |
| 462 | {"error": f"Failed to search headends: {str(e)}"}, |
| 463 | status=status.HTTP_502_BAD_GATEWAY |
| 464 | ) |
| 465 | # ───────────────────────────── |
| 466 | # 2) Program API (CRUD) |
nothing calls this directly
no test coverage detected