(self, request)
| 4010 | ), |
| 4011 | ) |
| 4012 | def post(self, request): |
| 4013 | data = request.data or {} |
| 4014 | tvg_id = str(data.get("tvg_id") or "").strip() |
| 4015 | mode = (data.get("mode") or "all").lower() |
| 4016 | title = data.get("title") or "" |
| 4017 | title_mode = (data.get("title_mode") or "exact").lower() |
| 4018 | description = data.get("description") or "" |
| 4019 | description_mode = (data.get("description_mode") or "contains").lower() |
| 4020 | channel_id = data.get("channel_id") |
| 4021 | if mode not in ("all", "new"): |
| 4022 | return Response({"error": "mode must be 'all' or 'new'"}, status=status.HTTP_400_BAD_REQUEST) |
| 4023 | if title_mode not in ("exact", "contains", "search", "regex"): |
| 4024 | return Response({"error": "title_mode must be one of exact, contains, search, regex"}, status=status.HTTP_400_BAD_REQUEST) |
| 4025 | if description_mode not in ("contains", "search", "regex"): |
| 4026 | return Response({"error": "description_mode must be one of contains, search, regex"}, status=status.HTTP_400_BAD_REQUEST) |
| 4027 | if not title.strip() and not description.strip(): |
| 4028 | return Response({"error": "A title or description is required"}, status=status.HTTP_400_BAD_REQUEST) |
| 4029 | |
| 4030 | # Coerce / validate optional pinned channel |
| 4031 | pinned_channel_id = None |
| 4032 | if channel_id not in (None, ""): |
| 4033 | try: |
| 4034 | pinned_channel_id = int(channel_id) |
| 4035 | except (TypeError, ValueError): |
| 4036 | return Response({"error": "channel_id must be an integer"}, status=status.HTTP_400_BAD_REQUEST) |
| 4037 | from .models import Channel |
| 4038 | if not Channel.objects.filter(id=pinned_channel_id).exists(): |
| 4039 | return Response({"error": "channel_id does not exist"}, status=status.HTTP_400_BAD_REQUEST) |
| 4040 | |
| 4041 | rule_record = { |
| 4042 | "tvg_id": tvg_id, |
| 4043 | "mode": mode, |
| 4044 | "title": title, |
| 4045 | "title_mode": title_mode, |
| 4046 | "description": description, |
| 4047 | "description_mode": description_mode, |
| 4048 | } |
| 4049 | if pinned_channel_id is not None: |
| 4050 | rule_record["channel_id"] = pinned_channel_id |
| 4051 | |
| 4052 | rules = CoreSettings.get_dvr_series_rules() |
| 4053 | # Upsert by tvg_id + title so multiple rules can target the same channel |
| 4054 | existing = next( |
| 4055 | (r for r in rules if |
| 4056 | str(r.get("tvg_id") or "") == tvg_id and |
| 4057 | str(r.get("title") or "") == title), |
| 4058 | None |
| 4059 | ) |
| 4060 | if existing: |
| 4061 | existing.clear() |
| 4062 | existing.update(rule_record) |
| 4063 | else: |
| 4064 | rules.append(rule_record) |
| 4065 | CoreSettings.set_dvr_series_rules(rules) |
| 4066 | # Note: frontend calls the evaluate endpoint explicitly after creating |
| 4067 | # the rule, so do NOT fire evaluate_series_rules.delay() here to |
| 4068 | # avoid a race that creates duplicate recordings. |
| 4069 | return Response({"success": True, "rules": rules}) |
nothing calls this directly
no test coverage detected