diff --git a/.env.example b/.env.example index 322cfcc..bf0bd6f 100644 --- a/.env.example +++ b/.env.example @@ -34,6 +34,12 @@ DATA_FILE_MAX_MB=500 DATA_FILE_TTL=604800 # 上传后用 LLM 分析文件（一句话说明、图表推荐、列的含义），false 关闭；没有配置 LLM_API_KEY 时同样不分析 DATA_FILE_INSIGHTS=true +# 大屏保存/发布的请求体上限（MB）、封面图目录（相对路径相对项目根目录）；大屏存取必须配置 MONGODB_URI，否则接口返回 503 +SCREEN_MAX_MB=10 +# 回收站里的大屏保留多少天，超过后自动彻底删除 +TRASH_RETENTION_DAYS=30 +THUMBNAIL_DIR=data/thumbnails + # 模型写的单条数据查询的超时（秒） DATA_QUERY_TIMEOUT=10 # apply_actions 里一个 $query 写进组件的最多行数 diff --git a/README.md b/README.md new file mode 100644 index 0000000..fe3da3c --- /dev/null +++ b/README.md @@ -0,0 +1,295 @@ +# Screen Agent · my-v0-server + +面向可视化大屏编辑器的 AI 后端。用户通过自然语言创建大屏、调整组件和分析上传的数据，服务将需求转换为经过校验的结构化编辑动作，交给前端执行。 + +本仓库提供 API、Agent 工作流、组件物料和数据处理能力。接入时，前端负责提交当前画布快照、展示流式反馈，并应用返回的动作。 + +[快速开始](#快速开始) · [使用指南](#使用指南) · [接口概览](#接口概览) · [架构设计](#架构设计) · [开发与测试](#开发与测试) + +## 功能特性 + +- **自然语言编辑大屏**：新建大屏，新增、删除或更换图表，修改组件属性、位置、尺寸与画布配置，支持对齐和分布等布局操作。 +- **数据驱动的问答与制图**：上传 CSV、TSV、Excel（XLSX / XLS / XLSB）、JSON、JSONL 或 Parquet，查询和汇总数据，用查询结果填充图表。 +- **动作校验与模拟执行**：按物料字段定义校验修改，并在画布副本上模拟每批动作；成功的批次累加在工作画布上，失败原因返回 Agent + 修正，通过的动作按顺序交付。 +- **多轮对话与澄清**：保存会话历史；需求需要补充时暂停任务，用户回答后继续，也可以取消等待中的任务。 +- **实时反馈**：通过 SSE 返回处理进度、问答文本增量和最终结果，区分业务失败与执行异常。 +- **组件能力扩展**：以 JSON 定义物料、以 Skills 描述任务做法，通过 MCP 提供组件目录与字段查询。 + +## 快速开始 + +### 环境要求 + +| 依赖 | 说明 | +|----------------|----------------------------------------------------------------| +| Python | 3.11 或以上，版本要求见 [pyproject.toml](pyproject.toml) | +| uv | 安装依赖并运行项目，仓库提供 `uv.lock` | +| MongoDB | 完整大屏编辑所需的物料目录，以及会话、检查点和文件元数据存储 | +| 聊天模型服务 | 通过 `ChatOpenAI` 接入；需要支持工具调用、结构化输出与流式响应 | +| TypeSafe / Jev | 判断请求是否进入问答流程，需要独立的 API Key | + +用户中心和 Langfuse 可在本地开发时关闭。默认物料 MCP 服务由应用自动启动，使用本机 `8201` 端口；API 默认使用 `8000` 端口。 + +> 不配置 MongoDB 时，会话和检查点可使用进程内存，但物料目录不会回退到本地 JSON，新增、更换组件等能力无法正常使用。首次运行请完成物料初始化。 + +### 1. 安装依赖 + +获取仓库后，在项目根目录运行： + +```sh +uv sync --locked +``` + +### 2. 配置环境 + +将 [.env.example](.env.example) 复制为 `.env`，已有配置无需覆盖。填写以下配置： + +| 配置 | 说明 | +|--------------------------------------------|----------------------------------------------| +| `LLM_API_KEY`、`LLM_MODEL`、`LLM_BASE_URL` | 聊天模型密钥、模型名称和 API 地址 | +| `TYPESAFE_API_KEY` | Jev 意图判断服务密钥 | +| `MONGODB_URI`、`MONGODB_DB` | MongoDB 连接地址和数据库名 | +| `AUTH_ENABLED` | 本地调试可设为 `false`，所有请求共用开发用户 | +| `LANGFUSE_TRACING_ENABLED` | 未配置 Langfuse 时设为 `false` | + +客户端默认发送 `reasoning_effort=low`；模型服务不支持该参数时，设置 `LLM_REASONING_EFFORT=`。 + +正式接入用户中心时设置 `AUTH_ENABLED=true` 和 `USER_CENTER_URL`，请求携带 `Authorization: Bearer `。后端通过用户中心的 +`GET /api/users/me` 获取当前用户。 + +### 3. 初始化物料目录 + +首次使用需将 `app/resources/materials` 中的物料定义导入 MongoDB。工作区包含 `scripts/seed_materials.py` 时，可运行： + +```sh +uv run python -m scripts.seed_materials +``` + +该脚本会整体替换目标数据库中的物料目录和共享字段分组，更新后需重启服务。当前 `scripts/` 未纳入版本控制，新克隆的工作区需另行取得初始化脚本。 + +### 4. 启动与验证 + +```sh +uv run uvicorn app.main:app --host 127.0.0.1 --port 8000 --reload +``` + +启动后访问： + +- [Swagger UI](http://127.0.0.1:8000/docs)：查看请求模型和调试普通接口。 +- [OpenAPI](http://127.0.0.1:8000/openapi.json)：获取接口定义。 +- [物料目录](http://127.0.0.1:8000/api/materials)：确认物料已载入，正常返回 `code: 0`。 + +## 使用指南 + +### 创建会话并发起任务 + +调用 `POST /api/conversations` 创建会话，取得 `data.conversation_id`，再向 `POST /api/agent/run` 提交需求和当前画布： + +```json +{ + "prompt": "在画布顶部添加标题：销售数据概览", + "screen": { + "canvas": { + "width": 1920, + "height": 1080 + }, + "nodes": [] + }, + "context": { + "conversation_id": "创建会话接口返回的 ID" + } +} +``` + +开启鉴权后，创建会话和后续请求都需要携带 Bearer token。每次新任务都应提交 **当前画布的完整快照**；继续同一段对话时复用 +`conversation_id`，历史消息由服务端读取。 + +### 处理流式结果 + +`/api/agent/run` 和 `/api/agent/resume` 均返回 `text/event-stream`。前端使用支持 POST 和响应流读取的客户端，例如 `fetch`。 + +| 事件 | 客户端处理方式 | +|----------------|------------------------------------------------------------------------| +| `progress` | 展示当前处理阶段 | +| `answer_delta` | 按顺序追加 `data.delta`；用于问答场景，最终以 `final.data.answer` 为准 | +| `final` | 根据 `data.status` 展示结果、应用动作或发起澄清 | +| `error` | 展示 `msg` 中的执行错误，结束本轮处理 | + +每个事件的 `data:` 行统一为 `{code, data, msg}` JSON，`final.data.status` 的含义如下： + +- `completed`：本轮完成；若有 `actions`，由前端按顺序应用。此状态不表示前端已经完成修改。 +- `waiting_for_input`：展示 `clarification`，保留 `thread_id` 等待用户回答。 +- `failed`：业务流程无法完成，`answer` 说明原因，不交付动作草稿。 +- `cancelled`：用户取消了等待补充的任务。 + +流开始前的错误通过 HTTP 状态码返回；流开始后的异常通过 `error` 事件返回。客户端不能仅凭 HTTP 200 判断成功，未收到 `final` 或 +`error` 就断开应视为结果不完整。 + +### 回答澄清问题 + +当结果为 `waiting_for_input` 时，向 `POST /api/agent/resume` 提交： + +```json +{ + "thread_id": "替换为等待补充的任务 UUID", + "reply": { + "answer": "使用月度销售额，放在画布左侧" + } +} +``` + +也可使用 `reply.option_id` 提交返回的选项，或用 `reply: {"cancel": true}` 取消暂停中的任务。恢复接口沿用任务所属会话。 + +### 上传数据并生成图表 + +1. 创建会话，将文件与 `conversation_id` 一起以 `multipart/form-data` 提交到 `POST /api/files`。 +2. 接口返回 HTTP 202 和 `data.file_id`，轮询 `GET /api/files/{file_id}/progress`。 +3. 等待 `status=ready` 后，在同一会话发起 Agent 请求，例如“按月份汇总销售额，生成柱状图”。解析失败时状态为 `failed`，原因在 + `error.msg` 中。 +4. Agent 查询文件数据，将结果转换为图表配置，再返回经过校验的编辑动作。 + +会话中已解析且未过期的文件会自动提供给 Agent，无需每轮重新上传。 + +## 接口概览 + +所有业务接口统一使用 `/api` 前缀，普通 JSON 响应同样采用 `{code, data, msg}` 结构。除物料目录和上传限制外，下列接口默认需要登录。 + +| 方法与路径 | 用途 | +|------------------------------------------------------|--------------------------------------| +| `POST /api/conversations` | 创建会话 | +| `GET /api/conversations` | 分页查询当前用户的会话 | +| `GET /api/conversations/{conversation_id}` | 读取会话历史 | +| `POST /api/agent/run` | 发起任务，返回 SSE | +| `POST /api/agent/resume` | 恢复或取消暂停中的任务，返回 SSE | +| `GET /api/materials` | 获取公开的组件物料目录 | +| `GET /api/files/limits` | 获取当前上传限制与支持格式，无需认证 | +| `POST /api/files` | 上传文件并后台解析 | +| `GET /api/files?conversation_id=...` | 获取会话文件和进行中的上传 | +| `GET /api/files/{file_id}` | 获取文件详情与表信息 | +| `GET /api/files/{file_id}/progress` | 查询解析进度 | +| `GET /api/files/{file_id}/tables/{table}/rows` | 分页读取并排序表数据 | +| `POST /api/files/{file_id}/tables/{table}/aggregate` | 分组汇总表数据 | +| `DELETE /api/files/{file_id}` | 删除文件 | + +完整字段、约束和响应模型以运行中的 [Swagger UI](http://127.0.0.1:8000/docs) +为准；动作协议定义见 [app/schemas/actions](app/schemas/actions)。 + +## 架构设计 + +```mermaid +flowchart TD + Client[大屏编辑器] --> API[FastAPI / AgentService] + API --> Input[输入与会话上下文] + Input --> Intent[Jev 意图判断] + Intent -->|明确问答且达到阈值| QA[问答 Agent] + Intent -->|其他请求| Screen[大屏 Agent] + Screen --> Materials[MCP 物料与字段查询] + Screen --> Batch[动作校验与整批模拟执行] + Screen -->|需要补充| Wait[LangGraph 暂停] + Wait -->|用户回答并恢复| Input + QA --> Data[数据描述与 DuckDB 查询] + Screen --> Data + Data --> Files[会话文件 / Parquet] + QA --> SSE[SSE 进度与结果] + Screen --> SSE + SSE --> Client +``` + +外层 LangGraph 管理路由、任务状态与澄清恢复；内层 LangChain Agent 根据上下文调用工具。只有 Jev 选中问答且概率达到阈值时进入问答分支，其余请求交给大屏 +Agent。 + +大屏 Agent 通过 `get_nodes` 读取组件完整配置，通过 MCP 查询物料，通过 `apply_actions` +提交修改。每批动作在工作画布（原始画布加上之前成功的修改）上模拟执行，读取和修改都基于它；字段和动作校验失败时这一批不生效，返回原因由 +Agent 修正。上传数据解析为 Parquet，查询结果可直接填入图表配置，避免模型逐行复写数据。 + +| 技术 | 职责 | +|------------------------------------|------------------------------------------| +| FastAPI / Pydantic | HTTP 接口、请求与响应模型、字段校验 | +| LangGraph | 外层工作流、检查点、暂停与恢复 | +| LangChain / langchain-openai | Agent 工具循环与聊天模型调用 | +| TypeSafe / Jev | 意图判断 | +| MCP | 组件物料和字段查询，接入额外工具服务 | +| MongoDB | 物料目录、会话历史、图检查点、文件元数据 | +| python-calamine / PyArrow / DuckDB | 表格读取、Parquet 存储、数据查询与汇总 | +| Loguru / Langfuse | 应用日志与可选的调用追踪 | + +## 项目结构 + +```text +app/ +├── main.py # 应用入口与服务生命周期 +├── config.py # 环境变量与配置校验 +├── api/ # Agent、会话、物料与文件接口 +├── services/ # 任务调度、会话存储与持久化装配 +├── processor/ # LangGraph 工作流、节点、状态与历史裁剪 +├── agent/ # 大屏和问答 Agent、本地工具、Skills 加载 +├── actions/ # 动作执行、布局计算与整批模拟 +├── datafiles/ # 文件解析、存储、查询、概况与后台任务 +├── schemas/ # 请求、画布、动作、会话和数据文件模型 +├── resources/ # 物料定义、目录缓存与字段规则 +├── clients/ # LLM、Jev、MCP、用户中心客户端 +├── prompt/ # 领域约束、意图、问答与数据分析提示词 +├── core/ # 认证、异常处理和日志 +├── common/ # 统一响应与 SSE 编码 +└── observability/ # Langfuse 追踪 +mcp_servers/ # 应用管理的本地 MCP 服务 +skills/ # 大屏设计、编辑、图表替换与数据制图规则 +mcp_servers.json # MCP 服务连接配置 +.env.example # 环境变量示例 +pyproject.toml # Python 依赖与项目元数据 +uv.lock # 依赖锁文件 +``` + +## 配置与运行边界 + +主要可调参数如下，完整配置见 [.env.example](.env.example) 和 [app/config.py](app/config.py)。 + +| 配置 | 默认值 | 用途 | +|-----------------------------------------------------|------------------|-----------------------------------------------------------------------------| +| `MAX_AGENT_STEPS` / `MAX_QA_STEPS` | `12` / `8` | 大屏和问答 Agent 的单次模型调用上限；大屏 Agent 没有给出结论时另加 1 次收尾 | +| `INTENT_QA_MIN_PROBABILITY` | `0.5` | 问答分支的判定阈值 | +| `CONVERSATION_HISTORY_TURNS` / `HISTORY_MAX_TOKENS` | `20` / `4096` | 历史读取轮数与带入模型的 token 预算 | +| `MONGODB_CHECKPOINT_TTL` | `604800` | 检查点保留秒数，`0` 表示永久保留 | +| `DATA_FILE_DIR` | `data/files` | 原始文件和解析结果的磁盘目录 | +| `DATA_FILE_MAX_MB` / `DATA_FILE_TTL` | `500` / `604800` | 单文件大小上限与保留秒数 | +| `DATA_FILE_INSIGHTS` | `true` | 上传后生成文件说明、列含义和图表建议，可关闭 | +| `DATA_QUERY_TIMEOUT` / `DATA_EMBED_MAX_ROWS` | `10` / `1000` | 单次 SQL 查询超时秒数与每处查询写入组件的行数上限 | +| `AUTH_ENABLED` / `AUTH_CACHE_TTL_SECONDS` | `true` / `30` | 用户中心鉴权与 token 校验缓存秒数 | +| `MCP_TOOL_TIMEOUT` | `5` | 单次 MCP 工具调用超时秒数 | + +- **单 worker 运行**：任务与会话的并发保护、上传进度保存在进程内；即使配置 MongoDB，也不能据此直接扩展为多 + worker。同一会话同时只能执行一个任务。 +- **文件存储**：MongoDB 保存元数据，原始文件和 Parquet 位于 `DATA_FILE_DIR`。容器部署需保留该目录，并为反向代理配置匹配的上传大小限制。 +- **流式连接**：断开连接会取消当前执行，不会转为后台任务。当前没有 SSE 心跳、事件重放或断线续传；代理应关闭响应缓冲，并给模型调用留出足够的等待时间。 +- **前端接入**：保持 `/api` 前缀；当前应用未配置 CORS，中间开发代理或同源反向代理可将 `/api` 转发到后端。 +- **日志与追踪**：日志写入 `logs/app.log`。启用 Langfuse 时填写真实的公钥、私钥和服务地址；追踪故障不阻断主业务。 + +## 开发与测试 + +阅读主链路可从 [app/main.py](app/main.py) +开始，依次查看 [Agent 接口](app/api/agent.py)、[AgentService](app/services/agent.py)、[外层工作流](app/processor/graph.py)、[大屏 Agent](app/agent/screen_agent.py) +和 [动作模拟](app/actions/batch.py)。 + +扩展组件时修改 [物料定义](app/resources/materials)，导入数据库后重启服务；修改任务处理规则时查看 [skills](skills)；新增 MCP +服务时修改 [mcp_servers.json](mcp_servers.json)，连接配置支持 `${环境变量名}` 展开。 + +当前 `.gitignore` 排除了 `tests/`、`docs/`、`scripts/` 等本地辅助目录，普通克隆不会包含这些内容。若工作区已有测试文件，在根目录使用标准库 +`unittest` 运行： + +```sh +uv run python -m unittest discover -s tests -v +``` + +测试所需的 `openpyxl` 已列入开发依赖，`uv sync --locked` 默认安装。单元测试不能替代真实模型、Jev、MongoDB、用户中心和前端动作应用的联调验证。 + +## 常见问题 + +| 现象 | 检查项 | +|---------------------------------------------|-------------------------------------------------------------------------------------------| +| 会话或 Agent 接口返回 401 | 本地调试确认 `AUTH_ENABLED=false`；开启鉴权时携带有效 Bearer token | +| 提示物料目录未加载，或 MCP 物料服务无法启动 | 检查 `MONGODB_URI`、物料初始化结果与 `8201` 端口，导入后重启服务 | +| Agent 返回 422，提示缺少会话 | 先创建会话，并在 `context.conversation_id` 中传入返回的 ID | +| 请求返回 409 | 同一会话或任务仍在运行，或尝试恢复未处于等待补充状态的任务 | +| API 可访问，但 Agent 执行失败 | 查看流中的 `error`、`final.data.answer` 和 `logs/app.log`，核对模型、Jev 密钥与服务连通性 | +| 上传返回 202，但文件暂时不可用 | 202 仅表示上传已接收，继续轮询解析进度，等待 `ready` | +| 前端一直等不到流式反馈 | 检查是否持续读取响应流，以及代理是否缓冲 SSE 或提前超时 | diff --git a/app/api/conversations.py b/app/api/conversations.py index d57f9bc..d6d8655 100644 --- a/app/api/conversations.py +++ b/app/api/conversations.py @@ -8,22 +8,34 @@ from app.common.response import ApiResponse, success_response from app.core.auth import get_current_user from app.core.exceptions import BizException from app.schemas.conversation import ( + ConversationCreate, ConversationCreated, ConversationDetail, ConversationList, + ConversationRename, ) from app.schemas.user import CurrentUser +MAX_OFFSET = 10000 + router = APIRouter(prefix="/conversations", tags=["Conversation"]) @router.post("") async def create_conversation( - request: Request, user: CurrentUser = Depends(get_current_user) + request: Request, + user: CurrentUser = Depends(get_current_user), + body: ConversationCreate | None = None, ) -> ApiResponse[ConversationCreated]: store = request.app.state.conversation_store + screen_id = body.screen_id if body else None + # 会话挂到大屏上，大屏必须是自己的；没配 MongoDB（没有大屏服务）时无从校验，直接放过 + screens = getattr(request.app.state, "screen_service", None) + if screen_id is not None and screens is not None: + if not await screens.exists(screen_id, user.id): + raise BizException(404, "大屏不存在或已删除") return success_response( - ConversationCreated(conversation_id=await store.create(user.id)) + ConversationCreated(conversation_id=await store.create(user.id, screen_id)) ) @@ -32,11 +44,15 @@ async def list_conversations( request: Request, user: Annotated[CurrentUser, Depends(get_current_user)], limit: Annotated[int, Query(ge=1, le=100)] = 20, - offset: Annotated[int, Query(ge=0)] = 0, + offset: Annotated[int, Query(ge=0, le=MAX_OFFSET)] = 0, + screen_id: Annotated[str | None, Query(min_length=1)] = None, ) -> ApiResponse[ConversationList]: store = request.app.state.conversation_store return success_response( - ConversationList(conversations=await store.list_by_user(user.id, limit, offset)) + ConversationList( + conversations=await store.list_by_user(user.id, limit, offset, screen_id), + total=await store.count_by_user(user.id, screen_id), + ) ) @@ -54,3 +70,28 @@ async def get_conversation( return success_response( ConversationDetail(conversation_id=conversation_id, turns=turns) ) + + +@router.patch("/{conversation_id}") +async def rename_conversation( + conversation_id: str, + body: ConversationRename, + request: Request, + user: CurrentUser = Depends(get_current_user), +) -> ApiResponse[dict]: + store = request.app.state.conversation_store + if not await store.rename(conversation_id, user.id, body.title): + raise BizException(404, "会话不存在或已过期") + return success_response() + + +@router.delete("/{conversation_id}") +async def delete_conversation( + conversation_id: str, + request: Request, + user: CurrentUser = Depends(get_current_user), +) -> ApiResponse[dict]: + store = request.app.state.conversation_store + if not await store.delete(conversation_id, user.id): + raise BizException(404, "会话不存在或已过期") + return success_response() diff --git a/app/api/datasource.py b/app/api/datasource.py new file mode 100644 index 0000000..dde9fa6 --- /dev/null +++ b/app/api/datasource.py @@ -0,0 +1,29 @@ +"""编辑态数据源取数：登录用户传 URL，服务端替浏览器请求，responsePath / transform 仍由前端做""" + +from typing import Annotated, Any, Literal + +from fastapi import APIRouter, Depends +from pydantic import BaseModel, Field + +from app.common.response import ApiResponse, success_response +from app.core.auth import get_current_user +from app.schemas.user import CurrentUser +from app.services.datasource_proxy import fetch_json + +router = APIRouter(prefix="/datasource", tags=["DataSource"]) + + +class DataSourceFetch(BaseModel): + url: str = Field(min_length=1, max_length=4096) + method: Literal["GET", "POST"] = "GET" + params: dict[str, Any] | None = None + headers: dict[str, Any] | None = None + + +@router.post("/fetch") +async def fetch_datasource( + body: DataSourceFetch, + user: Annotated[CurrentUser, Depends(get_current_user)], +) -> ApiResponse[dict[str, Any] | list[Any]]: + data = await fetch_json(body.url, body.method, body.params, body.headers) + return success_response(data) diff --git a/app/api/files.py b/app/api/files.py index 9a2dbb6..630dade 100644 --- a/app/api/files.py +++ b/app/api/files.py @@ -85,9 +85,13 @@ class UploadSizeLimit: return None +def _public(record: DataFileRecord) -> dict: + return {**record.model_dump(), "pinned": bool(record.pinned_by)} + + def _detail(record: DataFileRecord) -> DataFileDetail: # 按接口模型重新校验，去掉 owner_id、sha256、列概况等只在后端用的字段 - return DataFileDetail.model_validate(record.model_dump()) + return DataFileDetail.model_validate(_public(record)) @limits_router.get("/limits") @@ -130,9 +134,7 @@ async def list_files( records = await service.list_files(user.id, conversation_id) return success_response( DataFileList( - files=[ - DataFileInfo.model_validate(record.model_dump()) for record in records - ], + files=[DataFileInfo.model_validate(_public(record)) for record in records], uploads=service.list_uploads(user.id, conversation_id), ) ) diff --git a/app/api/public.py b/app/api/public.py new file mode 100644 index 0000000..d501b38 --- /dev/null +++ b/app/api/public.py @@ -0,0 +1,54 @@ +"""匿名接口，统一放在 /api/public 下，一眼能看出哪些不鉴权""" + +from typing import Any + +from fastapi import APIRouter +from fastapi.responses import FileResponse + +from app.api.screens import Service +from app.common.response import ApiResponse, success_response +from app.core.exceptions import BizException +from app.schemas.saved_screen import PublicScreen +from app.services.screens import public_page + +router = APIRouter(prefix="/public", tags=["Public"]) + + +@router.get("/screens/{share_id}") +async def get_public_screen( + share_id: str, service: Service +) -> ApiResponse[PublicScreen]: + """线上版本：返回发布快照，不是草稿；已下线、已删除、不存在都是 404""" + summary, snapshot = await service.get_live(share_id) + return success_response( + PublicScreen( + name=summary.name, + page=public_page(snapshot.page), + version=snapshot.version, + published_at=snapshot.published_at, + ) + ) + + +@router.get("/screens/{share_id}/datasources/{datasource_id}") +async def fetch_public_datasource( + share_id: str, datasource_id: str, service: Service +) -> ApiResponse[dict[str, Any] | list[Any]]: + """线上取数：地址取自已发布快照里的数据源配置，前端只传 ID""" + return success_response( + await service.fetch_live_datasource(share_id, datasource_id) + ) + + +@router.get("/thumbnails/{thumbnail_id}") +async def get_thumbnail(thumbnail_id: str, service: Service) -> FileResponse: + opened = service.open_thumbnail(thumbnail_id) + if opened is None: + raise BizException(404, "封面不存在") + path, media_type = opened + # 每次上传都换新 ID，内容永远不变，可以放心长期缓存 + return FileResponse( + path, + media_type=media_type, + headers={"Cache-Control": "public, max-age=31536000, immutable"}, + ) diff --git a/app/api/screens.py b/app/api/screens.py new file mode 100644 index 0000000..c3b5cf3 --- /dev/null +++ b/app/api/screens.py @@ -0,0 +1,249 @@ +"""大屏存取、发布和管理页接口 + +登录用户只能操作自己的大屏；别人的大屏与不存在同样返回 404 +""" + +from datetime import timedelta +from typing import Annotated, Literal + +from fastapi import APIRouter, Depends, File, Query, Request, UploadFile + +from app.common.response import ApiResponse, success_response +from app.core.auth import get_current_user +from app.core.exceptions import BizException +from app.schemas.saved_screen import ( + ListStatus, + PublishBrief, + PublishHistory, + PublishHistoryItem, + ScreenCopied, + ScreenCopyRequest, + ScreenCreate, + ScreenCreated, + ScreenDetail, + ScreenList, + ScreenListItem, + ScreenPublished, + ScreenPublishRequest, + ScreenRename, + ScreenRestored, + ScreenSave, + ScreenSaved, + ScreenSummary, + SortField, + ThumbnailResult, + TrashEmptied, +) +from app.schemas.user import CurrentUser +from app.services import thumbnails +from app.services.screens import ScreenService + +router = APIRouter(prefix="/screens", tags=["Screen"]) + +# 深翻页的 skip 会让 MongoDB 空转，管理页一屏 20 条，这个上限足够 +MAX_OFFSET = 10000 +THUMBNAIL_URL = "/api/public/thumbnails/{}" + +User = Annotated[CurrentUser, Depends(get_current_user)] + + +def get_screen_service(request: Request) -> ScreenService: + """没配 MongoDB 时大屏数据只会存在进程内存里，重启就丢，所以直接 503""" + service = getattr(request.app.state, "screen_service", None) + if service is None: + raise BizException(503, "未配置 MongoDB，大屏存取不可用") + return service + + +Service = Annotated[ScreenService, Depends(get_screen_service)] + + +def thumbnail_url(thumbnail_id: str | None) -> str | None: + return None if thumbnail_id is None else THUMBNAIL_URL.format(thumbnail_id) + + +def _list_item(summary: ScreenSummary, trash_days: int) -> ScreenListItem: + publish = summary.publish + return ScreenListItem( + screen_id=summary.screen_id, + name=summary.name, + thumbnail_url=thumbnail_url(summary.thumbnail), + canvas=summary.canvas, + status="draft" if publish is None else publish.status, + has_unpublished_changes=publish is not None + and summary.revision > publish.revision, + share_id=None if publish is None else publish.share_id, + updated_at=summary.updated_at, + published_at=None if publish is None else publish.published_at, + deleted_at=summary.deleted_at, + purge_at=None + if summary.deleted_at is None + else summary.deleted_at + timedelta(days=trash_days), + ) + + +@router.post("") +async def create_screen( + body: ScreenCreate, user: User, service: Service +) -> ApiResponse[ScreenCreated]: + record = await service.create(user.id, body.name, body.page) + return success_response( + ScreenCreated(screen_id=record.screen_id, revision=record.revision) + ) + + +@router.get("") +async def list_screens( + user: User, + service: Service, + keyword: Annotated[str | None, Query(max_length=100)] = None, + status: ListStatus | None = None, + sort: SortField = "updated_at", + order: Literal["asc", "desc"] = "desc", + limit: Annotated[int, Query(ge=1, le=100)] = 20, + offset: Annotated[int, Query(ge=0, le=MAX_OFFSET)] = 0, + deleted: bool = False, +) -> ApiResponse[ScreenList]: + items, total = await service.list_screens( + user.id, + keyword=keyword.strip() or None if keyword else None, + status=status, + sort=sort, + descending=order == "desc", + limit=limit, + offset=offset, + deleted=deleted, + ) + return success_response( + ScreenList( + items=[_list_item(i, service.trash_days) for i in items], total=total + ) + ) + + +@router.delete("/trash") +async def empty_trash(user: User, service: Service) -> ApiResponse[TrashEmptied]: + """清空回收站：彻底删除，不可恢复。要先于 /{screen_id} 注册""" + return success_response(TrashEmptied(count=await service.empty_trash(user.id))) + + +@router.get("/{screen_id}") +async def get_screen( + screen_id: str, user: User, service: Service +) -> ApiResponse[ScreenDetail]: + record = await service.get(screen_id, user.id) + publish = record.publish + return success_response( + ScreenDetail( + screen_id=record.screen_id, + name=record.name, + page=record.page, + revision=record.revision, + updated_at=record.updated_at, + publish=None + if publish is None + else PublishBrief.model_validate(publish.model_dump(exclude={"revision"})), + ) + ) + + +@router.put("/{screen_id}") +async def save_screen( + screen_id: str, body: ScreenSave, user: User, service: Service +) -> ApiResponse[ScreenSaved]: + saved = await service.save(screen_id, user.id, body.page, body.revision) + return success_response( + ScreenSaved(revision=saved.revision, updated_at=saved.updated_at) + ) + + +@router.patch("/{screen_id}") +async def rename_screen( + screen_id: str, body: ScreenRename, user: User, service: Service +) -> ApiResponse[dict]: + await service.rename(screen_id, user.id, body.name) + return success_response() + + +@router.delete("/{screen_id}") +async def delete_screen( + screen_id: str, user: User, service: Service +) -> ApiResponse[dict]: + await service.delete(screen_id, user.id) + return success_response() + + +@router.post("/{screen_id}/restore") +async def restore_deleted_screen( + screen_id: str, user: User, service: Service +) -> ApiResponse[dict]: + await service.restore_deleted(screen_id, user.id) + return success_response() + + +@router.post("/{screen_id}/publish") +async def publish_screen( + screen_id: str, body: ScreenPublishRequest, user: User, service: Service +) -> ApiResponse[ScreenPublished]: + info = await service.publish(screen_id, user.id, body.revision) + return success_response( + ScreenPublished( + version=info.version, share_id=info.share_id, published_at=info.published_at + ) + ) + + +@router.post("/{screen_id}/unpublish") +async def unpublish_screen( + screen_id: str, user: User, service: Service +) -> ApiResponse[dict]: + await service.unpublish(screen_id, user.id) + return success_response() + + +@router.post("/{screen_id}/copy") +async def copy_screen( + screen_id: str, + user: User, + service: Service, + body: ScreenCopyRequest | None = None, +) -> ApiResponse[ScreenCopied]: + new_id = await service.copy_screen(screen_id, user.id, body.name if body else None) + return success_response(ScreenCopied(screen_id=new_id)) + + +@router.put("/{screen_id}/thumbnail") +async def upload_thumbnail( + screen_id: str, + file: Annotated[UploadFile, File()], + user: User, + service: Service, +) -> ApiResponse[ThumbnailResult]: + data = await file.read(thumbnails.MAX_BYTES + 1) + thumbnail_id = await service.set_thumbnail(screen_id, user.id, data) + return success_response(ThumbnailResult(thumbnail_url=thumbnail_url(thumbnail_id))) + + +@router.get("/{screen_id}/publishes") +async def list_publishes( + screen_id: str, + user: User, + service: Service, + limit: Annotated[int, Query(ge=1, le=100)] = 20, + offset: Annotated[int, Query(ge=0, le=MAX_OFFSET)] = 0, +) -> ApiResponse[PublishHistory]: + entries, total = await service.list_publishes(screen_id, user.id, limit, offset) + return success_response( + PublishHistory( + items=[PublishHistoryItem.model_validate(e.model_dump()) for e in entries], + total=total, + ) + ) + + +@router.post("/{screen_id}/publishes/{version}/restore") +async def restore_publish( + screen_id: str, version: int, user: User, service: Service +) -> ApiResponse[ScreenRestored]: + revision = await service.restore_publish(screen_id, user.id, version) + return success_response(ScreenRestored(revision=revision)) diff --git a/app/config.py b/app/config.py index 37f7af1..35a89e9 100644 --- a/app/config.py +++ b/app/config.py @@ -117,6 +117,21 @@ class Settings: ) ) + # 大屏保存与发布：单次请求体上限（MB）；图表绑定数据文件时数据快照嵌在页面里，一屏很容易上 MB + screen_max_mb: int = field( + default_factory=lambda: int(os.getenv("SCREEN_MAX_MB", "10")) + ) + # 回收站保留天数：软删的大屏超过这个时间由后台任务彻底删除（连同发布历史和封面） + trash_retention_days: int = field( + default_factory=lambda: int(os.getenv("TRASH_RETENTION_DAYS", "30")) + ) + # 大屏封面图存放目录，相对路径相对项目根目录；和数据文件一样放本地磁盘 + thumbnail_dir: Path = field( + default_factory=lambda: ( + PROJECT_ROOT / (os.getenv("THUMBNAIL_DIR") or "data/thumbnails") + ) + ) + # 模型写的单条数据查询的超时（秒） data_query_timeout: float = field( default_factory=lambda: float(os.getenv("DATA_QUERY_TIMEOUT", "10")) @@ -171,6 +186,10 @@ class Settings: raise ValueError("DATA_FILE_MAX_MB 必须是正整数") if self.data_file_ttl < 1: raise ValueError("DATA_FILE_TTL 必须是正整数") + if self.trash_retention_days < 1: + raise ValueError("TRASH_RETENTION_DAYS 必须是正整数") + if self.screen_max_mb < 1: + raise ValueError("SCREEN_MAX_MB 必须是正整数") if self.data_query_timeout <= 0: raise ValueError("DATA_QUERY_TIMEOUT 必须是正数") if self.data_embed_max_rows < 1: diff --git a/app/core/body_limit.py b/app/core/body_limit.py new file mode 100644 index 0000000..ca481ab --- /dev/null +++ b/app/core/body_limit.py @@ -0,0 +1,71 @@ +"""限制请求体大小：声明的 Content-Length 超限直接 413，没声明（分块）或声明得不实的，边读边数 + +FastAPI 会先把整个请求体收完再调用接口函数，超大的请求体已经占满了带宽和内存 +和数据文件上传的 UploadSizeLimit 同一出发点；限制随路径而不同，所以用规则表，第一条命中的生效 +""" + +import re +from dataclasses import dataclass + +from fastapi.responses import JSONResponse +from starlette.exceptions import HTTPException +from starlette.types import ASGIApp, Message, Receive, Scope, Send + +from app.common.response import error_content + + +class BodyTooLarge(HTTPException): + """边读边数时超限；message 由全局 HTTP 异常处理放进信封，和声明超限的响应文案一致""" + + def __init__(self, message: str): + super().__init__(413) + self.message = message + + +@dataclass(frozen=True) +class BodyLimitRule: + methods: frozenset[str] + path: re.Pattern[str] + max_bytes: int + message: str + + +class BodyLimit: + def __init__(self, app: ASGIApp, rules: list[BodyLimitRule]): + self.app = app + self.rules = rules + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + rule = self._match(scope) if scope["type"] == "http" else None + if rule is None: + await self.app(scope, receive, send) + return + length = dict(scope["headers"]).get(b"content-length", b"") + if length.isdigit() and int(length) > rule.max_bytes: + response = JSONResponse(error_content(413, rule.message), 413) + await response(scope, receive, send) + return + received = 0 + + async def counted() -> Message: + nonlocal received + message = await receive() + if message["type"] == "http.request": + received += len(message.get("body", b"")) + if received > rule.max_bytes: + # 抛出的异常由应用里统一的 HTTP 异常处理转成 413 信封 + raise BodyTooLarge(rule.message) + return message + + await self.app(scope, counted, send) + + def _match(self, scope: Scope) -> BodyLimitRule | None: + return next( + ( + rule + for rule in self.rules + if scope["method"] in rule.methods + and rule.path.fullmatch(scope["path"]) + ), + None, + ) diff --git a/app/core/exceptions.py b/app/core/exceptions.py index b16d68a..60f32cc 100644 --- a/app/core/exceptions.py +++ b/app/core/exceptions.py @@ -50,6 +50,8 @@ async def handle_http_exception(_request: Request, exc: HTTPException) -> JSONRe message = HTTPStatus(exc.status_code).phrase except ValueError: message = "请求失败" + # 个别异常自带面向用户的文案，例如请求体超限 + message = getattr(exc, "message", None) or message return JSONResponse( status_code=exc.status_code, content=error_content(exc.status_code, message), diff --git a/app/datafiles/service.py b/app/datafiles/service.py index 26e1273..b91d9cd 100644 --- a/app/datafiles/service.py +++ b/app/datafiles/service.py @@ -195,7 +195,8 @@ class DataFileService: async def delete_file(self, user_id: int, file_id: str) -> None: await self.get_file(user_id, file_id) - await self.store.mark_deleted(file_id) + # 同时撤销大屏对它的固定：文件没了，记录不该再停在"永不过期" + await self.store.mark_deleted(file_id, datetime.now(UTC) + self._ttl) # Windows 上正在查询的 Parquet 删不掉，留给定时清理 await asyncio.to_thread(shutil.rmtree, self.root / file_id, ignore_errors=True) @@ -215,6 +216,21 @@ class DataFileService: raise BizException(404, NOT_FOUND) return records + async def sync_screen_files( + self, user_id: int, screen_id: str, file_ids: set[str] + ) -> None: + """大屏保存后调用：它引用的文件固定住，不再引用的放开（重新按保留时长计时）""" + current = await self.store.pinned_ids(screen_id) + if added := file_ids - current: + await self.store.pin(added, user_id, screen_id) + if removed := current - file_ids: + await self.store.unpin(screen_id, removed, datetime.now(UTC) + self._ttl) + + async def release_screen_files(self, screen_id: str) -> None: + """大屏进回收站或被彻底删除：它固定的文件全部放开""" + if current := await self.store.pinned_ids(screen_id): + await self.store.unpin(screen_id, current, datetime.now(UTC) + self._ttl) + async def aclose(self) -> None: """应用退出时取消还在进行的分析，记录里留下的 pending 下次启动后读取时按失败返回""" for task in list(self._tasks): diff --git a/app/datafiles/storage.py b/app/datafiles/storage.py index 568880e..71975d5 100644 --- a/app/datafiles/storage.py +++ b/app/datafiles/storage.py @@ -18,6 +18,10 @@ from app.schemas.data_file import DataFileRecord, DataTable, FileInsights SOURCE_NAME = "source" +# 被大屏固定的文件的占位过期时间：MongoDB 的 TTL 索引只认 expires_at，推到遥远的将来就不会被删 +PINNED_EXPIRES = datetime(9999, 1, 1, tzinfo=UTC) + + def table_path(root: Path, file_id: str, table: DataTable) -> Path: return root / file_id / table.parquet @@ -43,7 +47,11 @@ class DataFileStore(Protocol): """为会话预留 count 个连续的表号，返回第一个；表号从 1 开始，不会重复分配""" ... - async def mark_deleted(self, file_id: str) -> None: ... + async def mark_deleted( + self, file_id: str, expires_at: datetime | None = None + ) -> None: + """给了 expires_at 就同时撤销所有大屏对它的固定，并把过期时间改成它（被固定的文件不能停在 9999 年）""" + ... async def update_insights(self, file_id: str, insights: FileInsights) -> None: """记录已过期或不存在时什么都不做""" @@ -53,6 +61,20 @@ class DataFileStore(Protocol): """未过期且未删除的 file_id，清理目录时用""" ... + async def pin(self, file_ids: set[str], owner_id: int, screen_id: str) -> None: + """让 screen_id 固定这些文件（只对 owner 名下、未过期、未删除的生效），被固定的文件不再过期""" + ... + + async def unpin( + self, screen_id: str, file_ids: set[str], expires_at: datetime + ) -> None: + """撤销 screen_id 对这些文件的固定；不再被任何大屏固定的文件，过期时间改为 expires_at""" + ... + + async def pinned_ids(self, screen_id: str) -> set[str]: + """被 screen_id 固定的文件，包括已删除的（它们也要撤销固定）""" + ... + class InMemoryDataFileStore: def __init__(self) -> None: @@ -90,9 +112,14 @@ class InMemoryDataFileStore: self._table_numbers[conversation_id] = first + count - 1 return first - async def mark_deleted(self, file_id: str) -> None: + async def mark_deleted( + self, file_id: str, expires_at: datetime | None = None + ) -> None: if (record := self._records.get(file_id)) is not None: - self._records[file_id] = record.model_copy(update={"deleted": True}) + changes: dict = {"deleted": True} + if expires_at is not None: + changes.update(pinned_by=[], expires_at=expires_at) + self._records[file_id] = record.model_copy(update=changes) async def update_insights(self, file_id: str, insights: FileInsights) -> None: if (record := self._records.get(file_id)) is not None: @@ -102,6 +129,37 @@ class InMemoryDataFileStore: self._purge_expired() return {key for key, record in self._records.items() if not record.deleted} + async def pin(self, file_ids, owner_id, screen_id): + self._purge_expired() + for file_id in file_ids: + record = self._records.get(file_id) + if record is None or record.deleted or record.owner_id != owner_id: + continue + pinned_by = record.pinned_by + if screen_id not in pinned_by: + pinned_by = [*pinned_by, screen_id] + self._records[file_id] = record.model_copy( + update={"pinned_by": pinned_by, "expires_at": PINNED_EXPIRES} + ) + + async def unpin(self, screen_id, file_ids, expires_at): + for file_id in file_ids: + record = self._records.get(file_id) + if record is None or screen_id not in record.pinned_by: + continue + pinned_by = [item for item in record.pinned_by if item != screen_id] + changes = {"pinned_by": pinned_by} + if not pinned_by: + changes["expires_at"] = expires_at + self._records[file_id] = record.model_copy(update=changes) + + async def pinned_ids(self, screen_id): + return { + key + for key, record in self._records.items() + if screen_id in record.pinned_by + } + class MongoDataFileStore: def __init__(self, db: Any, prefix: str = ""): @@ -117,6 +175,8 @@ class MongoDataFileStore: await self._files.create_index( [("conversation_id", ASCENDING), ("created_at", ASCENDING)] ) + # 按大屏找它固定的文件 + await self._files.create_index("pinned_by") async def insert(self, record: DataFileRecord) -> None: await self._files.insert_one({"_id": record.file_id, **record.model_dump()}) @@ -149,8 +209,13 @@ class MongoDataFileStore: ) return document["tables"] - count + 1 - async def mark_deleted(self, file_id: str) -> None: - await self._files.update_one({"_id": file_id}, {"$set": {"deleted": True}}) + async def mark_deleted( + self, file_id: str, expires_at: datetime | None = None + ) -> None: + changes: dict = {"deleted": True} + if expires_at is not None: + changes.update(pinned_by=[], expires_at=expires_at) + await self._files.update_one({"_id": file_id}, {"$set": changes}) async def update_insights(self, file_id: str, insights: FileInsights) -> None: await self._files.update_one( @@ -164,6 +229,57 @@ class MongoDataFileStore: ) ) + async def pin(self, file_ids, owner_id, screen_id): + if not file_ids: + return + await self._files.update_many( + { + "_id": {"$in": list(file_ids)}, + "owner_id": owner_id, + "deleted": False, + "expires_at": {"$gt": _now()}, + }, + { + "$addToSet": {"pinned_by": screen_id}, + "$set": {"expires_at": PINNED_EXPIRES}, + }, + ) + + async def unpin(self, screen_id, file_ids, expires_at): + if not file_ids: + return + # 一条聚合管道更新：去掉这个大屏，且只有数组因此变空时才重新计时。 + # 单文档原子，不会停在"数组空了但仍是 9999 年"；只改确实固定了它的文档 + await self._files.update_many( + {"_id": {"$in": list(file_ids)}, "pinned_by": screen_id}, + [ + { + "$set": { + "pinned_by": { + "$filter": { + "input": "$pinned_by", + "cond": {"$ne": ["$$this", screen_id]}, + } + } + } + }, + { + "$set": { + "expires_at": { + "$cond": [ + {"$eq": [{"$size": "$pinned_by"}, 0]}, + expires_at, + "$expires_at", + ] + } + } + }, + ], + ) + + async def pinned_ids(self, screen_id): + return set(await self._files.distinct("_id", {"pinned_by": screen_id})) + def _record(document: dict[str, Any]) -> DataFileRecord: return DataFileRecord.model_validate( diff --git a/app/main.py b/app/main.py index 38e4040..ac9ddb9 100644 --- a/app/main.py +++ b/app/main.py @@ -1,6 +1,7 @@ """FastAPI 应用入口""" import asyncio +import re from contextlib import asynccontextmanager from fastapi import Depends, FastAPI @@ -8,10 +9,15 @@ from loguru import logger from app.api.agent import router from app.api.conversations import router as conversation_router +from app.api.datasource import router as datasource_router from app.api.files import UploadSizeLimit from app.api.files import limits_router as file_limits_router from app.api.files import router as file_router from app.api.materials import router as material_router +from app.api.public import router as public_router +from app.api.screens import router as screen_router +from app.config import settings +from app.core.body_limit import BodyLimit, BodyLimitRule from app.core.auth import get_current_user from app.core.exceptions import register_exception_handlers from app.core.logging import setup_logging @@ -32,12 +38,16 @@ async def lifespan(application: FastAPI): from app.services.agent import AgentService from app.services.material_mongo import warm_material_cache from app.services.persistence import open_persistence + from app.services.screen_mongo import MongoScreenStore + from app.services.screens import ScreenService + from app.services.thumbnails import ThumbnailStorage # 按配置选择 MongoDB 或进程内存，会话记录与图检查点使用同一种存储 persistence = await open_persistence() application.state.conversation_store = persistence.store mcp_processes = [] sweeper = None + trash_sweeper = None data_files = None try: # 数据文件的元数据和会话记录用同一种存储；启动时先清理一遍过期目录，之后每小时一次 @@ -48,6 +58,21 @@ async def lifespan(application: FastAPI): ) application.state.data_file_service = data_files sweeper = asyncio.create_task(data_files.run_sweeper()) + # 大屏存取必须落 MongoDB：没配时不创建服务，screens 相关接口返回 503，而不是存在内存里重启就丢 + screen_service = None + if persistence.db is not None: + screen_store = MongoScreenStore(persistence.db) + await screen_store.ensure_indexes() + screen_service = ScreenService( + screen_store, + ThumbnailStorage(settings.thumbnail_dir), + data_files, + settings.trash_retention_days, + ) + trash_sweeper = asyncio.create_task(screen_service.run_trash_sweeper()) + else: + logger.warning("未配置 MONGODB_URI，大屏存取接口不可用") + application.state.screen_service = screen_service # 物料目录启动时载入内存，动作校验与前端接口共用，避免重复查询数据库 await warm_material_cache(persistence.db) # 拉起 mcp_servers.json 中由应用管理的服务；单个服务失败只记录，用到它的请求执行时报错 @@ -64,6 +89,8 @@ async def lifespan(application: FastAPI): finally: if sweeper is not None: sweeper.cancel() + if trash_sweeper is not None: + trash_sweeper.cancel() if data_files is not None: await data_files.aclose() # 共享客户端只在应用退出时关闭，追踪刷新放到线程避免阻塞事件循环 @@ -102,6 +129,37 @@ def create_app() -> FastAPI: # 上传在读请求体之前先按 Content-Length 检查大小 application.add_middleware(UploadSizeLimit, path=f"/api{file_router.prefix}") application.include_router(material_router, prefix="/api") + application.include_router(screen_router, prefix="/api", dependencies=authenticated) + application.include_router( + datasource_router, prefix="/api", dependencies=authenticated + ) + # 匿名接口统一放 /api/public 下，不挂鉴权 + application.include_router(public_router, prefix="/api") + # 保存 / 发布的请求体比普通接口大得多，单独放宽；封面图 1 MB + application.add_middleware( + BodyLimit, + rules=[ + BodyLimitRule( + frozenset({"POST", "PUT"}), + re.compile(r"/api/screens(/[^/]+)?"), + settings.screen_max_mb * 1024 * 1024, + f"大屏数据不能超过 {settings.screen_max_mb} MB", + ), + BodyLimitRule( + frozenset({"PUT"}), + re.compile(r"/api/screens/[^/]+/thumbnail"), + 1024 * 1024 + 64 * 1024, + "封面图不能超过 1 MB", + ), + # 其余带请求体的接口（发布、复制、改名、数据源取数、会话）都只有几个字段 + BodyLimitRule( + frozenset({"POST", "PUT", "PATCH"}), + re.compile(r"/api/(screens|datasource|conversations)(/.*)?"), + 64 * 1024, + "请求体过大", + ), + ], + ) return application diff --git a/app/schemas/conversation.py b/app/schemas/conversation.py index 2f4fd6e..0cdce80 100644 --- a/app/schemas/conversation.py +++ b/app/schemas/conversation.py @@ -1,9 +1,15 @@ """会话及对话轮次，用于多轮上下文和后续持久化""" from datetime import datetime -from typing import Literal +from typing import Annotated, Literal -from pydantic import BaseModel +from pydantic import BaseModel, StringConstraints + +from .base import NonBlankString + +Title = Annotated[ + str, StringConstraints(strip_whitespace=True, min_length=1, max_length=100) +] class Turn(BaseModel): @@ -18,6 +24,15 @@ class Turn(BaseModel): created_at: datetime +class ConversationCreate(BaseModel): + # 会话归属的大屏；不传则不归属任何大屏（兼容旧调用） + screen_id: NonBlankString | None = None + + +class ConversationRename(BaseModel): + title: Title + + class ConversationCreated(BaseModel): conversation_id: str @@ -31,9 +46,11 @@ class ConversationSummary(BaseModel): """历史会话列表项，只包含已产生对话的会话""" conversation_id: str - title: str # 取自首轮用户输入，过长时截断 + title: str # 用户改过名用改的，否则取自首轮用户输入，过长时截断 + screen_id: str | None = None updated_at: datetime # 最近一轮对话的创建时间 class ConversationList(BaseModel): conversations: list[ConversationSummary] + total: int # 符合条件的会话总数（不受 limit / offset 影响），管理页翻页用 diff --git a/app/schemas/data_file.py b/app/schemas/data_file.py index 0f3849e..01c567d 100644 --- a/app/schemas/data_file.py +++ b/app/schemas/data_file.py @@ -212,7 +212,9 @@ class DataFileInfo(BaseModel): encoding: str | None size: int created_at: datetime + # 被已保存的大屏引用时固定为 True，此时 expires_at 是个遥远的占位时间，不会过期 expires_at: datetime + pinned: bool = False tables: list[DataTableInfo] warnings: list[ParseWarning] # 未启用智能分析，或这是功能上线前上传的文件时为空 @@ -229,6 +231,9 @@ class DataFileRecord(DataFileDetail): tables: list[DataTable] # 删除只做标记，记录留到过期，会话内的表名不会被新文件重复使用 deleted: bool = False + # 引用这个文件的大屏 ID。非空时文件不过期（expires_at 被推到 PINNED_EXPIRES）， + # 最后一个引用消失后重新按保留时长计时 + pinned_by: list[str] = Field(default_factory=list) UploadStatus = Literal["queued", "parsing", "ready", "failed"] diff --git a/app/schemas/saved_screen.py b/app/schemas/saved_screen.py new file mode 100644 index 0000000..e225597 --- /dev/null +++ b/app/schemas/saved_screen.py @@ -0,0 +1,175 @@ +"""保存的大屏：存储记录与接口模型。页面本身（PageSchema）按原始 JSON 存取，校验见 app/schemas/screen.py""" + +from datetime import datetime +from typing import Annotated, Any, Literal + +from pydantic import BaseModel, Field, StringConstraints + +NAME_MAX_LENGTH = 100 +DEFAULT_NAME = "未命名大屏" + +ScreenName = Annotated[ + str, + StringConstraints(strip_whitespace=True, min_length=1, max_length=NAME_MAX_LENGTH), +] +PublishStatus = Literal["published", "offline"] +ListStatus = Literal["draft", "published", "offline"] +SortField = Literal["updated_at", "created_at", "name"] + + +class PublishInfo(BaseModel): + """screens.publish：线上版本的指针，快照本身在 screen_publishes""" + + status: PublishStatus + version: int # 当前线上版本号 + revision: int # 线上版本是从哪个 revision 发出去的 + share_id: str + published_at: datetime + + +class ScreenSummary(BaseModel): + """不含页面的大屏记录，列表、元数据更新和线上读取用""" + + screen_id: str + owner_id: int + name: str + thumbnail: str | None = None + canvas: dict[str, Any] | None = None + revision: int + created_at: datetime + updated_at: datetime + deleted_at: datetime | None = None + publish: PublishInfo | None = None + + +class ScreenRecord(ScreenSummary): + page: dict[str, Any] + + +class PublishEntry(BaseModel): + screen_id: str + version: int + revision: int + published_at: datetime + published_by: int + + +class PublishSnapshot(PublishEntry): + """发布快照，只增不改""" + + page: dict[str, Any] + + +# ---- 请求 ---- + + +class ScreenCreate(BaseModel): + name: ScreenName = DEFAULT_NAME + page: dict[str, Any] | None = None + + +class ScreenSave(BaseModel): + page: dict[str, Any] + revision: int = Field(ge=0) + + +class ScreenRename(BaseModel): + name: ScreenName + + +class ScreenPublishRequest(BaseModel): + revision: int = Field(ge=0) + + +class ScreenCopyRequest(BaseModel): + name: ScreenName | None = None + + +# ---- 响应 ---- + + +class ScreenCreated(BaseModel): + screen_id: str + revision: int + + +class PublishBrief(BaseModel): + status: PublishStatus + version: int + share_id: str + published_at: datetime + + +class ScreenDetail(BaseModel): + screen_id: str + name: str + page: dict[str, Any] + revision: int + updated_at: datetime + publish: PublishBrief | None + + +class ScreenSaved(BaseModel): + revision: int + updated_at: datetime + + +class ScreenPublished(BaseModel): + version: int + share_id: str + published_at: datetime + + +class ScreenListItem(BaseModel): + screen_id: str + name: str + thumbnail_url: str | None + canvas: dict[str, Any] | None + status: ListStatus + has_unpublished_changes: bool + share_id: str | None + updated_at: datetime + published_at: datetime | None + deleted_at: datetime | None + # 在回收站里的大屏，到这个时间会被自动彻底删除 + purge_at: datetime | None + + +class ScreenList(BaseModel): + items: list[ScreenListItem] + total: int + + +class TrashEmptied(BaseModel): + count: int + + +class ScreenCopied(BaseModel): + screen_id: str + + +class ThumbnailResult(BaseModel): + thumbnail_url: str + + +class PublishHistoryItem(BaseModel): + version: int + revision: int + published_at: datetime + published_by: int + + +class PublishHistory(BaseModel): + items: list[PublishHistoryItem] + total: int + + +class ScreenRestored(BaseModel): + revision: int + + +class PublicScreen(BaseModel): + name: str + page: dict[str, Any] + version: int + published_at: datetime diff --git a/app/services/conversation.py b/app/services/conversation.py index 68f737b..db6505c 100644 --- a/app/services/conversation.py +++ b/app/services/conversation.py @@ -21,7 +21,7 @@ def make_title(prompt: str) -> str: class ConversationStore(Protocol): """内存和 MongoDB 实现同一组方法，业务代码不需要关心具体存储""" - async def create(self, user_id: int) -> str: ... + async def create(self, user_id: int, screen_id: str | None = None) -> str: ... async def exists(self, conversation_id: str, user_id: int) -> bool: """会话已签发、未过期且属于该用户；不属于时与不存在同样返回 False""" @@ -34,9 +34,21 @@ class ConversationStore(Protocol): async def save_turn(self, turn: Turn) -> None: ... async def list_by_user( - self, user_id: int, limit: int, offset: int + self, user_id: int, limit: int, offset: int, screen_id: str | None = None ) -> list[ConversationSummary]: - """该用户已产生对话的会话，按最近更新时间倒序；没有对话的空会话不返回""" + """该用户已产生对话的会话，按最近更新时间倒序；没有对话的空会话不返回；给了 screen_id 只返回该大屏的""" + ... + + async def count_by_user(self, user_id: int, screen_id: str | None = None) -> int: + """与 list_by_user 同一口径的总数""" + ... + + async def rename(self, conversation_id: str, user_id: int, title: str) -> bool: + """会话不存在或不属于该用户返回 False""" + ... + + async def delete(self, conversation_id: str, user_id: int) -> bool: + """连同对话轮次一起删除；会话不存在或不属于该用户返回 False""" ... @@ -51,6 +63,8 @@ class InMemoryConversationStore: # 会话 ID -> 创建时间；轮次按 thread_id 去重并保持首次写入顺序 self._created: dict[str, datetime] = {} self._owner: dict[str, int] = {} + self._screen: dict[str, str | None] = {} + self._titles: dict[str, str] = {} self._turns: dict[str, dict[str, Turn]] = {} def _purge_expired(self) -> None: @@ -61,15 +75,18 @@ class InMemoryConversationStore: for key, created in self._created.items() if created < deadline and not self._turns.get(key) ]: - del self._created[conversation_id] - self._owner.pop(conversation_id, None) - self._turns.pop(conversation_id, None) + self._forget(conversation_id) - async def create(self, user_id: int) -> str: + def _forget(self, conversation_id: str) -> None: + for table in (self._created, self._owner, self._screen, self._titles, self._turns): + table.pop(conversation_id, None) + + async def create(self, user_id: int, screen_id: str | None = None) -> str: self._purge_expired() conversation_id = str(uuid4()) self._created[conversation_id] = _now() self._owner[conversation_id] = user_id + self._screen[conversation_id] = screen_id return conversation_id async def exists(self, conversation_id: str, user_id: int) -> bool: @@ -94,8 +111,8 @@ class InMemoryConversationStore: turn = turn.model_copy(update={"created_at": existing.created_at}) self._turns[turn.conversation_id][turn.thread_id] = turn - async def list_by_user( - self, user_id: int, limit: int, offset: int + def _summaries( + self, user_id: int, screen_id: str | None ) -> list[ConversationSummary]: summaries = [] for conversation_id, owner in self._owner.items(): @@ -105,14 +122,37 @@ class InMemoryConversationStore: ) if owner != user_id or not turns: continue + if screen_id is not None and self._screen.get(conversation_id) != screen_id: + continue summaries.append( ConversationSummary( conversation_id=conversation_id, - title=make_title(turns[0].prompt), + title=self._titles.get(conversation_id) or make_title(turns[0].prompt), + screen_id=self._screen.get(conversation_id), updated_at=turns[-1].created_at, ) ) summaries.sort( key=lambda item: (item.updated_at, item.conversation_id), reverse=True ) - return summaries[offset : offset + limit] + return summaries + + async def list_by_user( + self, user_id: int, limit: int, offset: int, screen_id: str | None = None + ) -> list[ConversationSummary]: + return self._summaries(user_id, screen_id)[offset : offset + limit] + + async def count_by_user(self, user_id: int, screen_id: str | None = None) -> int: + return len(self._summaries(user_id, screen_id)) + + async def rename(self, conversation_id: str, user_id: int, title: str) -> bool: + if not await self.exists(conversation_id, user_id): + return False + self._titles[conversation_id] = title + return True + + async def delete(self, conversation_id: str, user_id: int) -> bool: + if not await self.exists(conversation_id, user_id): + return False + self._forget(conversation_id) + return True diff --git a/app/services/conversation_mongo.py b/app/services/conversation_mongo.py index 3b2e31a..de1323f 100644 --- a/app/services/conversation_mongo.py +++ b/app/services/conversation_mongo.py @@ -45,6 +45,8 @@ class MongoConversationStore: await self._conversations.create_index("created_at") # 按用户查询会话 await self._conversations.create_index("user_id") + # 按大屏过滤会话 + await self._conversations.create_index([("user_id", ASCENDING), ("screen_id", ASCENDING)]) async def _purge_expired(self) -> None: # 只清理没有任何轮次且超过保留时长的空会话，有对话的会话不受影响 @@ -66,12 +68,17 @@ class MongoConversationStore: if empty: await self._conversations.delete_many({"_id": {"$in": empty}}) - async def create(self, user_id: int) -> str: + async def create(self, user_id: int, screen_id: str | None = None) -> str: await self._purge_expired() conversation_id = str(uuid4()) # user_id 来自用户中心，会话只对创建者可见 await self._conversations.insert_one( - {"_id": conversation_id, "created_at": _now(), "user_id": user_id} + { + "_id": conversation_id, + "created_at": _now(), + "user_id": user_id, + "screen_id": screen_id, + } ) return conversation_id @@ -121,12 +128,13 @@ class MongoConversationStore: upsert=True, ) - async def list_by_user( - self, user_id: int, limit: int, offset: int - ) -> list[ConversationSummary]: + def _with_stats(self, user_id: int, screen_id: str | None) -> list[dict]: + match: dict = {"user_id": user_id} + if screen_id is not None: + match["screen_id"] = screen_id # $unwind 会丢弃没有任何轮次的空会话 - pipeline = [ - {"$match": {"user_id": user_id}}, + return [ + {"$match": match}, { "$lookup": { "from": self._turns.name, @@ -146,6 +154,13 @@ class MongoConversationStore: } }, {"$unwind": "$stats"}, + ] + + async def list_by_user( + self, user_id: int, limit: int, offset: int, screen_id: str | None = None + ) -> list[ConversationSummary]: + pipeline = [ + *self._with_stats(user_id, screen_id), {"$sort": {"stats.updated_at": -1, "_id": -1}}, {"$skip": offset}, {"$limit": limit}, @@ -154,8 +169,30 @@ class MongoConversationStore: return [ ConversationSummary( conversation_id=item["_id"], - title=make_title(item["stats"]["first_prompt"]), + title=item.get("title") or make_title(item["stats"]["first_prompt"]), + screen_id=item.get("screen_id"), updated_at=item["stats"]["updated_at"], ) for item in await cursor.to_list() ] + + async def count_by_user(self, user_id: int, screen_id: str | None = None) -> int: + pipeline = [*self._with_stats(user_id, screen_id), {"$count": "total"}] + cursor = await self._conversations.aggregate(pipeline) + counted = await cursor.to_list() + return counted[0]["total"] if counted else 0 + + async def rename(self, conversation_id: str, user_id: int, title: str) -> bool: + if not await self.exists(conversation_id, user_id): + return False + await self._conversations.update_one( + {"_id": conversation_id, "user_id": user_id}, {"$set": {"title": title}} + ) + return True + + async def delete(self, conversation_id: str, user_id: int) -> bool: + if not await self.exists(conversation_id, user_id): + return False + await self._turns.delete_many({"conversation_id": conversation_id}) + await self._conversations.delete_one({"_id": conversation_id, "user_id": user_id}) + return True diff --git a/app/services/datasource_proxy.py b/app/services/datasource_proxy.py new file mode 100644 index 0000000..b1325aa --- /dev/null +++ b/app/services/datasource_proxy.py @@ -0,0 +1,300 @@ +"""数据源代理：服务端替浏览器请求第三方接口 + +两个入口共用 fetch_json：编辑态由登录用户传 URL；线上态只传数据源 ID，配置取自已发布快照 +防护：只允许 http/https；DNS 解析后每个地址都必须是公网地址，并把连接钉在已检查过的地址上 +（不再让 httpx 二次解析，防 DNS 重绑定）；不读环境代理；重定向逐跳重新检查，跨域时丢掉调用方的密钥头； +总超时 5 秒，响应体上限 2 MB，按解压后的字节计，解压时带输出上限（防压缩炸弹） +""" + +import asyncio +import ipaddress +import json +import re +import socket +import time +import zlib +from collections.abc import Awaitable, Callable +from typing import Any, Literal +from urllib.parse import urljoin + +import httpx + +from app.core.exceptions import BizException + +TIMEOUT = 5.0 +MAX_BYTES = 2 * 1024 * 1024 +MAX_REDIRECTS = 3 +REDIRECT_STATUS = {301, 302, 303, 307, 308} +# 这些头由 httpx 或连接本身决定，不让调用方覆盖 +FORBIDDEN_HEADERS = { + "host", + "content-length", + "connection", + "transfer-encoding", + "upgrade", + "te", + "keep-alive", + "proxy-authorization", + "proxy-connection", + "accept-encoding", +} +# 只接受自己能带上限解压的编码；httpx 默认还会协商 br / zstd，那些解码器没有输出上限 +ACCEPT_ENCODING = "gzip, deflate" +# is_global 会放过的几段：NAT64（可以嵌入任意 IPv4）、IPv4 兼容地址、已废弃的站点本地地址 +EXTRA_BLOCKED = [ + ipaddress.ip_network(net) + for net in ("64:ff9b::/96", "64:ff9b:1::/48", "::/96", "fec0::/10") +] +# 十进制、十六进制、省略写法的 IPv4（2130706433、0x7f.1、127.1）：不是标准写法， +# 各平台的解析器解释不一，一律当作内网地址拒绝，不交给 DNS +NUMERIC_LABEL = re.compile(r"0x[0-9a-f]+|[0-9]+", re.IGNORECASE) + +Method = Literal["GET", "POST"] + + +def _blocked() -> BizException: + return BizException(422, "不允许访问内网或本机地址") + + +def _is_public(address: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: + # IPv4 映射的 IPv6 地址按内嵌的 IPv4 判断 + if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped is not None: + address = address.ipv4_mapped + return ( + address.is_global + and not address.is_multicast + and not any(address in net for net in EXTRA_BLOCKED) + ) + + +def _checked( + address: ipaddress.IPv4Address | ipaddress.IPv6Address, +) -> ipaddress.IPv4Address | ipaddress.IPv6Address: + if not _is_public(address): + raise _blocked() + return address + + +async def _resolve(host: str, port: int) -> str: + """解析主机名，所有地址都是公网才放行，返回第一个地址""" + host = host.rstrip(".") + try: + literal = ipaddress.ip_address(host) + except ValueError: + pass + else: + return str(_checked(literal)) + if host.lower() == "localhost" or host.lower().endswith(".localhost"): + raise _blocked() + if all(NUMERIC_LABEL.fullmatch(label) for label in host.split(".")): + raise _blocked() + try: + infos = await asyncio.get_running_loop().getaddrinfo( + host, port, type=socket.SOCK_STREAM + ) + except socket.gaierror: + raise BizException(502, "数据源域名无法解析") from None + addresses = [ipaddress.ip_address(info[4][0].split("%")[0]) for info in infos] + if not addresses: + raise BizException(502, "数据源域名无法解析") + for address in addresses: + _checked(address) + return str(addresses[0]) + + +def _clean_headers(headers: dict[str, Any] | None) -> dict[str, str]: + cleaned = {"Accept": "application/json", "Accept-Encoding": ACCEPT_ENCODING} + for name, value in (headers or {}).items(): + if name.lower() in FORBIDDEN_HEADERS or not isinstance( + value, (str, int, float) + ): + continue + cleaned[name] = str(value) + return cleaned + + +def _query_value(value: Any) -> Any: + return value if isinstance(value, (str, int, float, bool)) else json.dumps(value) + + +def _parse_url(raw: str) -> httpx.URL: + try: + url = httpx.URL(raw) + except httpx.InvalidURL: + raise BizException(422, "数据源地址不合法") from None + if url.scheme not in ("http", "https") or not url.host or url.userinfo: + raise BizException(422, "数据源地址只支持 http / https，且不能带账号密码") + return url + + +def _too_large() -> BizException: + return BizException(502, "数据源响应超过 2 MB") + + +async def _read_body(response: httpx.Response) -> bytes: + """读响应体：原始字节和解压后的字节都不超过 MAX_BYTES + + 不用 aiter_bytes：它对每个 64 KB 的块整块解压，没有输出上限，一个块就能膨胀到上 GB + """ + encoding = response.headers.get("content-encoding", "identity").strip().lower() + if encoding in ("", "identity"): + decoder = None + elif encoding in ("gzip", "x-gzip"): + decoder = zlib.decompressobj(16 + zlib.MAX_WBITS) + elif encoding == "deflate": + decoder = zlib.decompressobj() + else: + raise BizException(502, "数据源响应的压缩格式不支持") + declared = response.headers.get("content-length", "") + if declared.isdigit() and int(declared) > MAX_BYTES: + raise _too_large() + chunks: list[bytes] = [] + raw_size = size = 0 + try: + async for chunk in response.aiter_raw(): + raw_size += len(chunk) + if raw_size > MAX_BYTES: + raise _too_large() + if decoder is None: + chunks.append(chunk) + size += len(chunk) + continue + pending = chunk + while pending: + # max_length 为 0 表示不限，这里恒大于 0 + part = decoder.decompress(pending, MAX_BYTES - size + 1) + size += len(part) + if size > MAX_BYTES: + raise _too_large() + chunks.append(part) + pending = decoder.unconsumed_tail + if not part: + break + # 压缩流没有正常结束（被截断）也算失败，别把半截数据当成功 + if decoder is not None and not decoder.eof: + raise BizException(502, "数据源响应解压失败") + except zlib.error: + raise BizException(502, "数据源响应解压失败") from None + return b"".join(chunks) + + +async def _request( + client: httpx.AsyncClient, + method: str, + url: httpx.URL, + headers: dict[str, str], + body: Any, +) -> tuple[int, httpx.Headers, bytes]: + port = url.port or (443 if url.scheme == "https" else 80) + pinned = url.copy_with(host=await _resolve(url.host, port)) + request = client.build_request( + method, + pinned, + headers={**headers, "Host": url.netloc.decode()}, + json=body if method == "POST" else None, + # 连接用 IP，TLS 的 SNI 和证书校验仍然用原主机名 + extensions={"sni_hostname": url.host}, + ) + response = await client.send(request, stream=True) + try: + return response.status_code, response.headers, await _read_body(response) + finally: + await response.aclose() + + +def _origin(url: httpx.URL) -> tuple[str, str, int | None]: + return url.scheme, url.host, url.port or (443 if url.scheme == "https" else 80) + + +async def fetch_json( + url: str, + method: Method = "GET", + params: dict[str, Any] | None = None, + headers: dict[str, Any] | None = None, +) -> dict[str, Any] | list[Any]: + """GET 的 params 作为查询参数，POST 的 params 作为 JSON 请求体，和前端 fetchData 一致""" + target = _parse_url(url) + request_headers = _clean_headers(headers) + body = None + if method == "GET": + if params: + target = target.copy_merge_params( + {key: _query_value(value) for key, value in params.items()} + ) + else: + body = params or {} + try: + async with asyncio.timeout(TIMEOUT): + # trust_env=False：环境里的代理会绕过上面的地址检查 + # max_keepalive_connections=0：连接按 IP 复用，不关的话同一 IP 上的下一跳会沿用上一跳的 TLS 校验结果 + async with httpx.AsyncClient( + trust_env=False, + timeout=TIMEOUT, + limits=httpx.Limits(max_keepalive_connections=0), + ) as client: + for _ in range(MAX_REDIRECTS + 1): + status, response_headers, content = await _request( + client, method, target, request_headers, body + ) + location = response_headers.get("location") + if status not in REDIRECT_STATUS or not location: + break + previous = target + target = _parse_url(urljoin(str(target), location)) + if _origin(target) != _origin(previous): + # 密钥头只发给调用方指定的那个源 + request_headers = _clean_headers(None) + if status not in (307, 308): + method, body = "GET", None + else: + raise BizException(502, "数据源重定向次数过多") + except (TimeoutError, httpx.TimeoutException): + raise BizException(504, "数据源请求超时") from None + except httpx.HTTPError: + raise BizException(502, "数据源请求失败") from None + if not 200 <= status < 300: + raise BizException(502, f"数据源返回 HTTP {status}") + try: + data = json.loads(content) + except (ValueError, RecursionError): + # 2 MB 以内的深层嵌套会打满递归，同样当作非法 JSON + raise BizException(502, "数据源返回的不是 JSON") from None + if not isinstance(data, (dict, list)): + raise BizException(502, "数据源返回的不是 JSON 对象或数组") + return data + + +class FetchCache: + """线上大屏多人同时开、按 interval 轮询：同一个 key 几秒内只真正请求一次，并发的请求共用同一次 + + 失败不缓存；单实例部署，缓存放进程内存。单条最大 2 MB，条数上限乘出来就是内存上限，所以条数要小 + """ + + def __init__(self, ttl: float = 3.0, max_entries: int = 64): + self._ttl = ttl + self._max_entries = max_entries + self._entries: dict[Any, tuple[float, Any]] = {} + self._inflight: dict[Any, asyncio.Future] = {} + + async def get(self, key: Any, fetch: Callable[[], Awaitable[Any]]) -> Any: + now = time.monotonic() + hit = self._entries.get(key) + if hit is not None and hit[0] > now: + return hit[1] + future = self._inflight.get(key) + if future is None: + future = asyncio.ensure_future(fetch()) + self._inflight[key] = future + future.add_done_callback(lambda done: self._finish(key, done)) + # shield：某个请求者断开不能取消共用的那次请求 + return await asyncio.shield(future) + + def _finish(self, key: Any, future: asyncio.Future) -> None: + self._inflight.pop(key, None) + if future.cancelled() or future.exception() is not None: + return + now = time.monotonic() + # 每次写入先丢掉过期项；清完还满就不缓存这一条 + self._entries = {k: v for k, v in self._entries.items() if v[0] > now} + if len(self._entries) < self._max_entries: + self._entries[key] = (now + self._ttl, future.result()) diff --git a/app/services/screen_mongo.py b/app/services/screen_mongo.py new file mode 100644 index 0000000..0714cb1 --- /dev/null +++ b/app/services/screen_mongo.py @@ -0,0 +1,238 @@ +"""大屏存储的 MongoDB 实现，方法语义与内存实现一致 + +集合：screens（草稿，一屏一条）、screen_publishes（发布快照，只增不改） +""" + +import re +from contextlib import contextmanager +from datetime import UTC +from typing import Any + +from bson.codec_options import CodecOptions +from bson.errors import InvalidDocument +from pydantic import BaseModel +from pymongo import ASCENDING, DESCENDING, ReturnDocument +from pymongo.errors import DocumentTooLarge, DuplicateKeyError + +from app.core.exceptions import BizException +from app.schemas.saved_screen import ( + PublishEntry, + PublishSnapshot, + ScreenRecord, + ScreenSummary, +) +from app.services.screen_store import canvas_size + +# 列表和元数据更新不需要页面，只取画布；Mongo 的投影不能同时"排除 page"和"包含 page.canvas"，所以列字段 +SUMMARY_PROJECTION = { + "owner_id": 1, + "name": 1, + "thumbnail": 1, + "page.canvas": 1, + "revision": 1, + "created_at": 1, + "updated_at": 1, + "deleted_at": 1, + "publish": 1, +} + + +@contextmanager +def _storable(): + """页面里的超大整数、含 NUL 的字段名、编码后超过 16 MB 的文档，驱动会抛各种非业务异常 + + 10 MB 的 JSON 编码成 BSON 后会涨（数组的下标要当字符串键存），所以体积上限靠这里兜底 + """ + try: + yield + except (OverflowError, InvalidDocument, DocumentTooLarge): + raise BizException( + 422, "大屏数据无法保存：数字超出范围、字段名不合法或体积过大" + ) from None + + +# update 自己拼的条件，调用方的 where 不许覆盖 +RESERVED_FILTER_KEYS = {"_id", "owner_id", "deleted_at", "revision"} + + +def _owned(screen_id: str, owner_id: int, deleted: bool) -> dict[str, Any]: + return { + "_id": screen_id, + "owner_id": owner_id, + "deleted_at": {"$ne": None} if deleted else None, + } + + +def _encode(value: Any) -> Any: + return value.model_dump() if isinstance(value, BaseModel) else value + + +def _summary(document: dict[str, Any]) -> ScreenSummary: + canvas = (document.get("page") or {}).get("canvas") + data = {k: v for k, v in document.items() if k not in ("_id", "page")} + return ScreenSummary.model_validate( + {**data, "screen_id": document["_id"], "canvas": canvas_size(canvas)} + ) + + +class MongoScreenStore: + def __init__(self, db: Any, prefix: str = ""): + # 读回的时间统一带 UTC 时区；集合名前缀用于测试隔离 + options = CodecOptions(tz_aware=True, tzinfo=UTC) + self._screens = db.get_collection(f"{prefix}screens", codec_options=options) + self._publishes = db.get_collection( + f"{prefix}screen_publishes", codec_options=options + ) + + async def ensure_indexes(self) -> None: + await self._screens.create_index( + [ + ("owner_id", ASCENDING), + ("deleted_at", ASCENDING), + ("updated_at", DESCENDING), + ] + ) + # 分享链接唯一；没发布过的大屏没有 publish.share_id，不进索引 + await self._screens.create_index( + "publish.share_id", + unique=True, + partialFilterExpression={"publish.share_id": {"$type": "string"}}, + ) + await self._publishes.create_index( + [("screen_id", ASCENDING), ("version", DESCENDING)], unique=True + ) + # 回收站过期清理按 deleted_at 跨用户扫描；没删除的不进索引 + await self._screens.create_index( + "deleted_at", partialFilterExpression={"deleted_at": {"$type": "date"}} + ) + + async def insert(self, record: ScreenRecord) -> None: + document = record.model_dump(exclude={"screen_id", "canvas"}) + with _storable(): + await self._screens.insert_one({"_id": record.screen_id, **document}) + + async def get(self, screen_id, owner_id, *, deleted=False): + document = await self._screens.find_one(_owned(screen_id, owner_id, deleted)) + if document is None: + return None + return ScreenRecord.model_validate( + {**_summary(document).model_dump(), "page": document["page"]} + ) + + async def get_summary(self, screen_id, owner_id, *, deleted=False): + document = await self._screens.find_one( + _owned(screen_id, owner_id, deleted), SUMMARY_PROJECTION + ) + return None if document is None else _summary(document) + + async def update( + self, screen_id, owner_id, changes, *, revision=None, where=None, deleted=False + ): + query = _owned(screen_id, owner_id, deleted) + if revision is not None: + query["revision"] = revision + if clash := RESERVED_FILTER_KEYS & (where or {}).keys(): + raise ValueError(f"where 不能覆盖所有权和版本条件：{sorted(clash)}") + query.update(where or {}) + with _storable(): + document = await self._screens.find_one_and_update( + query, + {"$set": {key: _encode(value) for key, value in changes.items()}}, + projection=SUMMARY_PROJECTION, + return_document=ReturnDocument.AFTER, + ) + return None if document is None else _summary(document) + + async def search( + self, owner_id, *, keyword, status, sort, descending, limit, offset, deleted + ): + query: dict[str, Any] = { + "owner_id": owner_id, + "deleted_at": {"$ne": None} if deleted else None, + } + if keyword: + query["name"] = {"$regex": re.escape(keyword), "$options": "i"} + if status == "draft": + query["publish"] = None + elif status is not None: + query["publish.status"] = status + direction = DESCENDING if descending else ASCENDING + documents = await ( + self._screens.find(query, SUMMARY_PROJECTION) + .sort([(sort, direction), ("_id", direction)]) + .skip(offset) + .limit(limit) + .to_list() + ) + return [_summary(d) for d in documents], await self._screens.count_documents( + query + ) + + async def get_by_share_id(self, share_id): + document = await self._screens.find_one( + {"publish.share_id": share_id, "deleted_at": None}, SUMMARY_PROJECTION + ) + return None if document is None else _summary(document) + + async def list_expired_trash(self, deleted_before, limit): + documents = await ( + self._screens.find( + # 带 $type 才蕴含部分索引的过滤条件，查询才用得上索引 + {"deleted_at": {"$type": "date", "$lt": deleted_before}}, + SUMMARY_PROJECTION, + ) + .sort("deleted_at", ASCENDING) + .limit(limit) + .to_list() + ) + return [_summary(d) for d in documents] + + async def purge(self, screen_id, owner_id): + # 带"仍在回收站"的条件：清理期间被还原的大屏不能删 + result = await self._screens.delete_one(_owned(screen_id, owner_id, True)) + return result.deleted_count == 1 + + async def delete_publishes(self, screen_id): + await self._publishes.delete_many({"screen_id": screen_id}) + + async def last_publish_version(self, screen_id): + document = await self._publishes.find_one( + {"screen_id": screen_id}, {"version": 1}, sort=[("version", DESCENDING)] + ) + return 0 if document is None else document["version"] + + async def insert_publish(self, snapshot): + try: + with _storable(): + await self._publishes.insert_one(snapshot.model_dump()) + except DuplicateKeyError: + return False + return True + + async def get_publish(self, screen_id, version): + document = await self._publishes.find_one( + {"screen_id": screen_id, "version": version} + ) + if document is None: + return None + return PublishSnapshot.model_validate( + {k: v for k, v in document.items() if k != "_id"} + ) + + async def list_publishes(self, screen_id, limit, offset): + query = {"screen_id": screen_id} + documents = await ( + self._publishes.find(query, {"page": 0}) + .sort("version", DESCENDING) + .skip(offset) + .limit(limit) + .to_list() + ) + entries = [ + PublishEntry.model_validate({k: v for k, v in d.items() if k != "_id"}) + for d in documents + ] + return entries, await self._publishes.count_documents(query) + + async def delete_publish(self, screen_id, version): + await self._publishes.delete_one({"screen_id": screen_id, "version": version}) diff --git a/app/services/screen_store.py b/app/services/screen_store.py new file mode 100644 index 0000000..642c51b --- /dev/null +++ b/app/services/screen_store.py @@ -0,0 +1,241 @@ +"""大屏存储：协议和进程内存实现；MongoDB 实现见 screen_mongo + +内存实现只给测试用——大屏数据不能丢，没配 MongoDB 时 screens 接口直接返回 503 +所有查询都带 owner_id：别人的大屏与不存在同样返回 None +""" + +from datetime import datetime +from typing import Any, Protocol + +from app.schemas.saved_screen import ( + ListStatus, + PublishEntry, + PublishSnapshot, + ScreenRecord, + ScreenSummary, + SortField, +) + + +def canvas_size(canvas: Any) -> dict[str, Any] | None: + """列表卡片只需要画布宽高""" + if not isinstance(canvas, dict): + return None + return {key: canvas[key] for key in ("width", "height") if key in canvas} + + +def summarize(record: ScreenRecord) -> ScreenSummary: + """去掉页面，只保留画布宽高""" + data = record.model_dump(exclude={"page"}) + data["canvas"] = canvas_size(record.page.get("canvas")) + return ScreenSummary.model_validate(data) + + +class ScreenStore(Protocol): + async def insert(self, record: ScreenRecord) -> None: ... + + async def get( + self, screen_id: str, owner_id: int, *, deleted: bool = False + ) -> ScreenRecord | None: + """带页面的记录；deleted 为 True 时只找回收站里的""" + ... + + async def get_summary( + self, screen_id: str, owner_id: int, *, deleted: bool = False + ) -> ScreenSummary | None: ... + + async def update( + self, + screen_id: str, + owner_id: int, + changes: dict[str, Any], + *, + revision: int | None = None, + where: dict[str, Any] | None = None, + deleted: bool = False, + ) -> ScreenSummary | None: + """一步完成"匹配并更新"；不匹配返回 None + + 给了 revision 就要求库里的 revision 与之相等；where 是额外的等值条件，键可以带点， + 如 {"publish.version": 3}、{"publish": None}——更新是基于先前读到的值做的，值变了就不能盖上去 + """ + ... + + async def search( + self, + owner_id: int, + *, + keyword: str | None, + status: ListStatus | None, + sort: SortField, + descending: bool, + limit: int, + offset: int, + deleted: bool, + ) -> tuple[list[ScreenSummary], int]: ... + + async def get_by_share_id(self, share_id: str) -> ScreenSummary | None: + """未删除的大屏；是否已下线由调用方看 publish.status""" + ... + + async def list_expired_trash( + self, deleted_before: datetime, limit: int + ) -> list[ScreenSummary]: + """所有人的回收站里 deleted_at 早于 deleted_before 的大屏""" + ... + + async def purge(self, screen_id: str, owner_id: int) -> bool: + """彻底删除回收站里的大屏；已经被还原（不在回收站）或不存在时返回 False""" + ... + + async def delete_publishes(self, screen_id: str) -> None: ... + + async def last_publish_version(self, screen_id: str) -> int: ... + + async def insert_publish(self, snapshot: PublishSnapshot) -> bool: + """同一大屏同一版本号已存在时返回 False""" + ... + + async def get_publish( + self, screen_id: str, version: int + ) -> PublishSnapshot | None: ... + + async def list_publishes( + self, screen_id: str, limit: int, offset: int + ) -> tuple[list[PublishEntry], int]: + """版本号倒序""" + ... + + async def delete_publish(self, screen_id: str, version: int) -> None: ... + + +class InMemoryScreenStore: + def __init__(self) -> None: + self._screens: dict[str, ScreenRecord] = {} + self._publishes: dict[tuple[str, int], PublishSnapshot] = {} + + def _find( + self, screen_id: str, owner_id: int, deleted: bool + ) -> ScreenRecord | None: + record = self._screens.get(screen_id) + if record is None or record.owner_id != owner_id: + return None + return record if (record.deleted_at is not None) == deleted else None + + async def insert(self, record: ScreenRecord) -> None: + self._screens[record.screen_id] = record.model_copy(deep=True) + + async def get(self, screen_id, owner_id, *, deleted=False): + record = self._find(screen_id, owner_id, deleted) + return None if record is None else record.model_copy(deep=True) + + async def get_summary(self, screen_id, owner_id, *, deleted=False): + record = self._find(screen_id, owner_id, deleted) + return None if record is None else summarize(record) + + async def update( + self, screen_id, owner_id, changes, *, revision=None, where=None, deleted=False + ): + record = self._find(screen_id, owner_id, deleted) + if record is None or (revision is not None and record.revision != revision): + return None + if where is not None and not _matches(record, where): + return None + updated = record.model_copy(update=changes, deep=True) + self._screens[screen_id] = updated + return summarize(updated) + + async def search( + self, owner_id, *, keyword, status, sort, descending, limit, offset, deleted + ): + records = [ + record + for record in self._screens.values() + if record.owner_id == owner_id + and (record.deleted_at is not None) == deleted + and (keyword is None or keyword.lower() in record.name.lower()) + and _status_matches(record, status) + ] + # 排序字段相同时按 ID 兜底，翻页才稳定 + records.sort(key=lambda r: (getattr(r, sort), r.screen_id), reverse=descending) + return [summarize(r) for r in records[offset : offset + limit]], len(records) + + async def get_by_share_id(self, share_id): + for record in self._screens.values(): + if ( + record.deleted_at is None + and record.publish is not None + and record.publish.share_id == share_id + ): + return summarize(record) + return None + + async def list_expired_trash(self, deleted_before, limit): + expired = sorted( + ( + r + for r in self._screens.values() + if r.deleted_at is not None and r.deleted_at < deleted_before + ), + key=lambda r: r.deleted_at, + ) + return [summarize(r) for r in expired[:limit]] + + async def purge(self, screen_id, owner_id): + if self._find(screen_id, owner_id, True) is None: + return False + del self._screens[screen_id] + return True + + async def delete_publishes(self, screen_id): + for key in [key for key in self._publishes if key[0] == screen_id]: + del self._publishes[key] + + async def last_publish_version(self, screen_id): + return max( + (version for sid, version in self._publishes if sid == screen_id), default=0 + ) + + async def insert_publish(self, snapshot): + key = (snapshot.screen_id, snapshot.version) + if key in self._publishes: + return False + self._publishes[key] = snapshot.model_copy(deep=True) + return True + + async def get_publish(self, screen_id, version): + snapshot = self._publishes.get((screen_id, version)) + return None if snapshot is None else snapshot.model_copy(deep=True) + + async def list_publishes(self, screen_id, limit, offset): + snapshots = sorted( + (s for (sid, _), s in self._publishes.items() if sid == screen_id), + key=lambda s: s.version, + reverse=True, + ) + entries = [ + PublishEntry.model_validate(s.model_dump(exclude={"page"})) + for s in snapshots[offset : offset + limit] + ] + return entries, len(snapshots) + + async def delete_publish(self, screen_id, version): + self._publishes.pop((screen_id, version), None) + + +def _matches(record: ScreenRecord, where: dict[str, Any]) -> bool: + for path, expected in where.items(): + value: Any = record + for part in path.split("."): + value = None if value is None else getattr(value, part, None) + if value != expected: + return False + return True + + +def _status_matches(record: ScreenSummary, status: ListStatus | None) -> bool: + if status is None: + return True + if status == "draft": + return record.publish is None + return record.publish is not None and record.publish.status == status diff --git a/app/services/screens.py b/app/services/screens.py new file mode 100644 index 0000000..8aca333 --- /dev/null +++ b/app/services/screens.py @@ -0,0 +1,562 @@ +"""大屏的业务逻辑：存取、乐观锁、发布、历史、复制、封面、线上读取和数据源取数 + +存储层只提供"匹配并更新"这类原语，404 / 409 的判断都在这里 +""" + +import asyncio +import copy +import secrets +import weakref +from collections.abc import Callable +from datetime import UTC, datetime, timedelta +from typing import Any, Protocol +from uuid import uuid4 + +from loguru import logger +from pydantic import ValidationError + +from app.core.exceptions import BizException +from app.schemas.saved_screen import ( + NAME_MAX_LENGTH, + ListStatus, + PublishEntry, + PublishInfo, + PublishSnapshot, + ScreenRecord, + ScreenSummary, + SortField, +) +from app.schemas.screen import Screen +from app.services.datasource_proxy import FetchCache, fetch_json +from app.services.screen_store import ScreenStore +from app.services.thumbnails import ThumbnailStorage + +# 事件代码和数据源转换函数是字符串形式的 JS，线上会在浏览器里执行；写入权限只给 owner，再限制长度 +SCRIPT_MAX_CHARS = 64 * 1024 +COPY_SUFFIX = " 副本" +EMPTY_PAGE = { + "canvas": {"width": 1920, "height": 1080, "backgroundColor": "#ffffff10"}, + "nodes": [], + "dataSource": [], +} + +# 回收站过期清理：一批最多清多少个，两次扫描的间隔（秒） +PURGE_BATCH = 100 +TRASH_SWEEP_INTERVAL = 3600 + +STALE = "大屏已在别处修改" +# 下线 / 删除 / 换封面遇到并发修改时的重读次数 +UPDATE_ATTEMPTS = 3 + + +def _now() -> datetime: + return datetime.now(UTC) + + +class FilePinner(Protocol): + """数据文件服务里和大屏有关的两个方法""" + + async def sync_screen_files( + self, user_id: int, screen_id: str, file_ids: set[str] + ) -> None: ... + + async def release_screen_files(self, screen_id: str) -> None: ... + + +def referenced_file_ids(page: dict[str, Any]) -> set[str]: + """页面里图表绑定的数据文件：nodes[].fileQuery.file_id""" + ids = set() + nodes = page.get("nodes") + for node in nodes if isinstance(nodes, list) else []: + query = node.get("fileQuery") if isinstance(node, dict) else None + file_id = query.get("file_id") if isinstance(query, dict) else None + if isinstance(file_id, str) and file_id: + ids.add(file_id) + return ids + + +def _publish_is(publish: PublishInfo | None) -> dict[str, Any]: + """ "线上指针还是我读到的那个"的更新条件""" + if publish is None: + return {"publish": None} + return {"publish.version": publish.version, "publish.status": publish.status} + + +def _not_found() -> BizException: + # 别人的大屏与不存在同样返回 404 + return BizException(404, "大屏不存在或已删除") + + +def validate_page(page: dict[str, Any]) -> dict[str, Any]: + """按 Screen 校验结构，但返回的仍是原始 JSON：model_dump 会给没传的字段补出 null，前端读回来就变了""" + try: + Screen.model_validate(page) + except ValidationError as error: + first = error.errors()[0] + where = ".".join(str(part) for part in first["loc"]) + raise BizException(422, f"大屏数据不合法：{where}：{first['msg']}") from None + _check_script_length(page) + return page + + +def _check_script_length(page: dict[str, Any]) -> None: + scripts = [ + event.get("code") + for node in page["nodes"] + for event in (node.get("events") or []) + if isinstance(event, dict) + ] + [ + source.get("transform") + for source in page.get("dataSource", []) + if isinstance(source, dict) + ] + if any(isinstance(text, str) and len(text) > SCRIPT_MAX_CHARS for text in scripts): + raise BizException( + 422, f"事件代码和数据转换函数不能超过 {SCRIPT_MAX_CHARS // 1024} KB" + ) + + +def public_page(page: dict[str, Any]) -> dict[str, Any]: + """线上页面里去掉数据源的 headers：它们常放 API key，线上取数由服务端按快照里的完整配置发起，浏览器用不着""" + sources = page.get("dataSource") + if not isinstance(sources, list): + return {**page} + return { + **page, + "dataSource": [ + {key: value for key, value in item.items() if key != "headers"} + if isinstance(item, dict) + else item + for item in sources + ], + } + + +class ScreenService: + def __init__( + self, + store: ScreenStore, + thumbnails: ThumbnailStorage, + files: FilePinner | None = None, + trash_days: int = 30, + ): + self._store = store + self._thumbnails = thumbnails + # 已保存的大屏引用的数据文件不能过期，否则"刷新数据"会 404；没有文件服务时不管 + self._files = files + if trash_days < 1: + raise ValueError("trash_days 必须是正整数") + self.trash_days = trash_days + # 按大屏的锁：保存 / 回滚 / 删除 / 还原都是"改库再同步固定的文件"两步， + # 两个请求交错会让固定的文件跟着先开始、后结束的那个请求的旧页面走。单实例部署，进程内锁就够 + self._locks: weakref.WeakValueDictionary[str, asyncio.Lock] = ( + weakref.WeakValueDictionary() + ) + self._datasource_cache = FetchCache() + + def _lock(self, screen_id: str) -> asyncio.Lock: + lock = self._locks.get(screen_id) + if lock is None: + lock = self._locks[screen_id] = asyncio.Lock() + return lock + + async def _pin_files(self, owner_id: int, screen_id: str, page: dict[str, Any]): + """让数据文件跟着大屏的引用走。主操作已经成功，这一步失败只记日志，不让保存报错""" + if self._files is None: + return + try: + await self._files.sync_screen_files( + owner_id, screen_id, referenced_file_ids(page) + ) + except Exception as error: + logger.opt(exception=error).error( + "同步大屏 {} 引用的数据文件失败", screen_id + ) + + async def _release_files(self, screen_id: str) -> None: + if self._files is None: + return + try: + await self._files.release_screen_files(screen_id) + except Exception as error: + logger.opt(exception=error).error( + "放开大屏 {} 固定的数据文件失败", screen_id + ) + + # ---- 草稿 ---- + + async def create( + self, owner_id: int, name: str, page: dict[str, Any] | None + ) -> ScreenRecord: + now = _now() + record = ScreenRecord( + screen_id=f"s_{uuid4().hex}", + owner_id=owner_id, + name=name, + page=validate_page(copy.deepcopy(EMPTY_PAGE) if page is None else page), + revision=1, + created_at=now, + updated_at=now, + ) + await self._store.insert(record) + await self._pin_files(owner_id, record.screen_id, record.page) + return record + + async def get(self, screen_id: str, owner_id: int) -> ScreenRecord: + record = await self._store.get(screen_id, owner_id) + if record is None: + raise _not_found() + return record + + async def save( + self, screen_id: str, owner_id: int, page: dict[str, Any], revision: int + ) -> ScreenSummary: + validate_page(page) + async with self._lock(screen_id): + saved = await self._update_with_revision( + screen_id, + owner_id, + {"page": page, "revision": revision + 1, "updated_at": _now()}, + revision, + ) + await self._pin_files(owner_id, screen_id, page) + return saved + + async def _update_with_revision( + self, screen_id: str, owner_id: int, changes: dict[str, Any], revision: int + ) -> ScreenSummary: + updated = await self._store.update( + screen_id, owner_id, changes, revision=revision + ) + if updated is not None: + return updated + # 没更新成功：大屏不存在是 404，存在但 revision 不一致是 409 + if await self._store.get_summary(screen_id, owner_id) is None: + raise _not_found() + raise BizException(409, STALE) + + async def rename(self, screen_id: str, owner_id: int, name: str) -> None: + # 改名不算改内容，revision 不动 + changes = {"name": name, "updated_at": _now()} + if await self._store.update(screen_id, owner_id, changes) is None: + raise _not_found() + + async def _update_from_summary( + self, + screen_id: str, + owner_id: int, + build: Callable[[ScreenSummary], tuple[dict[str, Any], dict[str, Any]] | None], + ) -> None: + """读出当前状态，据此算出修改，再带着"状态没变"的条件写回；并发改了就重读再来 + + 下线、删除要整段改写 publish，不带条件就会把并发发布写进去的新指针盖回旧值 + build 返回 (changes, where)；返回 None 表示已经是目标状态，什么都不用做 + """ + for _ in range(UPDATE_ATTEMPTS): + summary = await self._store.get_summary(screen_id, owner_id) + if summary is None: + raise _not_found() + built = build(summary) + if built is None: + return + changes, where = built + if await self._store.update(screen_id, owner_id, changes, where=where): + return + raise BizException(409, "操作冲突，请重试") + + async def delete(self, screen_id: str, owner_id: int) -> None: + def build(summary: ScreenSummary): + changes: dict[str, Any] = {"deleted_at": _now()} + if summary.publish is None: + return changes, {"publish": None} + # 线上版本同时下线，分享链接立即 404 + changes["publish"] = summary.publish.model_copy( + update={"status": "offline"} + ) + return changes, _publish_is(summary.publish) + + async with self._lock(screen_id): + await self._update_from_summary(screen_id, owner_id, build) + # 进了回收站的大屏不再固定文件，还原时再固定回来 + await self._release_files(screen_id) + + async def restore_deleted(self, screen_id: str, owner_id: int) -> None: + async with self._lock(screen_id): + record = await self._store.get(screen_id, owner_id, deleted=True) + if record is None: + raise _not_found() + changes = {"deleted_at": None} + if ( + await self._store.update(screen_id, owner_id, changes, deleted=True) + is None + ): + raise _not_found() + await self._pin_files(owner_id, screen_id, record.page) + + # ---- 回收站 ---- + + async def _purge(self, summary: ScreenSummary) -> bool: + """彻底删除回收站里的一个大屏 + + 先删大屏本身（带"仍在回收站"的条件，清理期间被还原的不会误删），再清理附属数据 + """ + if not await self._store.purge(summary.screen_id, summary.owner_id): + return False + await self._store.delete_publishes(summary.screen_id) + self._thumbnails.delete(summary.thumbnail) + await self._release_files(summary.screen_id) + return True + + async def empty_trash(self, owner_id: int) -> int: + """清空自己的回收站，返回彻底删除的个数""" + count = 0 + previous: set[str] = set() + while True: + batch, _ = await self._store.search( + owner_id, + keyword=None, + status=None, + sort="updated_at", + descending=True, + limit=PURGE_BATCH, + offset=0, + deleted=True, + ) + ids = {summary.screen_id for summary in batch} + purged = [await self._purge(summary) for summary in batch] + count += sum(purged) + # 空批才算清完；整批都没删成（被并发还原了）就再查一次， + # 但同一批连续两次都删不掉说明有东西卡住，别空转 + if not batch or (not any(purged) and ids == previous): + return count + previous = ids + + async def purge_expired_trash(self) -> int: + """彻底删除在回收站里超过保留天数的大屏（所有人的）""" + deadline = _now() - timedelta(days=self.trash_days) + count = 0 + previous: set[str] = set() + while True: + batch = await self._store.list_expired_trash(deadline, PURGE_BATCH) + ids = {summary.screen_id for summary in batch} + purged = [await self._purge(summary) for summary in batch] + count += sum(purged) + if not batch or (not any(purged) and ids == previous): + return count + previous = ids + + async def run_trash_sweeper(self) -> None: + while True: + try: + if removed := await self.purge_expired_trash(): + logger.info( + "已彻底删除 {} 个超过 {} 天的回收站大屏", + removed, + self.trash_days, + ) + except Exception as error: + logger.opt(exception=error).error("清理回收站失败") + await asyncio.sleep(TRASH_SWEEP_INTERVAL) + + # ---- 管理页 ---- + + async def list_screens( + self, + owner_id: int, + *, + keyword: str | None, + status: ListStatus | None, + sort: SortField, + descending: bool, + limit: int, + offset: int, + deleted: bool, + ) -> tuple[list[ScreenSummary], int]: + return await self._store.search( + owner_id, + keyword=keyword, + status=status, + sort=sort, + descending=descending, + limit=limit, + offset=offset, + deleted=deleted, + ) + + async def copy_screen(self, screen_id: str, owner_id: int, name: str | None) -> str: + """只复制草稿和封面，不复制发布状态和历史""" + source = await self.get(screen_id, owner_id) + base = source.name[: NAME_MAX_LENGTH - len(COPY_SUFFIX)] + copied = await self.create( + owner_id, + name or f"{base}{COPY_SUFFIX}", + copy.deepcopy(source.page), + ) + if source.thumbnail is not None: + thumbnail = self._thumbnails.copy(source.thumbnail) + if thumbnail is not None: + try: + attached = await self._store.update( + copied.screen_id, owner_id, {"thumbnail": thumbnail} + ) + except BaseException: + self._thumbnails.delete(thumbnail) + raise + if not attached: + self._thumbnails.delete(thumbnail) + return copied.screen_id + + async def set_thumbnail(self, screen_id: str, owner_id: int, data: bytes) -> str: + thumbnail = self._thumbnails.save(data) + replaced: list[str | None] = [] + + def build(summary: ScreenSummary): + # 封面只是展示用的，不算改内容：不动 revision 和 updated_at + replaced[:] = [summary.thumbnail] + return {"thumbnail": thumbnail}, {"thumbnail": summary.thumbnail} + + try: + await self._update_from_summary(screen_id, owner_id, build) + except BaseException: + self._thumbnails.delete(thumbnail) + raise + # 删的是这次真正换掉的那张；并发上传时各自换掉各自读到的，不会留下孤儿文件 + self._thumbnails.delete(replaced[0]) + return thumbnail + + def open_thumbnail(self, thumbnail_id: str): + return self._thumbnails.open(thumbnail_id) + + # ---- 发布 ---- + + async def publish( + self, screen_id: str, owner_id: int, revision: int + ) -> PublishInfo: + record = await self.get(screen_id, owner_id) + if record.revision != revision: + raise BizException(409, "大屏有未保存或已在别处修改的内容，请先保存") + previous = record.publish + snapshot = PublishSnapshot( + screen_id=screen_id, + # 版本号取已有快照的最大值加一：写快照后没来得及更新指针留下的孤儿快照也不会撞号 + version=await self._store.last_publish_version(screen_id) + 1, + revision=record.revision, + page=record.page, + published_at=_now(), + published_by=owner_id, + ) + if not await self._store.insert_publish(snapshot): + raise BizException(409, "发布冲突，请重试") + info = PublishInfo( + status="published", + version=snapshot.version, + revision=snapshot.revision, + # 第一次发布生成 share_id，之后沿用；下线再发布也沿用 + share_id=previous.share_id + if previous + else f"p_{secrets.token_urlsafe(16)}", + published_at=snapshot.published_at, + ) + # 条件：草稿没被再保存过（revision），线上指针还是读到的那个——并发发布只有一个能成， + # 否则第一次发布时两边各生成一个 share_id，先返回给前端的那个会失效 + try: + updated = await self._store.update( + screen_id, + owner_id, + {"publish": info}, + revision=revision, + where=_publish_is(previous), + ) + except BaseException: + await self._store.delete_publish(screen_id, snapshot.version) + raise + if updated is None: + await self._store.delete_publish(screen_id, snapshot.version) + raise BizException(409, STALE) + return info + + async def unpublish(self, screen_id: str, owner_id: int) -> None: + def build(summary: ScreenSummary): + if summary.publish is None or summary.publish.status == "offline": + return None + offline = summary.publish.model_copy(update={"status": "offline"}) + return {"publish": offline}, _publish_is(summary.publish) + + await self._update_from_summary(screen_id, owner_id, build) + + async def exists(self, screen_id: str, owner_id: int) -> bool: + return await self._store.get_summary(screen_id, owner_id) is not None + + async def list_publishes( + self, screen_id: str, owner_id: int, limit: int, offset: int + ) -> tuple[list[PublishEntry], int]: + if await self._store.get_summary(screen_id, owner_id) is None: + raise _not_found() + return await self._store.list_publishes(screen_id, limit, offset) + + async def restore_publish(self, screen_id: str, owner_id: int, version: int) -> int: + """把历史版本的页面覆盖回草稿，不自动发布""" + async with self._lock(screen_id): + record = await self.get(screen_id, owner_id) + snapshot = await self._store.get_publish(screen_id, version) + if snapshot is None: + raise BizException(404, "发布版本不存在") + updated = await self._update_with_revision( + screen_id, + owner_id, + { + "page": snapshot.page, + "revision": record.revision + 1, + "updated_at": _now(), + }, + record.revision, + ) + await self._pin_files(owner_id, screen_id, snapshot.page) + return updated.revision + + # ---- 线上（匿名） ---- + + async def _live(self, share_id: str) -> tuple[ScreenSummary, PublishSnapshot]: + """线上版本：已下线 / 已删除 / 不存在都是 404；返回的是发布快照，不是草稿""" + summary = await self._store.get_by_share_id(share_id) + if ( + summary is None + or summary.publish is None + or summary.publish.status != "published" + ): + raise _not_found() + snapshot = await self._store.get_publish( + summary.screen_id, summary.publish.version + ) + if snapshot is None: + raise _not_found() + return summary, snapshot + + async def get_live(self, share_id: str) -> tuple[ScreenSummary, PublishSnapshot]: + return await self._live(share_id) + + async def fetch_live_datasource(self, share_id: str, datasource_id: str): + """前端只传数据源 ID，请求地址取自已发布快照；不能做成"前端传 URL 后端转发"，否则是匿名开放代理""" + _, snapshot = await self._live(share_id) + source = next( + ( + item + for item in snapshot.page.get("dataSource", []) + if isinstance(item, dict) and item.get("id") == datasource_id + ), + None, + ) + if source is None or source.get("type") != "api" or not source.get("url"): + raise BizException(404, "数据源不存在") + method = "POST" if source.get("method") == "POST" else "GET" + params = source.get("params") + headers = source.get("headers") + return await self._datasource_cache.get( + (share_id, snapshot.version, datasource_id), + lambda: fetch_json( + str(source["url"]), + method, + params if isinstance(params, dict) else None, + headers if isinstance(headers, dict) else None, + ), + ) diff --git a/app/services/thumbnails.py b/app/services/thumbnails.py new file mode 100644 index 0000000..d70f1ae --- /dev/null +++ b/app/services/thumbnails.py @@ -0,0 +1,63 @@ +"""大屏封面图：放本地磁盘，文件名就是封面 ID；每次上传换新 ID，旧图删掉 + +图片类 URL 要能被 匿名访问，靠不可猜的 ID 兜底 +""" + +import re +import secrets +from pathlib import Path + +from app.core.exceptions import BizException + +MAX_BYTES = 1024 * 1024 +# 按文件头判断类型，不信任客户端声明的 Content-Type +MAGIC = {b"\x89PNG\r\n\x1a\n": "image/png", b"\xff\xd8\xff": "image/jpeg"} +# ID 进了文件路径，读写前必须过这个格式检查，防路径穿越 +ID_PATTERN = re.compile(r"th_[A-Za-z0-9_-]{16,64}") + + +def sniff(head: bytes) -> str | None: + for magic, media_type in MAGIC.items(): + if head.startswith(magic): + return media_type + return None + + +class ThumbnailStorage: + def __init__(self, root: Path): + self._root = root + + def _path(self, thumbnail_id: str) -> Path | None: + if not ID_PATTERN.fullmatch(thumbnail_id): + return None + return self._root / thumbnail_id + + def save(self, data: bytes) -> str: + if len(data) > MAX_BYTES: + raise BizException(413, "封面图不能超过 1 MB") + if sniff(data[:8]) is None: + raise BizException(422, "封面图只支持 PNG 或 JPEG") + thumbnail_id = f"th_{secrets.token_urlsafe(16)}" + self._root.mkdir(parents=True, exist_ok=True) + self._path(thumbnail_id).write_bytes(data) + return thumbnail_id + + def open(self, thumbnail_id: str) -> tuple[Path, str] | None: + """文件路径和图片类型；ID 不合法、文件不存在或内容不是图片时返回 None""" + path = self._path(thumbnail_id) + if path is None or not path.is_file(): + return None + with path.open("rb") as file: + media_type = sniff(file.read(8)) + return None if media_type is None else (path, media_type) + + def copy(self, thumbnail_id: str) -> str | None: + path = self._path(thumbnail_id) + if path is None or not path.is_file(): + return None + return self.save(path.read_bytes()) + + def delete(self, thumbnail_id: str | None) -> None: + path = None if thumbnail_id is None else self._path(thumbnail_id) + if path is not None: + path.unlink(missing_ok=True)