(self, request)
| 4145 | ), |
| 4146 | ) |
| 4147 | def post(self, request): |
| 4148 | from apps.epg.models import EPGData, ProgramData |
| 4149 | from apps.epg.query_utils import parse_text_query |
| 4150 | |
| 4151 | data = request.data or {} |
| 4152 | tvg_id = str(data.get("tvg_id") or "").strip() |
| 4153 | mode = (data.get("mode") or "all").lower() |
| 4154 | title = (data.get("title") or "").strip() |
| 4155 | title_mode = (data.get("title_mode") or "exact").lower() |
| 4156 | description = (data.get("description") or "").strip() |
| 4157 | description_mode = (data.get("description_mode") or "contains").lower() |
| 4158 | try: |
| 4159 | limit = int(data.get("limit") or 25) |
| 4160 | except (TypeError, ValueError): |
| 4161 | limit = 25 |
| 4162 | limit = max(1, min(limit, 100)) |
| 4163 | |
| 4164 | if not title and not description: |
| 4165 | return Response({"error": "A title or description is required"}, status=status.HTTP_400_BAD_REQUEST) |
| 4166 | |
| 4167 | now = timezone.now() |
| 4168 | horizon = now + timedelta(days=7) |
| 4169 | |
| 4170 | if tvg_id: |
| 4171 | epg = EPGData.objects.filter(tvg_id=tvg_id).first() |
| 4172 | if not epg: |
| 4173 | return Response({"matches": [], "total": 0, "epg_found": False}) |
| 4174 | qs = ProgramData.objects.filter(epg=epg, end_time__gt=now, start_time__lte=horizon) |
| 4175 | else: |
| 4176 | qs = ProgramData.objects.filter(end_time__gt=now, start_time__lte=horizon) |
| 4177 | |
| 4178 | if title: |
| 4179 | if title_mode == "exact": |
| 4180 | qs = qs.filter(title__iexact=title) |
| 4181 | else: |
| 4182 | qs = qs.filter(parse_text_query( |
| 4183 | "title", title, |
| 4184 | use_regex=(title_mode == "regex"), |
| 4185 | whole_words=(title_mode == "search"), |
| 4186 | )) |
| 4187 | if description: |
| 4188 | qs = qs.filter(parse_text_query( |
| 4189 | "description", description, |
| 4190 | use_regex=(description_mode == "regex"), |
| 4191 | whole_words=(description_mode == "search"), |
| 4192 | )) |
| 4193 | |
| 4194 | qs = qs.distinct().order_by("start_time") |
| 4195 | |
| 4196 | # Apply "new" filter in Python (custom_properties JSON lookup), but only |
| 4197 | # over the bounded result set we already filtered down to. |
| 4198 | candidates = list(qs[:limit * 4]) # small overshoot to allow new-only filtering |
| 4199 | if mode == "new": |
| 4200 | candidates = [p for p in candidates if (p.custom_properties or {}).get("new")] |
| 4201 | |
| 4202 | total = len(candidates) |
| 4203 | candidates = candidates[:limit] |
| 4204 |
nothing calls this directly
no test coverage detected