import html import json import mimetypes import secrets from pathlib import Path from urllib.parse import urlencode import markdown from django.conf import settings from django.http import FileResponse, Http404, HttpResponse, HttpResponseForbidden from django.shortcuts import get_object_or_404, render from django.template.defaultfilters import slugify from django.urls import reverse from django.utils.dateparse import parse_date from django.utils import timezone from django.views.decorators.clickjacking import xframe_options_sameorigin from rest_framework.parsers import FormParser, JSONParser, MultiPartParser from rest_framework.response import Response from rest_framework.views import APIView from .models import Artifact TEXT_KINDS = {Artifact.HTML_FRAGMENT, Artifact.HTML_FULL, Artifact.MARKDOWN, Artifact.DECK} DEFAULT_THEMES = { Artifact.HTML_FRAGMENT: "report", Artifact.MARKDOWN: "report", Artifact.DECK: "deck", } class ArtifactListCreateView(APIView): parser_classes = [JSONParser, FormParser, MultiPartParser] def get(self, request): artifacts = Artifact.objects.all() if persona := request.query_params.get("persona"): artifacts = artifacts.filter(persona=persona) if day := request.query_params.get("day"): artifacts = artifacts.filter(day=day) if tag := request.query_params.get("tag"): artifacts = artifacts.filter(tags__contains=[tag]) if kind := request.query_params.get("kind"): artifacts = artifacts.filter(kind=kind) if title_query := request.query_params.get("q"): artifacts = artifacts.filter(title__icontains=title_query) return Response([serialize_artifact(artifact, request) for artifact in artifacts]) def post(self, request): if not has_valid_token(request): return HttpResponseForbidden("Invalid or missing API token") if request.content_type and request.content_type.startswith("multipart/"): artifact = create_binary_artifact(request.data) if isinstance(artifact, Response): return artifact return Response(serialize_artifact(artifact, request), status=201) data = request.data kind = data.get("kind") if kind not in TEXT_KINDS: return Response({"detail": "kind must be html_fragment, html_full, markdown, or deck"}, status=400) required = ["persona", "title", "content"] missing = [field for field in required if not data.get(field)] if missing: return Response({"detail": f"Missing required field(s): {', '.join(missing)}"}, status=400) source_error = validate_source_fields(data) if source_error: return source_error artifact = create_text_artifact(data, request) return Response(serialize_artifact(artifact, request), status=201) class ArtifactDetailView(APIView): def get(self, request, artifact_id): artifact = get_object_or_404(Artifact, pk=artifact_id) return Response(serialize_artifact(artifact, request)) def patch(self, request, artifact_id): if not has_valid_token(request): return HttpResponseForbidden("Invalid or missing API token") artifact = get_object_or_404(Artifact, pk=artifact_id) data = request.data if "title" in data: if not data["title"]: return Response({"detail": "title cannot be blank"}, status=400) artifact.title = data["title"] if "tags" in data: artifact.tags = normalize_tags(data["tags"]) if "source_prompt" in data: artifact.source_prompt = data["source_prompt"] or "" if "source_type" in data: if data["source_type"] not in {choice[0] for choice in Artifact.SOURCE_TYPE_CHOICES}: return Response({"detail": "source_type must be agent, person, or system"}, status=400) artifact.source_type = data["source_type"] if "source_name" in data: artifact.source_name = data["source_name"] or "" if "source_id" in data: artifact.source_id = data["source_id"] or "" if "metadata" in data: if not isinstance(data["metadata"], dict): return Response({"detail": "metadata must be an object"}, status=400) artifact.metadata = data["metadata"] artifact.save( update_fields=[ "title", "tags", "source_prompt", "source_type", "source_name", "source_id", "metadata", ] ) return Response(serialize_artifact(artifact, request)) class CapabilitiesView(APIView): def get(self, request): return Response( { "artifact_kinds": [choice[0] for choice in Artifact.KIND_CHOICES], "filters": ["persona", "day", "tag", "kind", "q"], "views": ["rendered", "raw"], "raw_supported_for": [Artifact.MARKDOWN, Artifact.HTML_FULL], "source_types": [choice[0] for choice in Artifact.SOURCE_TYPE_CHOICES], "auth": "bearer_token", } ) def feed(request): persona = request.GET.get("persona", "") selected_kind = request.GET.get("kind", "") title_query = request.GET.get("q", "") today = timezone.localdate() selected_day = parse_date(request.GET.get("day", "")) or today selected_artifact_id = request.GET.get("artifact", "") preview_none = request.GET.get("preview") == "none" selected_view = "raw" if request.GET.get("view") == "raw" else "rendered" all_artifacts = Artifact.objects.all() artifacts = all_artifacts if persona: artifacts = artifacts.filter(persona=persona) if selected_kind: artifacts = artifacts.filter(kind=selected_kind) if title_query: artifacts = artifacts.filter(title__icontains=title_query) days = list(artifacts.dates("day", "day", order="DESC")) for day in (today, selected_day): if day not in days: days.append(day) days.sort(reverse=True) day_artifacts = artifacts.filter(day=selected_day).order_by("-created_at") selected_artifact = None if preview_none else day_artifacts.filter(pk=selected_artifact_id).first() toggle_query = {"day": selected_day.isoformat(), "view": "rendered"} if persona: toggle_query["persona"] = persona if selected_kind: toggle_query["kind"] = selected_kind if title_query: toggle_query["q"] = title_query if selected_artifact: toggle_query["artifact"] = selected_artifact.id rendered_url = f"?{urlencode(toggle_query)}" toggle_query["view"] = "raw" raw_url = f"?{urlencode(toggle_query)}" personas = all_artifacts.order_by("persona").values_list("persona", flat=True).distinct() return render( request, "artifacts/feed.html", { "artifacts": day_artifacts, "days": days, "personas": personas, "raw_url": raw_url, "rendered_url": rendered_url, "selected_artifact": selected_artifact, "selected_day": selected_day, "selected_kind": selected_kind, "selected_persona": persona, "selected_view": selected_view, "preview_none": preview_none, "today": today, "title_query": title_query, "kind_choices": Artifact.KIND_CHOICES, }, ) @xframe_options_sameorigin def artifact_view(request, artifact_id): artifact = get_object_or_404(Artifact, pk=artifact_id) if request.GET.get("raw") == "1": raw_path = raw_artifact_path(artifact) if raw_path and raw_path.exists(): return FileResponse(raw_path.open("rb"), content_type="text/plain; charset=utf-8") return HttpResponse(raw_unavailable_html(artifact), content_type="text/html") path = safe_artifact_path(artifact.path) if not path.exists(): raise Http404("Artifact file not found") content_type, _ = mimetypes.guess_type(path.name) return FileResponse(path.open("rb"), content_type=content_type or "application/octet-stream") def create_text_artifact(data, request): now = timezone.localtime() artifact_id = generate_id() persona = slugify(data["persona"]) or "unknown" kind = data["kind"] theme = data.get("theme") or DEFAULT_THEMES.get(kind, "base") title = data["title"] content = data["content"] directory = Path(persona) / now.date().isoformat() target_dir = settings.ARTIFACTS_ROOT / directory target_dir.mkdir(parents=True, exist_ok=True) if kind == Artifact.HTML_FULL: rendered = content elif kind == Artifact.MARKDOWN: rendered = wrap_html(markdown.markdown(content, extensions=["extra", "sane_lists"]), title, theme, request) (target_dir / f"{artifact_id}.md").write_text(content, encoding="utf-8") elif kind == Artifact.DECK: rendered = wrap_deck(content, title, request) else: rendered = wrap_html(content, title, theme, request) stored_path = directory / f"{artifact_id}.html" (settings.ARTIFACTS_ROOT / stored_path).write_text(rendered, encoding="utf-8") return Artifact.objects.create( id=artifact_id, persona=persona, title=title, kind=kind, theme=theme, day=now.date(), path=stored_path.as_posix(), original_kind=kind, tags=normalize_tags(data.get("tags", [])), source_prompt=data.get("source_prompt", ""), **source_fields(data, persona), ) def create_binary_artifact(data): uploaded = data.get("file") required = ["persona", "title"] missing = [field for field in required if not data.get(field)] if not uploaded: missing.append("file") if missing: return Response({"detail": f"Missing required field(s): {', '.join(missing)}"}, status=400) source_error = validate_source_fields(data) if source_error: return source_error now = timezone.localtime() artifact_id = generate_id() persona = slugify(data["persona"]) or "unknown" title = data["title"] content_type = getattr(uploaded, "content_type", "") or mimetypes.guess_type(uploaded.name)[0] or "" kind = binary_kind(content_type) directory = Path(persona) / now.date().isoformat() target_dir = settings.ARTIFACTS_ROOT / directory target_dir.mkdir(parents=True, exist_ok=True) suffix = safe_suffix(uploaded.name, content_type) stored_path = directory / f"{artifact_id}{suffix}" with (settings.ARTIFACTS_ROOT / stored_path).open("wb") as target: for chunk in uploaded.chunks(): target.write(chunk) return Artifact.objects.create( id=artifact_id, persona=persona, title=title, kind=kind, theme="", day=now.date(), path=stored_path.as_posix(), original_kind=kind, tags=normalize_tags(data.get("tags", [])), source_prompt=data.get("source_prompt", ""), **source_fields(data, persona), ) def wrap_html(body, title, theme, request): stylesheet = request.build_absolute_uri(f"/static/themes/{theme}.css") return f""" {html.escape(title)} {body} """ def wrap_deck(body, title, request): stylesheet = request.build_absolute_uri("/static/themes/deck.css") return f""" {html.escape(title)}
{body}
""" def generate_id(): while True: artifact_id = secrets.token_hex(4) if not Artifact.objects.filter(pk=artifact_id).exists(): return artifact_id def has_valid_token(request): expected = getattr(settings, "DAYBOOK_API_TOKEN", "") auth = request.headers.get("Authorization", "") return bool(expected) and auth == f"Bearer {expected}" def normalize_tags(tags): if isinstance(tags, str): return [tag.strip() for tag in tags.split(",") if tag.strip()] if isinstance(tags, list): return [str(tag).strip() for tag in tags if str(tag).strip()] return [] def validate_source_fields(data): if source_type := data.get("source_type"): if source_type not in {choice[0] for choice in Artifact.SOURCE_TYPE_CHOICES}: return Response({"detail": "source_type must be agent, person, or system"}, status=400) try: parse_metadata_value(data.get("metadata", {})) except ValueError as exc: return Response({"detail": str(exc)}, status=400) return None def source_fields(data, persona): return { "source_type": data.get("source_type") or Artifact.AGENT, "source_name": data.get("source_name") or persona, "source_id": data.get("source_id") or "", "metadata": parse_metadata_value(data.get("metadata", {})), } def parse_metadata_value(value): if value in (None, ""): return {} if isinstance(value, dict): return value if isinstance(value, str): try: parsed = json.loads(value) except json.JSONDecodeError as exc: raise ValueError("metadata must be an object") from exc if isinstance(parsed, dict): return parsed raise ValueError("metadata must be an object") def binary_kind(content_type): if content_type.startswith("image/"): return Artifact.IMAGE if content_type == "application/pdf": return Artifact.PDF return Artifact.BINARY def safe_suffix(filename, content_type): suffix = Path(filename).suffix.lower() if suffix and suffix[1:].isalnum() and len(suffix) <= 10: return suffix return mimetypes.guess_extension(content_type) or ".bin" def serialize_artifact(artifact, request): return { "id": artifact.id, "url": request.build_absolute_uri(reverse("artifact", args=[artifact.id])), "api_url": request.build_absolute_uri(reverse("artifact-detail-api", args=[artifact.id])), "raw_url": request.build_absolute_uri(f"{reverse('artifact', args=[artifact.id])}?raw=1"), "raw_available": raw_artifact_path(artifact) is not None, "persona": artifact.persona, "title": artifact.title, "kind": artifact.kind, "theme": artifact.theme, "day": artifact.day.isoformat(), "created_at": artifact.created_at.isoformat(), "tags": artifact.tags, "source_prompt": artifact.source_prompt, "source_type": artifact.source_type, "source_name": artifact.source_name, "source_id": artifact.source_id, "metadata": artifact.metadata, } def safe_artifact_path(relative_path): root = settings.ARTIFACTS_ROOT.resolve() path = (root / relative_path).resolve() if root not in path.parents and path != root: raise Http404("Invalid artifact path") return path def raw_artifact_path(artifact): path = safe_artifact_path(artifact.path) if artifact.kind == Artifact.MARKDOWN: return path.with_suffix(".md") if artifact.kind == Artifact.HTML_FULL: return path return None def raw_unavailable_html(artifact): return f""" Raw source unavailable

Raw source unavailable

Daybook does not have preserved raw source for {html.escape(artifact.title)}.

"""