mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-17 16:50:53 +00:00
fix(epg): fix and clean up EPG program search API and tests
- Fix test user: set user_level=1 (STANDARD), REMOTE_ADDR="127.0.0.1" to
satisfy IsStandardUser + network_access_allowed checks; move user creation
to setUpTestData (1 DB write per class instead of per test)
- Fix parenthesis parser: group_result was computed then silently discarded;
rewrite parse_expression to recursively compose Q objects correctly so
queries like "(Newcastle OR NEW) AND (Villa OR AST)" work as documented
- Fix field selection performance: resolve allowed fields before serialization
and pass as context; get_channels/get_streams short-circuit when not requested
- Move import re to top of file; remove mid-file alias regex_module
- Convert ProgramSearchAPIView standalone APIView into @action(detail=False)
on ProgramViewSet so the router registers programs/search/ before
programs/{pk}/ automatically, eliminating the URL ordering dependency
- Scope permission to the action directly via permission_classes=[IsStandardUser]
rather than adding "search" to the global permission_classes_by_action dict
- Add test_title_parenthetical_grouping to cover the fixed parser path
- Tighten existing test assertions (case-insensitive, whole-word, description
AND, page_size cap) to catch regressions rather than passing vacuously
Co-authored-by: Copilot <copilot@github.com>
This commit is contained in:
SergeantPanda
parent
672f135b51
commit
f89df3284e
4 changed files with 256 additions and 279 deletions
|
|
@ -1,6 +1,6 @@
|
|||
from django.urls import path, include
|
||||
from rest_framework.routers import DefaultRouter
|
||||
from .api_views import EPGSourceViewSet, ProgramViewSet, EPGGridAPIView, EPGImportAPIView, EPGDataViewSet, CurrentProgramsAPIView, ProgramSearchAPIView
|
||||
from .api_views import EPGSourceViewSet, ProgramViewSet, EPGGridAPIView, EPGImportAPIView, EPGDataViewSet, CurrentProgramsAPIView
|
||||
|
||||
app_name = 'epg'
|
||||
|
||||
|
|
@ -13,7 +13,6 @@ urlpatterns = [
|
|||
path('grid/', EPGGridAPIView.as_view(), name='epg_grid'),
|
||||
path('import/', EPGImportAPIView.as_view(), name='epg_import'),
|
||||
path('current-programs/', CurrentProgramsAPIView.as_view(), name='current_programs'),
|
||||
path('programs/search/', ProgramSearchAPIView.as_view(), name='program_search'),
|
||||
]
|
||||
|
||||
urlpatterns += router.urls
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import logging, os
|
||||
import logging, os, re
|
||||
from rest_framework import viewsets, status, serializers
|
||||
from rest_framework.pagination import PageNumberPagination
|
||||
from rest_framework.response import Response
|
||||
|
|
@ -21,6 +21,7 @@ from .serializers import (
|
|||
from .tasks import refresh_epg_data
|
||||
from apps.accounts.permissions import (
|
||||
Authenticated,
|
||||
IsStandardUser,
|
||||
permission_classes_by_action,
|
||||
permission_classes_by_method,
|
||||
)
|
||||
|
|
@ -135,6 +136,208 @@ class ProgramViewSet(viewsets.ModelViewSet):
|
|||
logger.debug("Listing all EPG programs.")
|
||||
return super().list(request, *args, **kwargs)
|
||||
|
||||
@extend_schema(
|
||||
summary="Search EPG programs",
|
||||
description="""
|
||||
**Advanced EPG program search with multiple filter types and complex query support.**
|
||||
|
||||
### Text Search Features
|
||||
|
||||
**Title and Description Search**:
|
||||
- Supports AND/OR logical operators
|
||||
- Parenthetical grouping for complex queries: `(Newcastle OR NEW) AND (Villa OR AST)`
|
||||
- Regex pattern matching with `title_regex=true`
|
||||
- Whole word matching with `title_whole_words=true` to avoid partial matches
|
||||
|
||||
**Examples**:
|
||||
- Simple: `title=football`
|
||||
- AND operator: `title=premier AND league`
|
||||
- OR operator: `title=Newcastle OR Villa`
|
||||
- Nested groups: `title=(Newcastle OR NEW) AND (Villa OR AST)`
|
||||
- Regex: `title=^Premier&title_regex=true` (programs starting with "Premier")
|
||||
- Whole words: `title=NEW&title_whole_words=true` (matches "NEW" but not "News")
|
||||
|
||||
### Time Filtering
|
||||
|
||||
**airing_at**: Find programs airing at a specific moment (start_time ≤ airing_at < end_time)
|
||||
|
||||
**Time ranges**: Use combinations of start_after, start_before, end_after, end_before
|
||||
|
||||
### Response Customization
|
||||
|
||||
**fields**: Comma-separated list to include only specific fields in response
|
||||
- Available: id, title, sub_title, description, start_time, end_time, tvg_id, custom_properties, epg_source, epg_name, epg_icon_url, channels, streams
|
||||
|
||||
### Pagination
|
||||
|
||||
- Default: 50 results per page
|
||||
- Maximum: 500 results per page
|
||||
- Use `page` and `page_size` parameters to navigate results
|
||||
""",
|
||||
parameters=[
|
||||
OpenApiParameter(
|
||||
'title',
|
||||
OpenApiTypes.STR,
|
||||
description='Title search query. Supports AND/OR operators and parentheses: `(Newcastle OR NEW) AND (Villa OR AST)`. Space-separated terms default to AND.',
|
||||
examples=[
|
||||
'football',
|
||||
'premier AND league',
|
||||
'Newcastle OR Villa',
|
||||
'(Newcastle OR NEW) AND (Villa OR AST)'
|
||||
]
|
||||
),
|
||||
OpenApiParameter('title_regex', OpenApiTypes.BOOL, description='Enable regex matching for title (case-insensitive). Example: `^The` matches titles starting with "The".'),
|
||||
OpenApiParameter('title_whole_words', OpenApiTypes.BOOL, description='Match whole words only in title. Prevents "NEW" from matching "News".'),
|
||||
OpenApiParameter(
|
||||
'description',
|
||||
OpenApiTypes.STR,
|
||||
description='Description search query. Same syntax and features as title search.'
|
||||
),
|
||||
OpenApiParameter('description_regex', OpenApiTypes.BOOL, description='Enable regex matching for description (case-insensitive).'),
|
||||
OpenApiParameter('description_whole_words', OpenApiTypes.BOOL, description='Match whole words only in description.'),
|
||||
OpenApiParameter('start_after', OpenApiTypes.DATETIME, description='Filter programs starting at or after this time. ISO 8601 format.', examples=['2026-02-14T18:00:00Z']),
|
||||
OpenApiParameter('start_before', OpenApiTypes.DATETIME, description='Filter programs starting at or before this time. ISO 8601 format.'),
|
||||
OpenApiParameter('end_after', OpenApiTypes.DATETIME, description='Filter programs ending at or after this time. ISO 8601 format.'),
|
||||
OpenApiParameter('end_before', OpenApiTypes.DATETIME, description='Filter programs ending at or before this time. ISO 8601 format.'),
|
||||
OpenApiParameter('airing_at', OpenApiTypes.DATETIME, description='Find programs airing at this exact moment (start_time ≤ airing_at < end_time). ISO 8601 format.', examples=['2026-02-14T20:00:00Z']),
|
||||
OpenApiParameter('channel', OpenApiTypes.STR, description='Filter by channel name (case-insensitive substring match).', examples=['BBC One', 'Sky Sports']),
|
||||
OpenApiParameter('channel_id', OpenApiTypes.INT, description='Filter by exact channel ID.'),
|
||||
OpenApiParameter('stream', OpenApiTypes.STR, description='Filter by stream name (case-insensitive substring match).'),
|
||||
OpenApiParameter('group', OpenApiTypes.STR, description='Filter by channel group or stream group name (case-insensitive substring match).', examples=['Sports', 'UK Channels']),
|
||||
OpenApiParameter('epg_source', OpenApiTypes.INT, description='Filter by EPG source ID.'),
|
||||
OpenApiParameter('fields', OpenApiTypes.STR, description='Comma-separated list of fields to include. Omit to return all fields.', examples=['title,start_time,end_time', 'title,description,channels']),
|
||||
OpenApiParameter('page', OpenApiTypes.INT, description='Page number for pagination (default: 1).'),
|
||||
OpenApiParameter('page_size', OpenApiTypes.INT, description='Results per page (default: 50, max: 500).'),
|
||||
],
|
||||
responses={200: ProgramSearchResultSerializer(many=True)},
|
||||
tags=['EPG'],
|
||||
)
|
||||
@action(detail=False, methods=['get'], url_path='search', permission_classes=[IsStandardUser])
|
||||
def search(self, request):
|
||||
params = request.query_params
|
||||
|
||||
# Build base queryset with prefetching
|
||||
queryset = ProgramData.objects.select_related(
|
||||
'epg', 'epg__epg_source'
|
||||
).prefetch_related(
|
||||
'epg__channels', 'epg__channels__channel_group',
|
||||
'epg__channels__streams', 'epg__channels__streams__channel_group',
|
||||
'epg__channels__streams__m3u_account',
|
||||
)
|
||||
|
||||
filters = Q()
|
||||
|
||||
# Text filters
|
||||
title = params.get('title')
|
||||
if title:
|
||||
title_regex = params.get('title_regex', '').lower() in ('true', '1', 'yes')
|
||||
title_whole_words = params.get('title_whole_words', '').lower() in ('true', '1', 'yes')
|
||||
filters &= _parse_text_query('title', title, use_regex=title_regex, whole_words=title_whole_words)
|
||||
|
||||
description = params.get('description')
|
||||
if description:
|
||||
desc_regex = params.get('description_regex', '').lower() in ('true', '1', 'yes')
|
||||
desc_whole_words = params.get('description_whole_words', '').lower() in ('true', '1', 'yes')
|
||||
filters &= _parse_text_query('description', description, use_regex=desc_regex, whole_words=desc_whole_words)
|
||||
|
||||
# Time filters with validation
|
||||
start_after = params.get('start_after')
|
||||
if start_after:
|
||||
dt = parse_datetime(start_after)
|
||||
if dt is None:
|
||||
return Response(
|
||||
{"error": f"Invalid datetime format for start_after: {start_after}. Use ISO 8601 format (e.g., 2026-02-14T18:00:00Z)"},
|
||||
status=status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
filters &= Q(start_time__gte=dt)
|
||||
|
||||
start_before = params.get('start_before')
|
||||
if start_before:
|
||||
dt = parse_datetime(start_before)
|
||||
if dt is None:
|
||||
return Response(
|
||||
{"error": f"Invalid datetime format for start_before: {start_before}. Use ISO 8601 format."},
|
||||
status=status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
filters &= Q(start_time__lte=dt)
|
||||
|
||||
end_after = params.get('end_after')
|
||||
if end_after:
|
||||
dt = parse_datetime(end_after)
|
||||
if dt is None:
|
||||
return Response(
|
||||
{"error": f"Invalid datetime format for end_after: {end_after}. Use ISO 8601 format."},
|
||||
status=status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
filters &= Q(end_time__gte=dt)
|
||||
|
||||
end_before = params.get('end_before')
|
||||
if end_before:
|
||||
dt = parse_datetime(end_before)
|
||||
if dt is None:
|
||||
return Response(
|
||||
{"error": f"Invalid datetime format for end_before: {end_before}. Use ISO 8601 format."},
|
||||
status=status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
filters &= Q(end_time__lte=dt)
|
||||
|
||||
airing_at = params.get('airing_at')
|
||||
if airing_at:
|
||||
dt = parse_datetime(airing_at)
|
||||
if dt is None:
|
||||
return Response(
|
||||
{"error": f"Invalid datetime format for airing_at: {airing_at}. Use ISO 8601 format."},
|
||||
status=status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
filters &= Q(start_time__lte=dt, end_time__gt=dt)
|
||||
|
||||
# Channel/stream filters
|
||||
channel = params.get('channel')
|
||||
if channel:
|
||||
filters &= Q(epg__channels__name__icontains=channel)
|
||||
|
||||
channel_id = params.get('channel_id')
|
||||
if channel_id:
|
||||
try:
|
||||
filters &= Q(epg__channels__id=int(channel_id))
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
stream = params.get('stream')
|
||||
if stream:
|
||||
filters &= Q(epg__channels__streams__name__icontains=stream)
|
||||
|
||||
group = params.get('group')
|
||||
if group:
|
||||
filters &= (
|
||||
Q(epg__channels__channel_group__name__icontains=group)
|
||||
| Q(epg__channels__streams__channel_group__name__icontains=group)
|
||||
)
|
||||
|
||||
epg_source = params.get('epg_source')
|
||||
if epg_source:
|
||||
try:
|
||||
filters &= Q(epg__epg_source__id=int(epg_source))
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
queryset = queryset.filter(filters).distinct().order_by('start_time')
|
||||
|
||||
# Resolve field selection before serialization so expensive methods can short-circuit
|
||||
requested_fields = params.get('fields')
|
||||
allowed = set(f.strip() for f in requested_fields.split(',')) if requested_fields else None
|
||||
|
||||
# Paginate
|
||||
paginator = ProgramSearchPagination()
|
||||
page = paginator.paginate_queryset(queryset, request)
|
||||
serializer = ProgramSearchResultSerializer(page, many=True, context={'fields': allowed})
|
||||
data = serializer.data
|
||||
|
||||
if allowed:
|
||||
data = [{k: v for k, v in item.items() if k in allowed} for item in data]
|
||||
|
||||
return paginator.get_paginated_response(data)
|
||||
|
||||
|
||||
# ─────────────────────────────
|
||||
# 3) EPG Grid View
|
||||
|
|
@ -560,7 +763,6 @@ class CurrentProgramsAPIView(APIView):
|
|||
# ─────────────────────────────
|
||||
# 7) Program Search API
|
||||
# ─────────────────────────────
|
||||
import re as regex_module
|
||||
|
||||
|
||||
def _build_q_object(field_name, term, use_regex=False, whole_words=False):
|
||||
|
|
@ -582,7 +784,7 @@ def _build_q_object(field_name, term, use_regex=False, whole_words=False):
|
|||
return Q(**{f'{field_name}__iregex': term})
|
||||
elif whole_words:
|
||||
# Use word boundary regex (case-insensitive)
|
||||
pattern = r'\b' + regex_module.escape(term) + r'\b'
|
||||
pattern = r'\b' + re.escape(term) + r'\b'
|
||||
return Q(**{f'{field_name}__iregex': pattern})
|
||||
else:
|
||||
# Standard case-insensitive contains
|
||||
|
|
@ -613,54 +815,45 @@ def _parse_text_query(field_name, raw_value, use_regex=False, whole_words=False)
|
|||
"""Recursively parse expression with parentheses support"""
|
||||
expr = expr.strip()
|
||||
|
||||
# Handle parentheses by recursively processing innermost groups
|
||||
while '(' in expr:
|
||||
# Handle parentheses: parse the innermost group into a Q object, then
|
||||
# recursively combine it with the expressions on either side.
|
||||
if '(' in expr:
|
||||
paren_start = expr.rfind('(')
|
||||
paren_end = expr.find(')', paren_start)
|
||||
if paren_end == -1:
|
||||
return Q() # Mismatched parentheses
|
||||
|
||||
# Recursively parse the group
|
||||
group_expr = expr[paren_start + 1:paren_end]
|
||||
group_result = parse_expression(group_expr)
|
||||
group_q = parse_expression(expr[paren_start + 1:paren_end])
|
||||
|
||||
# Replace group with placeholder to avoid re-parsing
|
||||
# We build up results as we go
|
||||
before = expr[:paren_start]
|
||||
after = expr[paren_end + 1:]
|
||||
before_str = expr[:paren_start].rstrip()
|
||||
after_str = expr[paren_end + 1:].lstrip()
|
||||
|
||||
# For now, we need to handle this differently - evaluate left to right
|
||||
# Extract operators around the group
|
||||
if before:
|
||||
before = before.rstrip()
|
||||
if before.upper().endswith(' AND'):
|
||||
before = before[:-5].rstrip()
|
||||
operator = ' AND '
|
||||
elif before.upper().endswith(' OR'):
|
||||
before = before[:-3].rstrip()
|
||||
operator = ' OR '
|
||||
else:
|
||||
operator = None
|
||||
else:
|
||||
operator = None
|
||||
before = None
|
||||
# Operator connecting before-expression to the group
|
||||
before_op = '&'
|
||||
if before_str.upper().endswith(' AND'):
|
||||
before_str = before_str[:-4].rstrip()
|
||||
elif before_str.upper().endswith(' OR'):
|
||||
before_str = before_str[:-3].rstrip()
|
||||
before_op = '|'
|
||||
|
||||
if after:
|
||||
after = after.lstrip()
|
||||
if after.upper().startswith('AND '):
|
||||
after = after[4:].lstrip()
|
||||
operator = ' AND ' if not operator else operator
|
||||
elif after.upper().startswith('OR '):
|
||||
after = after[3:].lstrip()
|
||||
operator = ' OR ' if not operator else operator
|
||||
else:
|
||||
after = None
|
||||
# Operator connecting the group to the after-expression
|
||||
after_op = '&'
|
||||
if after_str.upper().startswith('AND '):
|
||||
after_str = after_str[4:].lstrip()
|
||||
elif after_str.upper().startswith('OR '):
|
||||
after_str = after_str[3:].lstrip()
|
||||
after_op = '|'
|
||||
|
||||
# Reconstruct without parentheses for simpler processing
|
||||
parts = [before, after]
|
||||
expr = ' AND '.join(p for p in parts if p)
|
||||
result = group_q
|
||||
if before_str:
|
||||
before_q = parse_expression(before_str)
|
||||
result = (before_q | result) if before_op == '|' else (before_q & result)
|
||||
if after_str:
|
||||
after_q = parse_expression(after_str)
|
||||
result = (result | after_q) if after_op == '|' else (result & after_q)
|
||||
return result
|
||||
|
||||
# Tokenize on " AND " and " OR " boundaries (no parentheses now)
|
||||
# No parentheses: tokenize on " AND " and " OR " boundaries
|
||||
tokens = []
|
||||
operators = []
|
||||
remaining = expr
|
||||
|
|
@ -710,239 +903,3 @@ class ProgramSearchPagination(PageNumberPagination):
|
|||
page_size_query_param = 'page_size'
|
||||
max_page_size = 500
|
||||
|
||||
|
||||
class ProgramSearchAPIView(APIView):
|
||||
"""
|
||||
Advanced search for EPG programs with complex filtering capabilities.
|
||||
|
||||
Features:
|
||||
- **Text Search**: Title/description search with AND/OR operators, parenthetical grouping, regex, and whole-word matching
|
||||
- **Time Filtering**: Find programs airing at specific times or within time ranges
|
||||
- **Channel/Stream Filtering**: Filter by channel, stream, or group names
|
||||
- **Field Selection**: Customize response to include only needed fields
|
||||
- **Pagination**: Results paginated (default 50, max 500 per page)
|
||||
|
||||
Text Query Syntax:
|
||||
- Simple: `title=football`
|
||||
- AND: `title=premier AND league`
|
||||
- OR: `title=Newcastle OR Villa`
|
||||
- Nested: `title=(Newcastle OR NEW) AND (Villa OR AST)`
|
||||
- Regex: `title=^Premier&title_regex=true`
|
||||
- Whole words: `title=NEW&title_whole_words=true`
|
||||
|
||||
Examples:
|
||||
- Find matches airing now: `?title=football&airing_at=2026-02-14T20:00:00Z`
|
||||
- Complex search: `?title=(Newcastle OR NEW) AND (Villa OR AST)&airing_at=2026-02-14T20:00:00Z`
|
||||
- Channel-specific: `?channel=BBC One&start_after=2026-02-14T18:00:00Z`
|
||||
- Minimal response: `?title=sports&fields=title,start_time,end_time`
|
||||
"""
|
||||
pagination_class = ProgramSearchPagination
|
||||
|
||||
def get_permissions(self):
|
||||
try:
|
||||
return [
|
||||
perm() for perm in permission_classes_by_method[self.request.method]
|
||||
]
|
||||
except KeyError:
|
||||
return [Authenticated()]
|
||||
|
||||
@extend_schema(
|
||||
summary="Search EPG programs",
|
||||
description="""
|
||||
**Advanced EPG program search with multiple filter types and complex query support.**
|
||||
|
||||
### Text Search Features
|
||||
|
||||
**Title and Description Search**:
|
||||
- Supports AND/OR logical operators
|
||||
- Parenthetical grouping for complex queries: `(Newcastle OR NEW) AND (Villa OR AST)`
|
||||
- Regex pattern matching with `title_regex=true`
|
||||
- Whole word matching with `title_whole_words=true` to avoid partial matches
|
||||
|
||||
**Examples**:
|
||||
- Simple: `title=football`
|
||||
- AND operator: `title=premier AND league`
|
||||
- OR operator: `title=Newcastle OR Villa`
|
||||
- Nested groups: `title=(Newcastle OR NEW) AND (Villa OR AST)`
|
||||
- Regex: `title=^Premier&title_regex=true` (programs starting with "Premier")
|
||||
- Whole words: `title=NEW&title_whole_words=true` (matches "NEW" but not "News")
|
||||
|
||||
### Time Filtering
|
||||
|
||||
**airing_at**: Find programs airing at a specific moment (start_time ≤ airing_at < end_time)
|
||||
|
||||
**Time ranges**: Use combinations of start_after, start_before, end_after, end_before
|
||||
|
||||
### Response Customization
|
||||
|
||||
**fields**: Comma-separated list to include only specific fields in response
|
||||
- Available: id, title, sub_title, description, start_time, end_time, tvg_id, custom_properties, epg_source, epg_name, epg_icon_url, channels, streams
|
||||
|
||||
### Pagination
|
||||
|
||||
- Default: 50 results per page
|
||||
- Maximum: 500 results per page
|
||||
- Use `page` and `page_size` parameters to navigate results
|
||||
""",
|
||||
parameters=[
|
||||
OpenApiParameter(
|
||||
'title',
|
||||
OpenApiTypes.STR,
|
||||
description='Title search query. Supports AND/OR operators and parentheses: `(Newcastle OR NEW) AND (Villa OR AST)`. Space-separated terms default to AND.',
|
||||
examples=[
|
||||
'football',
|
||||
'premier AND league',
|
||||
'Newcastle OR Villa',
|
||||
'(Newcastle OR NEW) AND (Villa OR AST)'
|
||||
]
|
||||
),
|
||||
OpenApiParameter('title_regex', OpenApiTypes.BOOL, description='Enable regex matching for title (case-insensitive). Example: `^The` matches titles starting with "The".'),
|
||||
OpenApiParameter('title_whole_words', OpenApiTypes.BOOL, description='Match whole words only in title. Prevents "NEW" from matching "News".'),
|
||||
OpenApiParameter(
|
||||
'description',
|
||||
OpenApiTypes.STR,
|
||||
description='Description search query. Same syntax and features as title search.'
|
||||
),
|
||||
OpenApiParameter('description_regex', OpenApiTypes.BOOL, description='Enable regex matching for description (case-insensitive).'),
|
||||
OpenApiParameter('description_whole_words', OpenApiTypes.BOOL, description='Match whole words only in description.'),
|
||||
OpenApiParameter('start_after', OpenApiTypes.DATETIME, description='Filter programs starting at or after this time. ISO 8601 format.', examples=['2026-02-14T18:00:00Z']),
|
||||
OpenApiParameter('start_before', OpenApiTypes.DATETIME, description='Filter programs starting at or before this time. ISO 8601 format.'),
|
||||
OpenApiParameter('end_after', OpenApiTypes.DATETIME, description='Filter programs ending at or after this time. ISO 8601 format.'),
|
||||
OpenApiParameter('end_before', OpenApiTypes.DATETIME, description='Filter programs ending at or before this time. ISO 8601 format.'),
|
||||
OpenApiParameter('airing_at', OpenApiTypes.DATETIME, description='Find programs airing at this exact moment (start_time ≤ airing_at < end_time). ISO 8601 format.', examples=['2026-02-14T20:00:00Z']),
|
||||
OpenApiParameter('channel', OpenApiTypes.STR, description='Filter by channel name (case-insensitive substring match).', examples=['BBC One', 'Sky Sports']),
|
||||
OpenApiParameter('channel_id', OpenApiTypes.INT, description='Filter by exact channel ID.'),
|
||||
OpenApiParameter('stream', OpenApiTypes.STR, description='Filter by stream name (case-insensitive substring match).'),
|
||||
OpenApiParameter('group', OpenApiTypes.STR, description='Filter by channel group or stream group name (case-insensitive substring match).', examples=['Sports', 'UK Channels']),
|
||||
OpenApiParameter('epg_source', OpenApiTypes.INT, description='Filter by EPG source ID.'),
|
||||
OpenApiParameter('fields', OpenApiTypes.STR, description='Comma-separated list of fields to include. Omit to return all fields.', examples=['title,start_time,end_time', 'title,description,channels']),
|
||||
OpenApiParameter('page', OpenApiTypes.INT, description='Page number for pagination (default: 1).'),
|
||||
OpenApiParameter('page_size', OpenApiTypes.INT, description='Results per page (default: 50, max: 500).'),
|
||||
],
|
||||
responses={200: ProgramSearchResultSerializer(many=True)},
|
||||
tags=['EPG'],
|
||||
)
|
||||
def get(self, request, format=None):
|
||||
params = request.query_params
|
||||
|
||||
# Build base queryset with prefetching
|
||||
queryset = ProgramData.objects.select_related(
|
||||
'epg', 'epg__epg_source'
|
||||
).prefetch_related(
|
||||
'epg__channels', 'epg__channels__channel_group',
|
||||
'epg__channels__streams', 'epg__channels__streams__channel_group',
|
||||
'epg__channels__streams__m3u_account',
|
||||
)
|
||||
|
||||
filters = Q()
|
||||
|
||||
# Text filters
|
||||
title = params.get('title')
|
||||
if title:
|
||||
title_regex = params.get('title_regex', '').lower() in ('true', '1', 'yes')
|
||||
title_whole_words = params.get('title_whole_words', '').lower() in ('true', '1', 'yes')
|
||||
filters &= _parse_text_query('title', title, use_regex=title_regex, whole_words=title_whole_words)
|
||||
|
||||
description = params.get('description')
|
||||
if description:
|
||||
desc_regex = params.get('description_regex', '').lower() in ('true', '1', 'yes')
|
||||
desc_whole_words = params.get('description_whole_words', '').lower() in ('true', '1', 'yes')
|
||||
filters &= _parse_text_query('description', description, use_regex=desc_regex, whole_words=desc_whole_words)
|
||||
|
||||
# Time filters with validation
|
||||
start_after = params.get('start_after')
|
||||
if start_after:
|
||||
dt = parse_datetime(start_after)
|
||||
if dt is None:
|
||||
return Response(
|
||||
{"error": f"Invalid datetime format for start_after: {start_after}. Use ISO 8601 format (e.g., 2026-02-14T18:00:00Z)"},
|
||||
status=status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
filters &= Q(start_time__gte=dt)
|
||||
|
||||
start_before = params.get('start_before')
|
||||
if start_before:
|
||||
dt = parse_datetime(start_before)
|
||||
if dt is None:
|
||||
return Response(
|
||||
{"error": f"Invalid datetime format for start_before: {start_before}. Use ISO 8601 format."},
|
||||
status=status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
filters &= Q(start_time__lte=dt)
|
||||
|
||||
end_after = params.get('end_after')
|
||||
if end_after:
|
||||
dt = parse_datetime(end_after)
|
||||
if dt is None:
|
||||
return Response(
|
||||
{"error": f"Invalid datetime format for end_after: {end_after}. Use ISO 8601 format."},
|
||||
status=status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
filters &= Q(end_time__gte=dt)
|
||||
|
||||
end_before = params.get('end_before')
|
||||
if end_before:
|
||||
dt = parse_datetime(end_before)
|
||||
if dt is None:
|
||||
return Response(
|
||||
{"error": f"Invalid datetime format for end_before: {end_before}. Use ISO 8601 format."},
|
||||
status=status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
filters &= Q(end_time__lte=dt)
|
||||
|
||||
airing_at = params.get('airing_at')
|
||||
if airing_at:
|
||||
dt = parse_datetime(airing_at)
|
||||
if dt is None:
|
||||
return Response(
|
||||
{"error": f"Invalid datetime format for airing_at: {airing_at}. Use ISO 8601 format."},
|
||||
status=status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
filters &= Q(start_time__lte=dt, end_time__gt=dt)
|
||||
|
||||
# Channel/stream filters
|
||||
channel = params.get('channel')
|
||||
if channel:
|
||||
filters &= Q(epg__channels__name__icontains=channel)
|
||||
|
||||
channel_id = params.get('channel_id')
|
||||
if channel_id:
|
||||
try:
|
||||
filters &= Q(epg__channels__id=int(channel_id))
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
stream = params.get('stream')
|
||||
if stream:
|
||||
filters &= Q(epg__channels__streams__name__icontains=stream)
|
||||
|
||||
group = params.get('group')
|
||||
if group:
|
||||
filters &= (
|
||||
Q(epg__channels__channel_group__name__icontains=group)
|
||||
| Q(epg__channels__streams__channel_group__name__icontains=group)
|
||||
)
|
||||
|
||||
epg_source = params.get('epg_source')
|
||||
if epg_source:
|
||||
try:
|
||||
filters &= Q(epg__epg_source__id=int(epg_source))
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
queryset = queryset.filter(filters).distinct().order_by('start_time')
|
||||
|
||||
# Paginate
|
||||
paginator = self.pagination_class()
|
||||
page = paginator.paginate_queryset(queryset, request)
|
||||
serializer = ProgramSearchResultSerializer(page, many=True)
|
||||
data = serializer.data
|
||||
|
||||
# Field selection
|
||||
requested_fields = params.get('fields')
|
||||
if requested_fields:
|
||||
allowed = set(f.strip() for f in requested_fields.split(','))
|
||||
data = [{k: v for k, v in item.items() if k in allowed} for item in data]
|
||||
|
||||
return paginator.get_paginated_response(data)
|
||||
|
||||
|
|
|
|||
|
|
@ -209,10 +209,16 @@ class ProgramSearchResultSerializer(serializers.ModelSerializer):
|
|||
]
|
||||
|
||||
def get_channels(self, obj):
|
||||
fields = self.context.get('fields')
|
||||
if fields is not None and 'channels' not in fields:
|
||||
return []
|
||||
channels = obj.epg.channels.all() if obj.epg else []
|
||||
return ProgramSearchChannelSerializer(channels, many=True).data
|
||||
|
||||
def get_streams(self, obj):
|
||||
fields = self.context.get('fields')
|
||||
if fields is not None and 'streams' not in fields:
|
||||
return []
|
||||
channels = obj.epg.channels.all() if obj.epg else []
|
||||
stream_ids = set()
|
||||
streams = []
|
||||
|
|
|
|||
|
|
@ -160,6 +160,21 @@ class ProgramSearchAPIViewTests(TestCase):
|
|||
for program in response.json()["results"]:
|
||||
self.assertTrue(program["title"].startswith("Premier"))
|
||||
|
||||
def test_title_parenthetical_grouping(self):
|
||||
"""Parenthetical groups with AND/OR are evaluated correctly."""
|
||||
# (Newcastle OR Football) AND (Villa OR League) should match both seeded programs:
|
||||
# "Premier League Football" matches Football AND League
|
||||
# "Newcastle vs Villa" matches Newcastle AND Villa
|
||||
response = self.client.get(
|
||||
SEARCH_URL, {"title": "(Newcastle OR Football) AND (Villa OR League)"}
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
titles = {r["title"] for r in response.json()["results"]}
|
||||
self.assertIn("Premier League Football", titles)
|
||||
self.assertIn("Newcastle vs Villa", titles)
|
||||
self.assertNotIn("BBC News at Ten", titles)
|
||||
self.assertNotIn("Nature Documentary", titles)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Description search
|
||||
# ------------------------------------------------------------------
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue