463 lines
16 KiB
Python
463 lines
16 KiB
Python
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"""<!doctype html>
|
|
<html>
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1">
|
|
<title>{html.escape(title)}</title>
|
|
<link rel="stylesheet" href="{stylesheet}">
|
|
</head>
|
|
<body class="artifact">
|
|
{body}
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
def wrap_deck(body, title, request):
|
|
stylesheet = request.build_absolute_uri("/static/themes/deck.css")
|
|
return f"""<!doctype html>
|
|
<html>
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1">
|
|
<title>{html.escape(title)}</title>
|
|
<link rel="stylesheet" href="{stylesheet}">
|
|
</head>
|
|
<body class="deck">
|
|
<main class="slides">{body}</main>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
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"""<!doctype html>
|
|
<html>
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1">
|
|
<title>Raw source unavailable</title>
|
|
<style>
|
|
body {{ background: #f4f1ea; color: #1d2528; font-family: system-ui, sans-serif; margin: 0; padding: 2rem; }}
|
|
main {{ background: #fffaf0; border-radius: 16px; max-width: 42rem; padding: 1.4rem; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<main>
|
|
<h1>Raw source unavailable</h1>
|
|
<p>Daybook does not have preserved raw source for <strong>{html.escape(artifact.title)}</strong>.</p>
|
|
</main>
|
|
</body>
|
|
</html>
|
|
"""
|