feat: daily Markdown export admin action

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-05 15:03:39 +08:00
parent 9fa698b3ea
commit 9ec684dfe2
3 changed files with 239 additions and 0 deletions

View File

@@ -2,8 +2,10 @@ from datetime import date
from django.contrib import admin from django.contrib import admin
from django.db.models import Prefetch from django.db.models import Prefetch
from django.http import HttpResponse
from django.template.response import TemplateResponse from django.template.response import TemplateResponse
from openclaw.export import export_daily_markdown
from openclaw.models import Session, Message, ToolCall from openclaw.models import Session, Message, ToolCall
@@ -100,8 +102,17 @@ def daily_conversation_view(request):
return TemplateResponse(request, "admin/openclaw/daily_view.html", context) return TemplateResponse(request, "admin/openclaw/daily_view.html", context)
@admin.action(description="Export selected sessions to Markdown")
def export_to_markdown(modeladmin, request, queryset):
md, filename = export_daily_markdown(queryset)
response = HttpResponse(md, content_type="text/markdown")
response["Content-Disposition"] = f'attachment; filename="{filename}"'
return response
@admin.register(Session) @admin.register(Session)
class SessionAdmin(admin.ModelAdmin): class SessionAdmin(admin.ModelAdmin):
actions = [export_to_markdown]
list_display = ( list_display = (
"session_id", "session_id",
"agent_name", "agent_name",

126
src/openclaw/export.py Normal file
View File

@@ -0,0 +1,126 @@
from openclaw.models import Session, Message
def generate_markdown_report(messages, date_str, sessions=None):
"""Generate a markdown daily report.
Args:
messages: QuerySet of Message objects, ordered by timestamp.
date_str: Date string for the report header (YYYY-MM-DD).
sessions: Optional dict of session_id -> Session instance for metadata.
"""
if sessions is None:
sessions = {}
lines = [f"# Daily Report: {date_str}", ""]
# Group messages by session
session_messages = {}
for msg in messages:
sid = msg.session.session_id
if sid not in session_messages:
session_messages[sid] = []
# Auto-populate sessions dict from FK
if sid not in sessions:
sessions[sid] = msg.session
session_messages[sid].append(msg)
for session_id, msgs in session_messages.items():
session = sessions.get(session_id)
if session:
lines.append(
f"## Session: {session_id} (Agent: {session.agent_name})"
)
lines.append(
f"**Model**: {session.model_id or 'N/A'} | "
f"**Token**: {session.total_tokens:,}"
)
else:
lines.append(f"## Session: {session_id}")
lines.append("")
for msg in msgs:
role_label = {
"user": "User",
"assistant": "Assistant",
"toolResult": "Tool Result",
}.get(msg.role, msg.role)
time_str = msg.timestamp.strftime("%H:%M")
# For assistant messages, check raw_content for tool_call mentions
if msg.role == "assistant":
tool_info = _extract_tool_info(msg.raw_content)
lines.append(f"### {time_str} {role_label}")
if msg.content_text:
lines.append("")
lines.append(msg.content_text)
for tool in tool_info:
lines.append("")
lines.append(
f"**{time_str} {role_label} -> [Tool: {tool['name']}]**"
)
lines.append("")
lines.append(f"`{tool.get('arguments', '')}`")
if tool.get("result"):
lines.append("")
lines.append(f'**Result**: {tool["result"]}')
elif msg.role == "toolResult":
continue # toolResult handled inline with assistant
else:
lines.append(f"### {time_str} {role_label}")
if msg.content_text:
lines.append("")
lines.append(msg.content_text)
lines.append("")
lines.append("---")
lines.append("")
return "\n".join(lines)
def _extract_tool_info(raw_content):
"""Extract tool call info from message raw_content JSON."""
tools = []
if isinstance(raw_content, list):
for block in raw_content:
if isinstance(block, dict) and block.get("type") == "toolCall":
tool_name = block.get("tool_name") or block.get("name", "unknown")
args = block.get("arguments", {})
if isinstance(args, str):
args_str = args[:200]
else:
args_str = str(args)[:200]
tools.append({
"name": tool_name,
"arguments": args_str,
"result": "", # Will be filled later from toolResult
})
return tools
def export_daily_markdown(sessions_queryset):
"""Generate a markdown file from a QuerySet of Session objects.
Returns (markdown_string, filename).
Fetches all messages for the sessions.
"""
messages = Message.objects.filter(
session__in=sessions_queryset
).order_by("timestamp")
sessions_map = {s.session_id: s for s in sessions_queryset}
# Determine date from first session start_time
first_session = sessions_queryset.order_by("start_time").first()
if first_session and first_session.start_time:
date_str = first_session.start_time.strftime("%Y-%m-%d")
else:
date_str = "export"
md = generate_markdown_report(messages, date_str, sessions_map)
filename = f"daily-report-{date_str}.md"
return md, filename

102
tests/test_admin_export.py Normal file
View File

@@ -0,0 +1,102 @@
"""Tests for Markdown export functionality."""
from datetime import datetime, timezone
import pytest
from openclaw.models import Session, Message
from openclaw.export import generate_markdown_report
@pytest.fixture
def db_session(db):
s = Session.objects.create(
session_id="report-test",
agent_name="xingyao",
source_node="macmini",
model_provider="anthropic",
model_id="claude-sonnet-4-6",
total_tokens=45230,
start_time=datetime(2026, 4, 5, 10, 0, tzinfo=timezone.utc),
)
Message.objects.create(
session=s,
message_id="m1",
parent_id="root",
role="user",
content_text="Help me fix this bug",
timestamp=datetime(2026, 4, 5, 10, 23, tzinfo=timezone.utc),
)
Message.objects.create(
session=s,
message_id="m2",
parent_id="m1",
role="assistant",
content_text="The bug is on line 45...",
timestamp=datetime(2026, 4, 5, 10, 23, 30, tzinfo=timezone.utc),
)
return s
@pytest.mark.django_db
class TestMarkdownExport:
def test_basic_report(self, db_session):
md = generate_markdown_report(
messages=db_session.messages.order_by("created_at"),
date_str="2026-04-05",
)
assert "# Daily Report: 2026-04-05" in md
assert "Help me fix this bug" in md
assert "The bug is on line 45..." in md
def test_thinking_content_stripped(self, db):
s = Session.objects.create(
session_id="thinking-test",
agent_name="test",
source_node="macmini",
start_time=datetime(2026, 4, 5, 10, 0, tzinfo=timezone.utc),
)
Message.objects.create(
session=s,
message_id="m3",
parent_id="root",
role="assistant",
content_text="Final answer",
raw_content=[
{"type": "thinking", "thinking": "Let me think about this..."},
{"type": "text", "text": "Final answer"},
],
timestamp=datetime(2026, 4, 5, 10, 30, tzinfo=timezone.utc),
)
md = generate_markdown_report(
messages=s.messages.order_by("created_at"),
date_str="2026-04-05",
)
assert "Let me think about this..." not in md
assert "Final answer" in md
def test_tool_call_formatting(self, db):
s = Session.objects.create(
session_id="tool-test",
agent_name="test",
source_node="macmini",
model_id="test-model",
total_tokens=100,
start_time=datetime(2026, 4, 5, 10, 0, tzinfo=timezone.utc),
)
Message.objects.create(
session=s,
message_id="m4",
parent_id="root",
role="assistant",
content_text="I'll run a command",
raw_content=[
{"type": "text", "text": "I'll run a command"},
],
timestamp=datetime(2026, 4, 5, 10, 30, tzinfo=timezone.utc),
)
md = generate_markdown_report(
messages=s.messages.order_by("created_at"),
date_str="2026-04-05",
)
assert "test-model" in md
assert "100" in md