from django.shortcuts import get_object_or_404 from django.utils import timezone from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.permissions import IsAdminUser from tracker.api.envelope import EnvelopeMixin, ok from tracker.models import Site def _norm_domains(raw) -> list[str]: """域名归一:去协议/端口/路径、转小写、去空去重。""" if isinstance(raw, str) or not isinstance(raw, (list, tuple)): return [] # 标量/字符串(会逐字符迭代成单字符域名)→ 走 invalid_domains 400 out = [] for d in raw: d = str(d).strip().lower() for pre in ("https://", "http://"): if d.startswith(pre): d = d[len(pre):] d = d.split("/")[0].split(":")[0] if d and d not in out: out.append(d) return out def _to_bool(v) -> bool: """宽松布尔解析:bool 直通,"true"/"false" 字符串按字面(bool("false") 是 True, 对 allow_no_origin 这类放宽开关方向不利),其余(数字/未知串/None)一律 False。""" if isinstance(v, bool): return v if isinstance(v, str) and v.lower() in ("true", "false"): return v.lower() == "true" return False def _serialize(site: Site) -> dict: # USE_TZ=True 下 isoformat() 是 UTC("+00:00"),前端直接切串显示会与 # Dashboard 的上海自然日口径同屏矛盾——统一按上海时区序列化。 return {"id": site.pk, "name": site.name, "site_id": site.site_id, "domains": site.domains, "allow_no_origin": site.allow_no_origin, "is_active": site.is_active, "created_at": timezone.localtime(site.created_at).isoformat()} class SiteListView(EnvelopeMixin, APIView): def get(self, request): return ok([_serialize(s) for s in Site.objects.order_by("-created_at")]) def post(self, request): data = request.data if isinstance(request.data, dict) else {} # JSON 数组等畸形 body → 空表(.get 会 500) name = data.get("name") if not isinstance(name, str) or not name.strip(): return Response({"code": 400, "data": None, "message": "invalid_name"}, status=400) domains = _norm_domains(data.get("domains")) if not domains: return Response({"code": 400, "data": None, "message": "invalid_domains"}, status=400) site = Site.objects.create(name=name.strip()[:100], domains=domains, allow_no_origin=_to_bool(data.get("allow_no_origin"))) return ok(_serialize(site), "created") class SiteDetailView(EnvelopeMixin, APIView): def get_permissions(self): # DELETE(软删)仅 admin;PATCH 登录成员即可(settings 默认 IsAuthenticated;无 detail GET,前端走列表)。 # 在 initial() 阶段判权限:403 先于 404,且不泄露站点存在性。 if self.request.method == "DELETE": return [IsAdminUser()] return super().get_permissions() def _get(self, request, pk): return get_object_or_404(Site, pk=pk) def patch(self, request, pk): site = self._get(request, pk) data = request.data if isinstance(request.data, dict) else {} if "name" in data: name = data["name"] if not isinstance(name, str) or not name.strip(): return Response({"code": 400, "data": None, "message": "invalid_name"}, status=400) site.name = name.strip()[:100] if "domains" in data: domains = _norm_domains(data["domains"]) if not domains: return Response({"code": 400, "data": None, "message": "invalid_domains"}, status=400) site.domains = domains if "allow_no_origin" in data: site.allow_no_origin = _to_bool(data["allow_no_origin"]) if "is_active" in data: site.is_active = _to_bool(data["is_active"]) site.save() return ok(_serialize(site)) def delete(self, request, pk): # 软删:默认停用即可(上报侧 collect 见 is_active=False 即 403,Task7 联动)。 # 物理清除走未来管理命令分批删——大站单语句级联删除有锁表/binlog 风暴风险, # CASCADE 仅作完整性兜底,不在请求路径触发。 site = self._get(request, pk) site.is_active = False site.save(update_fields=["is_active", "updated_at"]) return ok(_serialize(site), "deactivated")