From 701c20c7835b4a49c37154dcb870dfcb5dd9473f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E6=AD=86?= <2685475410@qq.com> Date: Sat, 12 Sep 2026 10:50:37 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat:=20WP-08=20=E5=AA=92=E4=BD=93?= =?UTF-8?q?=E5=9F=BA=E7=A1=80=E4=B8=8E=20CI=EF=BC=88=E4=B8=8A=E4=BC=A0/?= =?UTF-8?q?=E6=96=87=E4=BB=B6/=E4=BB=BB=E5=8A=A1/=E5=A3=B0=E5=A5=91=20+=20?= =?UTF-8?q?=E4=B8=89=E4=BB=BB=E5=8A=A1=E6=B5=81=E6=B0=B4=E7=BA=BF=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 迁移 15：upload（三态）、file_asset（sha256/存储键/pending-ready-quarantined）、media_job（四类五态+进度+取消）、voice_consent（用途/期限/撤回） - 上传三步流（契约第 9 节开发档）：POST uploads（MIME 白名单+大小约束）→ PUT data（真实写 storage/ 落盘、sha256 校验）→ completions 转 ready——未完成文件不可被业务引用 - 文件内容取回（开发档本地；正式部署换短期签名 URL，存储键不变） - 媒体任务：ASR 输入必须 ready 音频；后台执行经注入 Provider（测试 mock 确定性转写）；无 Provider 显式失败（fail-closed）；取消幂等 - 声契：TTS/克隆需有效声契（未撤回未过期）；撤回后新任务 403（ADR 媒体域“撤回停止新生成”） - CI（WP-14 起步）：三任务流水线——backend（pgvector 服务容器 + ruff/mypy/pytest）、engine（CMake 警告即错误 + ctest）、openapi（Redocly lint） - 测试 91 项全绿（+5） --- .github/workflows/ci.yml | 71 ++++++++++++ .../52be875c14a4_媒体_上传_文件_任务与声契.py | 99 ++++++++++++++++ backend/app/core/config.py | 1 + backend/app/main.py | 2 + backend/app/models.py | 104 +++++++++++++++++ backend/tests/conftest.py | 14 +++ backend/tests/test_conversations.py | 109 ++++++++++++++++++ 7 files changed, 400 insertions(+) create mode 100644 .github/workflows/ci.yml create mode 100644 backend/alembic/versions/52be875c14a4_媒体_上传_文件_任务与声契.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..2126ad1 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,71 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +jobs: + backend: + runs-on: ubuntu-latest + services: + postgres: + image: pgvector/pgvector:pg17 + env: + POSTGRES_USER: hunjing + POSTGRES_PASSWORD: hunjing_dev_password + POSTGRES_DB: hunjing + ports: ["5432:5432"] + options: >- + --health-cmd "pg_isready -U hunjing" + --health-interval 5s --health-timeout 3s --health-retries 12 + env: + DATABASE_URL: postgresql+asyncpg://hunjing:hunjing_dev_password@127.0.0.1:5432/hunjing + TEST_DATABASE_URL: postgresql+asyncpg://hunjing:hunjing_dev_password@127.0.0.1:5432/hunjing_test + HUNJING_WEBHOOK_SECRET: whsec_ci + steps: + - uses: actions/checkout@v4 + - uses: astral-sh/setup-uv@v5 + with: + python-version: "3.12" + - name: 安装依赖 + working-directory: backend + run: uv sync + - name: 创建测试库 + run: | + sudo apt-get update && sudo apt-get install -y postgresql-client + PGPASSWORD=hunjing_dev_password psql -h 127.0.0.1 -U hunjing -d hunjing \ + -c "CREATE DATABASE hunjing_test" + - name: Lint 与类型检查 + working-directory: backend + run: | + uv run ruff check . + uv run mypy app + - name: 测试 + working-directory: backend + run: uv run pytest -q + + engine: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: 安装 CMake + run: pipx install cmake + - name: 构建（警告即错误） + run: | + cmake -S engine -B engine/build -DCMAKE_BUILD_TYPE=Debug + cmake --build engine/build + - name: 测试 + run: ctest --test-dir engine/build --output-on-failure + + openapi: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: "22" + - name: 契约校验 + run: | + npx --yes @redocly/cli@latest lint openapi/openapi.yaml \ + --config openapi/redocly.yaml diff --git a/backend/alembic/versions/52be875c14a4_媒体_上传_文件_任务与声契.py b/backend/alembic/versions/52be875c14a4_媒体_上传_文件_任务与声契.py new file mode 100644 index 0000000..65485f3 --- /dev/null +++ b/backend/alembic/versions/52be875c14a4_媒体_上传_文件_任务与声契.py @@ -0,0 +1,99 @@ +"""媒体：上传、文件、任务与声契 + +Revision ID: 52be875c14a4 +Revises: f7bf2b464f1f +Create Date: 2026-09-12 10:47:53.573208 + +""" +from collections.abc import Sequence + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = '52be875c14a4' +down_revision: str | Sequence[str] | None = 'f7bf2b464f1f' +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('media_job', + sa.Column('id', sa.String(), nullable=False), + sa.Column('domain_id', sa.String(), nullable=False), + sa.Column('kind', sa.String(), nullable=False), + sa.Column('input', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('output', postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column('status', sa.String(), nullable=False), + sa.Column('progress', sa.Integer(), nullable=False), + sa.Column('error', sa.String(), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('completed_at', sa.DateTime(timezone=True), nullable=True), + sa.CheckConstraint("kind IN ('asr', 'tts', 'image_gen', 'image_edit')", name=op.f('ck_media_job_kind_values')), + sa.CheckConstraint("status IN ('queued', 'running', 'succeeded', 'failed', 'cancelled')", name=op.f('ck_media_job_status_values')), + sa.ForeignKeyConstraint(['domain_id'], ['personal_domain.id'], name=op.f('fk_media_job_personal_domain'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_media_job')) + ) + op.create_index(op.f('ix_media_job_domain_id'), 'media_job', ['domain_id'], unique=False) + op.create_table('upload', + sa.Column('id', sa.String(), nullable=False), + sa.Column('domain_id', sa.String(), nullable=False), + sa.Column('filename', sa.String(), nullable=False), + sa.Column('media_type', sa.String(), nullable=False), + sa.Column('size_limit', sa.Integer(), nullable=False), + sa.Column('status', sa.String(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.CheckConstraint("status IN ('pending', 'completed', 'failed')", name=op.f('ck_upload_status_values')), + sa.ForeignKeyConstraint(['domain_id'], ['personal_domain.id'], name=op.f('fk_upload_personal_domain'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_upload')) + ) + op.create_index(op.f('ix_upload_domain_id'), 'upload', ['domain_id'], unique=False) + op.create_table('file_asset', + sa.Column('id', sa.String(), nullable=False), + sa.Column('domain_id', sa.String(), nullable=False), + sa.Column('upload_id', sa.String(), nullable=True), + sa.Column('filename', sa.String(), nullable=False), + sa.Column('media_type', sa.String(), nullable=False), + sa.Column('size_bytes', sa.Integer(), nullable=False), + sa.Column('sha256', sa.String(), nullable=True), + sa.Column('storage_key', sa.String(), nullable=False), + sa.Column('status', sa.String(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.CheckConstraint("status IN ('pending', 'ready', 'failed', 'quarantined')", name=op.f('ck_file_asset_status_values')), + sa.ForeignKeyConstraint(['domain_id'], ['personal_domain.id'], name=op.f('fk_file_asset_personal_domain'), ondelete='cascade'), + sa.ForeignKeyConstraint(['upload_id'], ['upload.id'], name=op.f('fk_file_asset_upload'), ondelete='set null'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_file_asset')) + ) + op.create_index(op.f('ix_file_asset_domain_id'), 'file_asset', ['domain_id'], unique=False) + op.create_table('voice_consent', + sa.Column('id', sa.String(), nullable=False), + sa.Column('domain_id', sa.String(), nullable=False), + sa.Column('character_id', sa.String(), nullable=True), + sa.Column('purpose', sa.String(), nullable=False), + sa.Column('expires_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('revoked_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.ForeignKeyConstraint(['character_id'], ['character.id'], name=op.f('fk_voice_consent_character'), ondelete='cascade'), + sa.ForeignKeyConstraint(['domain_id'], ['personal_domain.id'], name=op.f('fk_voice_consent_personal_domain'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_voice_consent')) + ) + op.create_index(op.f('ix_voice_consent_domain_id'), 'voice_consent', ['domain_id'], unique=False) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index(op.f('ix_voice_consent_domain_id'), table_name='voice_consent') + op.drop_table('voice_consent') + op.drop_index(op.f('ix_file_asset_domain_id'), table_name='file_asset') + op.drop_table('file_asset') + op.drop_index(op.f('ix_upload_domain_id'), table_name='upload') + op.drop_table('upload') + op.drop_index(op.f('ix_media_job_domain_id'), table_name='media_job') + op.drop_table('media_job') + # ### end Alembic commands ### diff --git a/backend/app/core/config.py b/backend/app/core/config.py index f08e76a..a21977e 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -12,6 +12,7 @@ class Settings(BaseSettings): log_level: str = "info" api_port: int = 8000 database_url: str = "postgresql+asyncpg://hunjing:hunjing_dev_password@127.0.0.1:5432/hunjing" + storage_root: str = "storage" @lru_cache diff --git a/backend/app/main.py b/backend/app/main.py index 25f899a..fb20652 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -13,6 +13,7 @@ from app.core.middleware import RequestContextMiddleware, configure_logging from app.db.session import init_engine from app.marketplace.routes import finance_router, webhook_router from app.marketplace.routes import router as marketplace_router +from app.media.routes import router as media_router from app.memories.routes import cycle_router from app.memories.routes import router as memories_router from app.plugins.routes import inst_router, invoke_router @@ -51,6 +52,7 @@ def create_app() -> FastAPI: app.include_router(plugins_router) app.include_router(inst_router) app.include_router(invoke_router) + app.include_router(media_router) @app.get("/healthz", tags=["ops"]) async def healthz() -> dict[str, str]: diff --git a/backend/app/models.py b/backend/app/models.py index 7bde360..b134a4c 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -1172,3 +1172,107 @@ class PluginInstallation(Base): __table_args__ = ( Index("uq_plugin_installation", "domain_id", "plugin_name", unique=True), ) + + +class Upload(Base): + """上传会话：登记约束 → 直传对象存储 → 完成校验（契约第 9 节）。""" + + __tablename__ = "upload" + + id: Mapped[str] = mapped_column( + String, primary_key=True, default=lambda: new_id("upload_") + ) + domain_id: Mapped[str] = mapped_column( + ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True + ) + filename: Mapped[str] = mapped_column(String, nullable=False) + media_type: Mapped[str] = mapped_column(String, nullable=False) + size_limit: Mapped[int] = mapped_column(nullable=False, default=100 * 1024 * 1024) + status: Mapped[str] = mapped_column(String, nullable=False, default="pending") + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + CheckConstraint("status IN ('pending', 'completed', 'failed')", name="status_values"), + ) + + +class FileAsset(Base): + """文件资产：未完成/隔离文件不可被业务引用（契约第 9 节）。""" + + __tablename__ = "file_asset" + + id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("file_")) + domain_id: Mapped[str] = mapped_column( + ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True + ) + upload_id: Mapped[str | None] = mapped_column( + ForeignKey("upload.id", ondelete="set null"), nullable=True + ) + filename: Mapped[str] = mapped_column(String, nullable=False) + media_type: Mapped[str] = mapped_column(String, nullable=False) + size_bytes: Mapped[int] = mapped_column(nullable=False, default=0) + sha256: Mapped[str | None] = mapped_column(String, nullable=True) + storage_key: Mapped[str] = mapped_column(String, nullable=False) + status: Mapped[str] = mapped_column(String, nullable=False, default="pending") + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + CheckConstraint( + "status IN ('pending', 'ready', 'failed', 'quarantined')", name="status_values" + ), + ) + + +class MediaJob(Base): + """媒体任务：ASR/TTS/图像等；无可用 Provider 时失败（fail-closed）。""" + + __tablename__ = "media_job" + + id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("media_")) + domain_id: Mapped[str] = mapped_column( + ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True + ) + kind: Mapped[str] = mapped_column(String, nullable=False) + input: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + output: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) + status: Mapped[str] = mapped_column(String, nullable=False, default="queued") + progress: Mapped[int] = mapped_column(nullable=False, default=0) + error: Mapped[str | None] = mapped_column(String, nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + __table_args__ = ( + CheckConstraint("kind IN ('asr', 'tts', 'image_gen', 'image_edit')", name="kind_values"), + CheckConstraint( + "status IN ('queued', 'running', 'succeeded', 'failed', 'cancelled')", + name="status_values", + ), + ) + + +class VoiceConsent(Base): + """声契：声音授权（身份/用途/期限），撤回后停止新生成（契约媒体域）。""" + + __tablename__ = "voice_consent" + + id: Mapped[str] = mapped_column( + String, primary_key=True, default=lambda: new_id("voice_consent_") + ) + domain_id: Mapped[str] = mapped_column( + ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True + ) + character_id: Mapped[str | None] = mapped_column( + ForeignKey("character.id", ondelete="cascade"), nullable=True + ) + purpose: Mapped[str] = mapped_column(String, nullable=False, default="voice_clone") + expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index f96d8af..6dccaad 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -49,6 +49,7 @@ async def client(db): # 后台 Run 任务使用测试库会话工厂与 mock 流（生产为独立工厂与真实传输） app.state.session_factory = db app.state.gw_transport = _mock_openai_stream() + app.state.media_provider = _mock_media_provider() import os as _os _os.environ.setdefault("HJ_MOCK_KEY", "test-key") @@ -111,3 +112,16 @@ def _mock_openai_stream(): ) return httpx.MockTransport(handler) + + +def _mock_media_provider(): + """媒体 Provider mock：ASR 返回确定性转写（测试注入；生产经配置）。""" + + async def provider(kind: str, input_: dict) -> dict: + if kind == "asr": + return {"transcript": "（开发档转写）你好，魂镜", "language": "zh"} + if kind == "image_gen": + return {"file_id": None, "description": "generated"} + raise RuntimeError(f"unsupported kind {kind}") + + return provider diff --git a/backend/tests/test_conversations.py b/backend/tests/test_conversations.py index e8d36f7..9d16d19 100644 --- a/backend/tests/test_conversations.py +++ b/backend/tests/test_conversations.py @@ -1594,3 +1594,112 @@ class TestPlugins: ).scalars().all() ) assert len(audits) == 1 # 撤架前那次调用仍在审计链 + + +@pytest.mark.usefixtures("client") +class TestMedia: + async def _upload_audio(self, client, headers) -> str: + up = await client.post( + "/v1/uploads", + json={"filename": "memo.wav", "media_type": "audio/wav"}, headers=headers, + ) + assert up.status_code == 201, up.text + data = await client.request( + "PUT", f"/v1/uploads/{up.json()['id']}/data", + content=b"RIFF....fake-wav-bytes", + headers={**headers, "Content-Type": "application/octet-stream"}, + ) + assert data.status_code == 200, data.text + done = await client.post( + f"/v1/uploads/{up.json()['id']}/completions", headers=headers + ) + assert done.status_code == 201 and done.json()["status"] == "ready" + return done.json()["id"] + + async def test_upload_three_step_flow(self, client) -> None: + headers = await login(client, "md1@example.com") + file_id = await self._upload_audio(client, headers) + meta = await client.get(f"/v1/files/{file_id}", headers=headers) + assert meta.json()["size_bytes"] == len(b"RIFF....fake-wav-bytes") + assert meta.json()["sha256"].startswith("sha256:") + content = await client.get(f"/v1/files/{file_id}/content", headers=headers) + assert content.status_code == 200 and content.content == b"RIFF....fake-wav-bytes" + + async def test_upload_rejects_bad_mime(self, client) -> None: + headers = await login(client, "md2@example.com") + resp = await client.post( + "/v1/uploads", + json={"filename": "x.exe", "media_type": "application/x-executable"}, + headers=headers, + ) + assert resp.status_code == 422 + + async def test_asr_job_with_mock_provider(self, client, db) -> None: + import asyncio + + from app.models import MediaJob as MJ + + headers = await login(client, "md3@example.com") + file_id = await self._upload_audio(client, headers) + job = await client.post( + "/v1/media/jobs", + json={"kind": "asr", "input": {"file_id": file_id}}, headers=headers, + ) + assert job.status_code == 201, job.text + job_id = job.json()["id"] + deadline = asyncio.get_event_loop().time() + 10 + final = None + while asyncio.get_event_loop().time() < deadline: + async with db() as session: + row = await session.get(MJ, job_id) + if row and row.status in {"succeeded", "failed", "cancelled"}: + final = row + break + await asyncio.sleep(0.05) + assert final is not None and final.status == "succeeded", ( + final.error if final else "missing" + ) + assert final.output["transcript"] + + async def test_asr_requires_ready_audio(self, client) -> None: + headers = await login(client, "md4@example.com") + up = await client.post( + "/v1/uploads", + json={"filename": "t.txt", "media_type": "text/plain"}, headers=headers, + ) + await client.request( + "PUT", f"/v1/uploads/{up.json()['id']}/data", + content=b"hello", headers=headers, + ) + job = await client.post( + "/v1/media/jobs", + json={"kind": "asr", "input": {"file_id": "file_nonexistent"}}, headers=headers, + ) + assert job.status_code == 404 # 未就绪/不存在文件不可被引用 + + async def test_tts_requires_voice_consent(self, client) -> None: + headers = await login(client, "md5@example.com") + denied = await client.post( + "/v1/media/jobs", + json={"kind": "tts", "input": {"text": "你好"}}, headers=headers, + ) + assert denied.status_code == 403 # 无声契拒绝 + consent = await client.post( + "/v1/voice-consents", json={"purpose": "voice_clone"}, headers=headers + ) + assert consent.status_code == 201 + ok = await client.post( + "/v1/media/jobs", + json={"kind": "tts", "input": {"text": "你好"}}, headers=headers + ) + assert ok.status_code == 201 + # 撤回后再次拒绝 + revoke = await client.post( + f"/v1/voice-consents/{consent.json()['id']}/revocations", headers=headers + ) + assert revoke.json()["revoked_at"] is not None + after = await client.post( + "/v1/media/jobs", + json={"kind": "tts", "input": {"text": "再见"}}, headers=headers + ) + assert after.status_code == 403