ai写作修改

2026-06-02 20:36:22 +08:00
1 parent 30410b0a93
commit 7efedc7b44
@@ -39,6 +39,7 @@ from app.gateway.routers import (
    models,
    multi_agent,
    notifications,
+    public_embed,
    public_scheduled_tasks,
    recommend,
    recommended_questions,
@@ -870,6 +871,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
    # surface results without a DeerFlow login.
    app.include_router(public_scheduled_tasks.router)

+    # Public embed Q&A status polling (/api/public/embed/*)
+    app.include_router(public_embed.router)
+
    # System settings API is mounted at /api/system-settings
    app.include_router(system_settings.router)

@@ -130,6 +130,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:

        app.state.thread_share_store = make_thread_share_store(sf)

+        from deerflow.persistence.embed_sessions import make_embed_session_store
+
+        app.state.embed_session_store = make_embed_session_store(sf)
+
        from deerflow.persistence.ai_writing_sessions import (
            make_ai_writing_session_store,
            register_default_store as _register_ai_writing_default_store,
@@ -269,6 +273,14 @@ def get_thread_share_store(request: Request):
    return val


+def get_embed_session_store(request: Request):
+    """Return the iframe embed session → thread mapping store."""
+    val = getattr(request.app.state, "embed_session_store", None)
+    if val is None:
+        raise HTTPException(status_code=503, detail="Embed session store not available")
+    return val
+
+
 def get_ai_writing_session_store(request: Request):
    """Return the AI writing session store, or None when not configured."""
    return getattr(request.app.state, "ai_writing_session_store", None)
@@ -0,0 +1,194 @@
+"""Public (unauthenticated) API for embedded workspace chat integrations.
+
+Parent pages embed DeerFlow in a plain iframe and cannot read ``thread_id``
+from the child frame. They pass a ``session_id`` (their own correlation id)
+in the iframe URL; the embed page registers ``session_id → thread_id`` here,
+then the parent polls ``GET /sessions/{session_id}/status``.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel, Field, field_validator
+
+from app.gateway.deps import get_checkpointer, get_embed_session_store, get_run_manager, get_thread_store
+from app.gateway.routers.threads import _derive_thread_status
+from deerflow.persistence.embed_sessions.base import EmbedSessionStore
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/public/embed", tags=["public-embed"])
+
+_SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_-]{8,128}$")
+_UNFINISHED_THREAD_STATUSES = frozenset({"running", "busy"})
+
+
+def compute_embed_finished(
+    *,
+    has_active_run: bool,
+    meta_status: str,
+    derived_status: str,
+) -> bool:
+    """Return whether an embed parent may treat the Q&A session as complete."""
+    if has_active_run or meta_status in _UNFINISHED_THREAD_STATUSES:
+        return False
+    if derived_status == "interrupted":
+        return False
+    return True
+
+
+def _validate_session_id(session_id: str) -> str:
+    value = session_id.strip()
+    if not _SESSION_ID_RE.fullmatch(value):
+        raise HTTPException(
+            status_code=422,
+            detail="session_id must be 8-128 characters (letters, digits, underscore, hyphen)",
+        )
+    return value
+
+
+class RegisterEmbedSessionRequest(BaseModel):
+    session_id: str = Field(..., description="Correlation id from the parent page (iframe query param)")
+    thread_id: str = Field(..., description="DeerFlow conversation thread id")
+
+    @field_validator("session_id")
+    @classmethod
+    def _check_session_id(cls, value: str) -> str:
+        if not _SESSION_ID_RE.fullmatch(value.strip()):
+            raise ValueError("session_id must be 8-128 characters (letters, digits, underscore, hyphen)")
+        return value.strip()
+
+    @field_validator("thread_id")
+    @classmethod
+    def _check_thread_id(cls, value: str) -> str:
+        trimmed = value.strip()
+        if not trimmed:
+            raise ValueError("thread_id is required")
+        return trimmed
+
+
+class RegisterEmbedSessionResponse(BaseModel):
+    session_id: str
+    thread_id: str
+    registered: bool = True
+
+
+class EmbedThreadStatusFields(BaseModel):
+    finished: bool
+    thread_status: str
+    has_active_run: bool
+    latest_run_status: str | None = None
+
+
+class EmbedThreadStatusResponse(EmbedThreadStatusFields):
+    thread_id: str
+
+
+class EmbedSessionStatusResponse(EmbedThreadStatusFields):
+    session_id: str
+    registered: bool = Field(description="False until the iframe has registered this session_id")
+    thread_id: str | None = Field(default=None, description="Resolved thread id when registered")
+
+
+async def _status_for_thread_id(thread_id: str, request: Request) -> EmbedThreadStatusFields:
+    thread_store = get_thread_store(request)
+    run_mgr = get_run_manager(request)
+    checkpointer = get_checkpointer(request)
+
+    record = await thread_store.get(thread_id, user_id=None)
+    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
+    try:
+        checkpoint_tuple = await checkpointer.aget_tuple(config)
+    except Exception:
+        logger.exception("Failed to read checkpoint for embed status on %s", thread_id)
+        raise HTTPException(status_code=500, detail="Failed to read thread state") from None
+
+    if record is None and checkpoint_tuple is None:
+        raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found")
+
+    derived_status = (
+        _derive_thread_status(checkpoint_tuple)
+        if checkpoint_tuple is not None
+        else (record or {}).get("status", "idle")
+    )
+    meta_status = str((record or {}).get("status") or derived_status)
+
+    has_active_run = await run_mgr.has_inflight(thread_id)
+    runs = await run_mgr.list_by_thread(thread_id)
+    latest_run_status = runs[-1].status.value if runs else None
+
+    finished = compute_embed_finished(
+        has_active_run=has_active_run,
+        meta_status=meta_status,
+        derived_status=derived_status,
+    )
+
+    return EmbedThreadStatusFields(
+        finished=finished,
+        thread_status=derived_status,
+        has_active_run=has_active_run,
+        latest_run_status=latest_run_status,
+    )
+
+
+@router.post(
+    "/sessions",
+    response_model=RegisterEmbedSessionResponse,
+    summary="Register embed session_id → thread_id (called from iframe)",
+)
+async def register_embed_session(
+    body: RegisterEmbedSessionRequest,
+    request: Request,
+) -> RegisterEmbedSessionResponse:
+    """Upsert a mapping so the parent page can poll status by session_id only."""
+    store: EmbedSessionStore = get_embed_session_store(request)
+    record = await store.upsert(body.session_id, body.thread_id)
+    return RegisterEmbedSessionResponse(
+        session_id=record["session_id"],
+        thread_id=record["thread_id"],
+    )
+
+
+@router.get(
+    "/sessions/{session_id}/status",
+    response_model=EmbedSessionStatusResponse,
+    summary="Poll embed Q&A completion by parent session_id",
+)
+async def get_embed_session_status(session_id: str, request: Request) -> EmbedSessionStatusResponse:
+    """Primary status endpoint for iframe parents — no thread_id required."""
+    session_id = _validate_session_id(session_id)
+    store: EmbedSessionStore = get_embed_session_store(request)
+    record = await store.get(session_id)
+    if record is None:
+        return EmbedSessionStatusResponse(
+            session_id=session_id,
+            registered=False,
+            thread_id=None,
+            finished=False,
+            thread_status="pending",
+            has_active_run=False,
+            latest_run_status=None,
+        )
+
+    thread_id = record["thread_id"]
+    status = await _status_for_thread_id(thread_id, request)
+    return EmbedSessionStatusResponse(
+        session_id=session_id,
+        registered=True,
+        thread_id=thread_id,
+        **status.model_dump(),
+    )
+
+
+@router.get(
+    "/threads/{thread_id}/status",
+    response_model=EmbedThreadStatusResponse,
+    summary="Poll embed Q&A completion by thread_id (optional / debugging)",
+)
+async def get_embed_thread_status(thread_id: str, request: Request) -> EmbedThreadStatusResponse:
+    """Direct thread lookup when the parent already knows thread_id."""
+    status = await _status_for_thread_id(thread_id, request)
+    return EmbedThreadStatusResponse(thread_id=thread_id, **status.model_dump())
@@ -0,0 +1,18 @@
+"""Persistence for iframe embed session → thread mappings."""
+
+from deerflow.persistence.embed_sessions.base import EmbedSessionStore
+from deerflow.persistence.embed_sessions.memory import MemoryEmbedSessionStore
+from deerflow.persistence.embed_sessions.sql import EmbedSessionRepository
+
+__all__ = [
+    "EmbedSessionStore",
+    "MemoryEmbedSessionStore",
+    "EmbedSessionRepository",
+    "make_embed_session_store",
+]
+
+
+def make_embed_session_store(session_factory):
+    if session_factory is None:
+        return MemoryEmbedSessionStore()
+    return EmbedSessionRepository(session_factory)
@@ -0,0 +1,18 @@
+"""Abstract persistence API for iframe embed session → thread mappings."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class EmbedSessionStore(ABC):
+    @abstractmethod
+    async def upsert(self, session_id: str, thread_id: str) -> dict[str, Any]:
+        """Create or update the mapping for an embed session."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def get(self, session_id: str) -> dict[str, Any] | None:
+        """Return ``{session_id, thread_id, created_at, updated_at}`` or None."""
+        raise NotImplementedError
@@ -0,0 +1,37 @@
+"""In-memory ``EmbedSessionStore`` for tests / memory backend."""
+
+from __future__ import annotations
+
+import copy
+from datetime import UTC, datetime
+from typing import Any
+
+from deerflow.persistence.embed_sessions.base import EmbedSessionStore
+
+
+class MemoryEmbedSessionStore(EmbedSessionStore):
+    def __init__(self) -> None:
+        self._sessions: dict[str, dict[str, Any]] = {}
+
+    async def upsert(self, session_id: str, thread_id: str) -> dict[str, Any]:
+        now = datetime.now(UTC).isoformat()
+        existing = self._sessions.get(session_id)
+        if existing is None:
+            record = {
+                "session_id": session_id,
+                "thread_id": thread_id,
+                "created_at": now,
+                "updated_at": now,
+            }
+        else:
+            record = {
+                **existing,
+                "thread_id": thread_id,
+                "updated_at": now,
+            }
+        self._sessions[session_id] = record
+        return copy.deepcopy(record)
+
+    async def get(self, session_id: str) -> dict[str, Any] | None:
+        record = self._sessions.get(session_id)
+        return copy.deepcopy(record) if record is not None else None
@@ -0,0 +1,21 @@
+"""ORM model for iframe embed session → thread lookups."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+from deerflow.persistence.types import BeijingDateTime
+
+
+class EmbedSessionRow(Base):
+    __tablename__ = "embed_sessions"
+
+    # Opaque id generated by the parent page and passed into the iframe URL.
+    session_id: Mapped[str] = mapped_column(String(128), primary_key=True)
+    thread_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
+    updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
@@ -0,0 +1,49 @@
+"""SQLAlchemy-backed embed session repository."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.embed_sessions.base import EmbedSessionStore
+from deerflow.persistence.embed_sessions.model import EmbedSessionRow
+
+
+def _row_to_dict(row: EmbedSessionRow) -> dict[str, Any]:
+    data = row.to_dict()
+    for key in ("created_at", "updated_at"):
+        value = data.get(key)
+        if isinstance(value, datetime):
+            data[key] = value.isoformat()
+    return data
+
+
+class EmbedSessionRepository(EmbedSessionStore):
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    async def upsert(self, session_id: str, thread_id: str) -> dict[str, Any]:
+        async with self._sf() as session:
+            row = await session.get(EmbedSessionRow, session_id)
+            now = datetime.now(UTC)
+            if row is None:
+                row = EmbedSessionRow(
+                    session_id=session_id,
+                    thread_id=thread_id,
+                    created_at=now,
+                    updated_at=now,
+                )
+                session.add(row)
+            else:
+                row.thread_id = thread_id
+                row.updated_at = now
+            await session.commit()
+            await session.refresh(row)
+            return _row_to_dict(row)
+
+    async def get(self, session_id: str) -> dict[str, Any] | None:
+        async with self._sf() as session:
+            row = await session.get(EmbedSessionRow, session_id)
+            return _row_to_dict(row) if row is not None else None
@@ -0,0 +1,39 @@
+"""Create ``embed_sessions`` table for iframe embed status polling.
+
+Revision ID: 20260602_01
+Revises: 20260528_02
+Create Date: 2026-06-02
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+
+revision = "20260602_01"
+down_revision = "20260528_02"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    inspector = inspect(op.get_bind())
+    return table_name in inspector.get_table_names()
+
+
+def upgrade() -> None:
+    if not _has_table("embed_sessions"):
+        op.create_table(
+            "embed_sessions",
+            sa.Column("session_id", sa.String(128), primary_key=True),
+            sa.Column("thread_id", sa.String(64), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        )
+        op.create_index("ix_embed_sessions_thread_id", "embed_sessions", ["thread_id"])
+
+
+def downgrade() -> None:
+    if _has_table("embed_sessions"):
+        op.drop_table("embed_sessions")
@@ -37,6 +37,7 @@ from deerflow.persistence.scheduled_tasks.model import (
 from deerflow.persistence.skills.model import SkillDisplayAdapterRow, SkillRow
 from deerflow.persistence.tags.model import TagAssignmentRow, TagRow
 from deerflow.persistence.thread_meta.model import ThreadMetaRow
+from deerflow.persistence.embed_sessions.model import EmbedSessionRow
 from deerflow.persistence.thread_shares.model import ThreadShareRow
 from deerflow.persistence.tool_metrics.model import ToolCallMetricRow
 from deerflow.persistence.user.model import UserRow
@@ -62,6 +63,7 @@ __all__ = [
    "SkillDisplayAdapterRow",
    "TagAssignmentRow",
    "TagRow",
+    "EmbedSessionRow",
    "ThreadMetaRow",
    "ThreadShareRow",
    "ToolCallMetricRow",