Merge pull request #984 from northernpowerhouse/epgsearchendpoint

Added EPG Search API Endpoint with regex filters.

CHANGELOG.md

@@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **EPG Program Search API** (`GET /api/epg/programs/search/`): a new endpoint for querying EPG program data with rich filtering and query support. - Thanks [@northernpowerhouse](https://github.com/northernpowerhouse)
+  - **Text search** on title and description with AND/OR boolean operators (case-insensitive), quoted phrase matching (`"Law and Order"` treats _and_ as literal text), parenthetical grouping (`(Newcastle OR NEW) AND (Villa OR AST)`), whole-word mode (`title_whole_words=true`), and regex mode (`title_regex=true`).
+  - **Time filters**: `airing_at` (programs live at a specific instant), `start_after`, `start_before`, `end_after`, `end_before`.
+  - **Relational filters**: `channel`, `channel_id`, `stream`, `group`, `epg_source`.
+  - **Field selection**: `fields=title,start_time,channels` returns only the requested keys; channel and stream data is skipped server-side (no wasted serialization) when not requested.
+  - **Pagination**: default 50 results per page, configurable up to 500 via `page_size`.
+  - **Access control**: results are scoped to channels the requesting user can access. `user_level` and the per-user adult-content filter are both enforced, matching the access model used by the M3U playlist, XC API, and stream proxy. Admin users receive unfiltered results.
+  - Full OpenAPI/Swagger documentation available.
+  - Requires `IsStandardUser` permission (user level ≥ 1).
 - **Per-user IP/CIDR network allowlists**. Admins can now assign IP address and CIDR range restrictions to individual user accounts via the API & XC tab on the user edit form. When a user has one or more allowed ranges configured, requests from IPs outside that list are rejected with `403 Forbidden` regardless of the global network access policy; if no ranges are configured, the user inherits global settings unchanged. The existing `network_access_allowed()` utility is extended with an optional `user` argument so the per-user check is enforced at all access-controlled entry points (M3U/EPG, Streams, XC API, UI) without duplicating IP-matching logic. Per-user restrictions are stored in `custom_properties['allowed_networks']`; no model changes or migrations are required. — Thanks [@sethwv](https://github.com/sethwv)
 - **`reset_user_network` management command**: `manage.py reset_user_network <username>` clears the per-user `allowed_networks` restriction for the specified account, restoring it to global-policy inheritance. Useful for recovering a user locked out by a misconfigured allowlist. — Thanks [@sethwv](https://github.com/sethwv)
 - **Auto-sync overhaul**: comprehensive rebuild of the M3U auto-channel-sync flow. Introduces a per-field override system, hide-from-output flag, range-bounded auto-numbering with a re-pack helper, multi-stream channel safety, multi-provider shared-range merging, and an across-the-board move from per-row writes to bulk operations. (Closes #1196) — Thanks [@CodeBormen](https://github.com/CodeBormen)

apps/epg/api_views.py

@@ -1,11 +1,14 @@
-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
 from rest_framework.views import APIView
 from rest_framework.decorators import action
 from drf_spectacular.utils import extend_schema, OpenApiParameter, inline_serializer
 from drf_spectacular.types import OpenApiTypes
+from django.db.models import Q
 from django.utils import timezone
+from django.utils.dateparse import parse_datetime
 from datetime import timedelta
 from .models import EPGSource, ProgramData, EPGData
 from .serializers import (
@@ -13,10 +16,12 @@ from .serializers import (
     ProgramDetailSerializer,
     EPGSourceSerializer,
     EPGDataSerializer,
+    ProgramSearchResultSerializer,
 )
 from .tasks import refresh_epg_data
 from apps.accounts.permissions import (
     Authenticated,
+    IsStandardUser,
     permission_classes_by_action,
     permission_classes_by_method,
 )
@@ -131,6 +136,214 @@ 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 (case-insensitive: `and`/`AND` both work)
+- Wrap phrases in double quotes to match them literally: `"Law and Order"`
+- Parenthetical grouping for complex queries: `(Newcastle OR NEW) AND (Villa OR AST)`
+- Regex pattern matching with `title_regex=true` (evaluated by the database engine)
+- 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`
+- Quoted phrase: `title="Law and Order"` (matches the exact phrase; 'and' is literal)
+- Mixed: `title="Law and Order" AND crime`
+- 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 (case-insensitive), quoted phrases, and parentheses. Double-quote a phrase to match it literally: `"Law and Order"`. Unquoted space-separated terms are matched as a phrase; use AND/OR to combine separate terms.',
+            ),
+            OpenApiParameter('title_regex', OpenApiTypes.BOOL, description='Enable regex matching for title (case-insensitive, default: false). e.g. `^The` matches titles starting with "The".'),
+            OpenApiParameter('title_whole_words', OpenApiTypes.BOOL, description='Match whole words only in title (default: false). e.g. `new` matches "Newcastle" normally but not with whole words enabled.'),
+            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, default: false).'),
+            OpenApiParameter('description_whole_words', OpenApiTypes.BOOL, description='Match whole words only in description (default: false). Same behaviour as title_whole_words.'),
+            OpenApiParameter('start_after', OpenApiTypes.DATETIME, description='Filter programs starting at or after this time. ISO 8601 format, e.g. `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, e.g. `2026-02-14T20:00:00Z`.'),
+            OpenApiParameter('channel', OpenApiTypes.STR, description='Filter by channel name (case-insensitive substring match). e.g. `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). e.g. `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. e.g. `title,start_time,end_time`.'),
+            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')
+
+        # Restrict results to programs on channels the user can access
+        user = request.user
+        if user.user_level < 10:
+            access_filter = Q(epg__channels__user_level__lte=user.user_level)
+            custom_props = user.custom_properties or {}
+            if custom_props.get('hide_adult_content', False):
+                access_filter &= Q(epg__channels__is_adult=False)
+            queryset = queryset.filter(access_filter).distinct()
+
+        # 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, 'user': request.user})
+        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
@@ -549,5 +762,183 @@ class CurrentProgramsAPIView(APIView):
                 program_data['channel_uuid'] = str(channel.uuid)
                 current_programs.append(program_data)
 
+
         return Response(current_programs, status=status.HTTP_200_OK)
 
+
+# ─────────────────────────────
+# 7) Program Search API
+# ─────────────────────────────
+
+
+def _build_q_object(field_name, term, use_regex=False, whole_words=False):
+    """
+    Build a single Q object for a search term.
+
+    Args:
+        field_name: Django ORM field name
+        term: Search term
+        use_regex: If True, use regex matching
+        whole_words: If True, use word boundary matching
+    """
+    term = term.strip()
+    if not term:
+        return Q()
+
+    if use_regex:
+        # Use Django's __iregex (case-insensitive regex)
+        return Q(**{f'{field_name}__iregex': term})
+    elif whole_words:
+        # Word boundary syntax differs by database engine:
+        #   PostgreSQL uses \y (or \m/\M); Python re (SQLite) uses \b.
+        from django.db import connection
+        if connection.vendor == 'postgresql':
+            boundary = r'\y'
+        else:
+            boundary = r'\b'
+        pattern = boundary + re.escape(term) + boundary
+        return Q(**{f'{field_name}__iregex': pattern})
+    else:
+        # Standard case-insensitive contains
+        return Q(**{f'{field_name}__icontains': term})
+
+
+def _parse_text_query(field_name, raw_value, use_regex=False, whole_words=False):
+    """
+    Parse a text search value with AND/OR operators (including nested groups, parentheses,
+    and quoted phrases) into a Q object.
+
+    Quoted phrases (double-quoted) are treated as atomic literals and are never split on
+    AND/OR. Outside of quotes, AND and OR are case-insensitive boolean operators.
+
+    Examples:
+        "sports AND football"                        → Q(field__icontains="sports") & Q(field__icontains="football")
+        "news or weather"                            → Q(field__icontains="news") | Q(field__icontains="weather")
+        '"Law and Order"'                            → Q(field__icontains="Law and Order")  [quoted phrase]
+        '"Law and Order" AND crime'                  → phrase AND bare term
+        "(Newcastle OR NEW) AND (Villa OR AST)"      → Grouped nested operations
+        "breaking news"                              → Q(field__icontains="breaking news")  [unquoted phrase match]
+
+    Args:
+        field_name: Django ORM field name to query
+        raw_value: Text value with optional AND/OR operators and quoted phrases
+        use_regex: If True, use regex matching instead of icontains
+        whole_words: If True, match whole words only (requires word boundaries)
+
+    Supports mixed operators evaluated left-to-right: "sports and football or basketball"
+    Supports nested groups: "(A OR B) AND (C OR D)"
+    """
+
+    # Step 1: Extract quoted phrases into a lookup dict and replace with opaque placeholders.
+    # This lets AND/OR detection be case-insensitive without splitting phrases like
+    # "Law and Order" that happen to contain conjunctions.
+    phrases = {}
+
+    def extract_quoted(text):
+        def replacer(m):
+            key = f'\x00P{len(phrases)}\x00'
+            phrases[key] = m.group(1)
+            return key
+        return re.sub(r'"([^"]*)"', replacer, text)
+
+    processed = extract_quoted(raw_value)
+
+    def build_q(token):
+        """Build a Q object, resolving any quoted-phrase placeholder first."""
+        return _build_q_object(field_name, phrases.get(token, token), use_regex, whole_words)
+
+    def parse_expression(expr):
+        """Recursively parse expression with parentheses support."""
+        expr = expr.strip()
+
+        # 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
+
+            group_q = parse_expression(expr[paren_start + 1:paren_end])
+
+            before_str = expr[:paren_start].rstrip()
+            after_str = expr[paren_end + 1:].lstrip()
+
+            # Operator connecting before-expression to the group (case-insensitive)
+            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 = '|'
+
+            # Operator connecting the group to the after-expression (case-insensitive)
+            after_op = '&'
+            after_upper = after_str.upper()
+            if after_upper.startswith('AND '):
+                after_str = after_str[4:].lstrip()
+            elif after_upper.startswith('OR '):
+                after_str = after_str[3:].lstrip()
+                after_op = '|'
+
+            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
+
+        # No parentheses: tokenize on AND/OR boundaries (case-insensitive).
+        # Quoted phrases have been replaced with placeholders and are never split.
+        tokens = []
+        operators = []
+        remaining = expr
+
+        while remaining:
+            upper = remaining.upper()
+            and_pos = upper.find(' AND ')
+            or_pos = upper.find(' OR ')
+
+            if and_pos == -1 and or_pos == -1:
+                tokens.append(remaining.strip())
+                break
+
+            if and_pos == -1:
+                pos, op, op_len = or_pos, '|', 4
+            elif or_pos == -1:
+                pos, op, op_len = and_pos, '&', 5
+            elif and_pos < or_pos:
+                pos, op, op_len = and_pos, '&', 5
+            else:
+                pos, op, op_len = or_pos, '|', 4
+
+            token = remaining[:pos].strip()
+            if token:
+                tokens.append(token)
+                operators.append(op)
+            remaining = remaining[pos + op_len:]
+
+        if not tokens:
+            return Q()
+
+        # Build Q chain
+        result = build_q(tokens[0])
+        for i, op in enumerate(operators):
+            next_q = build_q(tokens[i + 1])
+            if op == '&':
+                result = result & next_q
+            else:
+                result = result | next_q
+
+        return result
+
+    return parse_expression(processed)
+
+
+class ProgramSearchPagination(PageNumberPagination):
+    page_size = 50
+    page_size_query_param = 'page_size'
+    max_page_size = 500
+

apps/epg/serializers.py

@@ -1,7 +1,7 @@
 from core.utils import validate_flexible_url
 from rest_framework import serializers
 from .models import EPGSource, EPGData, ProgramData
-from apps.channels.models import Channel
+from apps.channels.models import Channel, Stream
 
 class EPGSourceSerializer(serializers.ModelSerializer):
     epg_data_count = serializers.SerializerMethodField()
@@ -170,3 +170,72 @@ class EPGDataSerializer(serializers.ModelSerializer):
             'icon_url',
             'epg_source',
         ]
+
+
+class ProgramSearchChannelSerializer(serializers.ModelSerializer):
+    """Lightweight channel info for search results."""
+    channel_group = serializers.CharField(source='channel_group.name', default=None)
+
+    class Meta:
+        model = Channel
+        fields = ['id', 'name', 'channel_number', 'channel_group', 'tvg_id']
+
+
+class ProgramSearchStreamSerializer(serializers.ModelSerializer):
+    """Lightweight stream info for search results."""
+    channel_group = serializers.CharField(source='channel_group.name', default=None)
+    m3u_account = serializers.CharField(source='m3u_account.name', default=None)
+
+    class Meta:
+        model = Stream
+        fields = ['id', 'name', 'channel_group', 'tvg_id', 'm3u_account']
+
+
+class ProgramSearchResultSerializer(serializers.ModelSerializer):
+    """Full program data with associated channels and streams for search results."""
+    epg_source = serializers.CharField(source='epg.epg_source.name', default=None)
+    epg_name = serializers.CharField(source='epg.name', default=None)
+    epg_icon_url = serializers.URLField(source='epg.icon_url', default=None)
+    channels = serializers.SerializerMethodField()
+    streams = serializers.SerializerMethodField()
+
+    class Meta:
+        model = ProgramData
+        fields = [
+            'id', 'title', 'sub_title', 'description',
+            'start_time', 'end_time', 'tvg_id', 'custom_properties',
+            'epg_source', 'epg_name', 'epg_icon_url',
+            'channels', 'streams',
+        ]
+
+    def _accessible_channels(self, obj):
+        """Return prefetched channels filtered to those the requesting user can access."""
+        channels = list(obj.epg.channels.all()) if obj.epg else []
+        user = self.context.get('user')
+        if user is None or user.user_level >= 10:
+            return channels
+        custom_props = user.custom_properties or {}
+        hide_adult = custom_props.get('hide_adult_content', False)
+        return [
+            ch for ch in channels
+            if ch.user_level <= user.user_level and (not hide_adult or not ch.is_adult)
+        ]
+
+    def get_channels(self, obj):
+        fields = self.context.get('fields')
+        if fields is not None and 'channels' not in fields:
+            return []
+        return ProgramSearchChannelSerializer(self._accessible_channels(obj), many=True).data
+
+    def get_streams(self, obj):
+        fields = self.context.get('fields')
+        if fields is not None and 'streams' not in fields:
+            return []
+        stream_ids = set()
+        streams = []
+        for ch in self._accessible_channels(obj):
+            for s in ch.streams.all():
+                if s.id not in stream_ids:
+                    stream_ids.add(s.id)
+                    streams.append(s)
+        return ProgramSearchStreamSerializer(streams, many=True).data

apps/epg/tests/test_epg_search_api.py

@@ -0,0 +1,339 @@
+from datetime import timedelta
+
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+from django.utils import timezone
+from rest_framework import status
+from rest_framework.test import APIClient
+
+from apps.epg.models import EPGData, EPGSource, ProgramData
+
+User = get_user_model()
+
+SEARCH_URL = "/api/epg/programs/search/"
+
+
+class ProgramSearchAPIViewTests(TestCase):
+    """Tests for the /api/epg/programs/search/ endpoint."""
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.epg_source = EPGSource.objects.create(name="Test Source", source_type="xmltv")
+        cls.epg = EPGData.objects.create(
+            tvg_id="test-tvg", name="Test EPG", epg_source=cls.epg_source
+        )
+
+        now = timezone.now().replace(microsecond=0)
+
+        # Premier League Football — airing now
+        cls.prog_football = ProgramData.objects.create(
+            epg=cls.epg,
+            title="Premier League Football",
+            description="Live coverage of the Premier League match.",
+            start_time=now - timedelta(minutes=30),
+            end_time=now + timedelta(hours=1),
+        )
+
+        # Newcastle vs Villa — also airing now
+        cls.prog_newcastle = ProgramData.objects.create(
+            epg=cls.epg,
+            title="Newcastle vs Villa",
+            description="Match highlights.",
+            start_time=now - timedelta(minutes=15),
+            end_time=now + timedelta(hours=2),
+        )
+
+        # BBC News — starts in 3 hours
+        cls.prog_news = ProgramData.objects.create(
+            epg=cls.epg,
+            title="BBC News at Ten",
+            description="The latest news from around the world.",
+            start_time=now + timedelta(hours=3),
+            end_time=now + timedelta(hours=4),
+        )
+
+        # Nature Documentary — starts in 5 hours
+        cls.prog_doc = ProgramData.objects.create(
+            epg=cls.epg,
+            title="Nature Documentary",
+            description="Exploring wildlife in the Amazon.",
+            start_time=now + timedelta(hours=5),
+            end_time=now + timedelta(hours=6),
+        )
+
+        cls.now = now
+        cls.user = User.objects.create_user(username="testuser", password="pass", user_level=1)
+
+    def setUp(self):
+        self.client = APIClient(REMOTE_ADDR="127.0.0.1")
+        self.client.force_authenticate(user=self.user)
+
+    # ------------------------------------------------------------------
+    # Response structure
+    # ------------------------------------------------------------------
+
+    def test_response_structure(self):
+        """Response includes pagination envelope and all expected program fields."""
+        response = self.client.get(SEARCH_URL, {"page_size": 1})
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        data = response.json()
+        self.assertIn("count", data)
+        self.assertIn("results", data)
+        self.assertIn("next", data)
+        self.assertIn("previous", data)
+
+        program = data["results"][0]
+        for field in ("id", "title", "start_time", "end_time", "tvg_id", "channels", "streams"):
+            self.assertIn(field, program)
+
+    # ------------------------------------------------------------------
+    # No filter — returns all programs
+    # ------------------------------------------------------------------
+
+    def test_no_filters_returns_all(self):
+        """Omitting filters returns all seeded programs."""
+        response = self.client.get(SEARCH_URL)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.json()["count"], 4)
+
+    # ------------------------------------------------------------------
+    # Title search
+    # ------------------------------------------------------------------
+
+    def test_title_simple_match(self):
+        """Simple title search returns matching programs."""
+        response = self.client.get(SEARCH_URL, {"title": "football"})
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.json()
+        self.assertEqual(data["count"], 1)
+        self.assertEqual(data["results"][0]["title"], "Premier League Football")
+
+    def test_title_multi_word_is_phrase_not_implicit_and(self):
+        """Space-separated words without AND/OR are matched as a phrase, not as implicit AND.
+
+        'Premier Football' contains both words from 'Premier League Football' but
+        not as a consecutive phrase — so it should return 0 results.
+        Use 'Premier AND Football' to match both words independently.
+        """
+        phrase = self.client.get(SEARCH_URL, {"title": "Premier Football"}).json()
+        explicit_and = self.client.get(SEARCH_URL, {"title": "Premier AND Football"}).json()
+        # Phrase match: "Premier Football" is not a substring of "Premier League Football"
+        self.assertEqual(phrase["count"], 0)
+        # Explicit AND: both words present → matches
+        self.assertEqual(explicit_and["count"], 1)
+
+    def test_title_quoted_phrase(self):
+        """Double-quoted phrases are matched literally; 'and'/'or' inside quotes are not operators.
+
+        This is the standard way to search for program titles that contain conjunctions,
+        e.g. "Law and Order". Without quotes, lowercase 'and'/'or' are still treated as
+        case-insensitive boolean operators — so quoting is the reliable way to do a phrase match.
+        """
+        prog = ProgramData.objects.create(
+            epg=self.epg,
+            title="Law and Order",
+            description="Crime drama.",
+            start_time=self.now + timedelta(hours=10),
+            end_time=self.now + timedelta(hours=11),
+        )
+        try:
+            # Quoted phrase → exact substring match → finds the program
+            quoted = self.client.get(SEARCH_URL, {"title": '"Law and Order"'}).json()
+            self.assertEqual(quoted["count"], 1)
+            self.assertEqual(quoted["results"][0]["title"], "Law and Order")
+
+            # Quoted phrase that is not a substring → no match
+            non_phrase = self.client.get(SEARCH_URL, {"title": '"Law Order"'}).json()
+            self.assertEqual(non_phrase["count"], 0)
+
+            # Mix: quoted phrase AND bare term present in the title → matches
+            mixed_match = self.client.get(SEARCH_URL, {"title": '"Law and Order" AND order'}).json()
+            self.assertEqual(mixed_match["count"], 1)
+
+            # Mix: quoted phrase AND bare term NOT in the title → no match
+            mixed_no_match = self.client.get(SEARCH_URL, {"title": '"Law and Order" AND crime'}).json()
+            self.assertEqual(mixed_no_match["count"], 0)
+        finally:
+            prog.delete()
+    def test_title_case_insensitive(self):
+        """Title search is case-insensitive."""
+        lower = self.client.get(SEARCH_URL, {"title": "football"}).json()
+        upper = self.client.get(SEARCH_URL, {"title": "FOOTBALL"}).json()
+        self.assertEqual(lower["count"], 1)
+        self.assertEqual(upper["count"], 1)
+        self.assertEqual(lower["results"][0]["title"], upper["results"][0]["title"])
+
+    def test_title_and_operator(self):
+        """AND operator (case-insensitive) requires both terms to be present in the title."""
+        upper = self.client.get(SEARCH_URL, {"title": "Premier AND League"}).json()
+        lower = self.client.get(SEARCH_URL, {"title": "Premier and League"}).json()
+        self.assertEqual(upper["count"], 1)
+        self.assertEqual(lower["count"], 1)
+        self.assertIn("Premier", upper["results"][0]["title"])
+
+    def test_title_or_operator(self):
+        """OR operator (case-insensitive) returns programs matching either term."""
+        upper = self.client.get(SEARCH_URL, {"title": "Newcastle OR Football"})
+        lower = self.client.get(SEARCH_URL, {"title": "Newcastle or Football"})
+        self.assertEqual(upper.status_code, status.HTTP_200_OK)
+        for response in (upper, lower):
+            titles = [r["title"] for r in response.json()["results"]]
+            self.assertIn("Premier League Football", titles)
+            self.assertIn("Newcastle vs Villa", titles)
+
+    def test_title_no_match_returns_empty(self):
+        """Search with no matching title returns empty results."""
+        response = self.client.get(SEARCH_URL, {"title": "XYZNONEXISTENT999"})
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.json()
+        self.assertEqual(data["count"], 0)
+        self.assertEqual(data["results"], [])
+
+    def test_title_whole_word_matching(self):
+        """title_whole_words=true matches complete words but not partial words."""
+        # 'new' as substring matches 'Newcastle vs Villa' and 'BBC News at Ten'
+        partial = self.client.get(SEARCH_URL, {"title": "new"}).json()
+        # Whole-word \bnew\b matches neither 'Newcastle' nor 'News' (partial matches)
+        whole_no_match = self.client.get(SEARCH_URL, {"title": "new", "title_whole_words": "true"}).json()
+        self.assertEqual(partial["count"], 2)
+        self.assertEqual(whole_no_match["count"], 0)
+
+        # 'football' is a complete word in 'Premier League Football' — must still match
+        whole_match = self.client.get(SEARCH_URL, {"title": "football", "title_whole_words": "true"}).json()
+        self.assertEqual(whole_match["count"], 1)
+        self.assertEqual(whole_match["results"][0]["title"], "Premier League Football")
+
+        # 'league' is also a complete word — and 'Premier AND league' with whole_words works
+        both_words = self.client.get(SEARCH_URL, {"title": "Premier AND league", "title_whole_words": "true"}).json()
+        self.assertEqual(both_words["count"], 1)
+
+    def test_title_regex(self):
+        """title_regex=true applies the query as a regex pattern."""
+        response = self.client.get(
+            SEARCH_URL, {"title": "^Premier", "title_regex": "true"}
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        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
+    # ------------------------------------------------------------------
+
+    def test_description_simple_match(self):
+        """Description search returns programs whose description contains the term."""
+        response = self.client.get(SEARCH_URL, {"description": "Premier League"})
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.json()
+        self.assertEqual(data["count"], 1)
+        self.assertEqual(data["results"][0]["title"], "Premier League Football")
+
+    def test_description_and_operator(self):
+        """AND operator in description requires both terms."""
+        response = self.client.get(SEARCH_URL, {"description": "latest AND news"})
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.json()
+        self.assertEqual(data["count"], 1)
+        self.assertEqual(data["results"][0]["title"], "BBC News at Ten")
+
+    # ------------------------------------------------------------------
+    # Time filters
+    # ------------------------------------------------------------------
+
+    def test_airing_at_returns_current_programs(self):
+        """airing_at returns programs where start_time <= t < end_time."""
+        ts = self.now.isoformat()
+        response = self.client.get(SEARCH_URL, {"airing_at": ts})
+        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)
+
+    def test_start_after_filter(self):
+        """start_after excludes programs that start before the given time."""
+        cutoff = (self.now + timedelta(hours=4)).isoformat()
+        response = self.client.get(SEARCH_URL, {"start_after": cutoff})
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        titles = [r["title"] for r in response.json()["results"]]
+        self.assertIn("Nature Documentary", titles)
+        self.assertNotIn("Premier League Football", titles)
+        self.assertNotIn("BBC News at Ten", titles)
+
+    def test_start_before_filter(self):
+        """start_before excludes programs that start after the given time."""
+        cutoff = (self.now + timedelta(hours=1)).isoformat()
+        response = self.client.get(SEARCH_URL, {"start_before": cutoff})
+        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("Nature Documentary", titles)
+
+    def test_invalid_datetime_returns_400(self):
+        """An unparseable datetime value returns a 400 error."""
+        response = self.client.get(SEARCH_URL, {"airing_at": "not-a-date"})
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn("error", response.json())
+
+    # ------------------------------------------------------------------
+    # Field selection
+    # ------------------------------------------------------------------
+
+    def test_field_selection_limits_response_keys(self):
+        """fields param restricts the keys present in each result."""
+        response = self.client.get(SEARCH_URL, {"fields": "title,start_time,end_time"})
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        for program in response.json()["results"]:
+            self.assertIn("title", program)
+            self.assertIn("start_time", program)
+            self.assertIn("end_time", program)
+            self.assertNotIn("description", program)
+            self.assertNotIn("channels", program)
+            self.assertNotIn("streams", program)
+
+    # ------------------------------------------------------------------
+    # Pagination
+    # ------------------------------------------------------------------
+
+    def test_pagination_page_size(self):
+        """page_size limits the number of results returned."""
+        response = self.client.get(SEARCH_URL, {"page_size": 2})
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.json()
+        self.assertEqual(len(data["results"]), 2)
+        self.assertEqual(data["count"], 4)
+        self.assertIsNotNone(data["next"])
+
+    def test_pagination_second_page(self):
+        """Page 2 returns different results from page 1."""
+        page1 = self.client.get(SEARCH_URL, {"page": 1, "page_size": 2}).json()
+        page2 = self.client.get(SEARCH_URL, {"page": 2, "page_size": 2}).json()
+        ids_p1 = {r["id"] for r in page1["results"]}
+        ids_p2 = {r["id"] for r in page2["results"]}
+        self.assertTrue(ids_p1.isdisjoint(ids_p2))
+
+    def test_page_size_capped_at_maximum(self):
+        """page_size beyond the 500 maximum is clamped, not rejected."""
+        response = self.client.get(SEARCH_URL, {"page_size": 10000})
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.json()
+        # All 4 seeded programs are returned — request was accepted and clamped
+        self.assertEqual(data["count"], 4)
+        self.assertEqual(len(data["results"]), 4)