✨ feat: WP-08 媒体基础与 CI(上传/文件/任务/声契 + 三任务流水线)

- 迁移 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)
This commit is contained in:
chenxin committed 2026-09-12 10:50:37 +08:00
1 parent 9ef2568c3b
commit 701c20c783
7 files changed
+400

No files matched your search

+71
View File
@@ -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
@@ -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 ###
+1
View File
@@ -12,6 +12,7 @@ class Settings(BaseSettings):
log_level: str = "info" log_level: str = "info"
api_port: int = 8000 api_port: int = 8000
database_url: str = "postgresql+asyncpg://hunjing:hunjing_dev_password@127.0.0.1:5432/hunjing" database_url: str = "postgresql+asyncpg://hunjing:hunjing_dev_password@127.0.0.1:5432/hunjing"
storage_root: str = "storage"
@lru_cache @lru_cache
+2
View File
@@ -13,6 +13,7 @@ from app.core.middleware import RequestContextMiddleware, configure_logging
from app.db.session import init_engine from app.db.session import init_engine
from app.marketplace.routes import finance_router, webhook_router from app.marketplace.routes import finance_router, webhook_router
from app.marketplace.routes import router as marketplace_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 cycle_router
from app.memories.routes import router as memories_router from app.memories.routes import router as memories_router
from app.plugins.routes import inst_router, invoke_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(plugins_router)
app.include_router(inst_router) app.include_router(inst_router)
app.include_router(invoke_router) app.include_router(invoke_router)
app.include_router(media_router)
@app.get("/healthz", tags=["ops"]) @app.get("/healthz", tags=["ops"])
async def healthz() -> dict[str, str]: async def healthz() -> dict[str, str]:
+104
View File
@@ -1172,3 +1172,107 @@ class PluginInstallation(Base):
__table_args__ = ( __table_args__ = (
Index("uq_plugin_installation", "domain_id", "plugin_name", unique=True), 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()
)
+14
View File
@@ -49,6 +49,7 @@ async def client(db):
# 后台 Run 任务使用测试库会话工厂与 mock 流(生产为独立工厂与真实传输) # 后台 Run 任务使用测试库会话工厂与 mock 流(生产为独立工厂与真实传输)
app.state.session_factory = db app.state.session_factory = db
app.state.gw_transport = _mock_openai_stream() app.state.gw_transport = _mock_openai_stream()
app.state.media_provider = _mock_media_provider()
import os as _os import os as _os
_os.environ.setdefault("HJ_MOCK_KEY", "test-key") _os.environ.setdefault("HJ_MOCK_KEY", "test-key")
@@ -111,3 +112,16 @@ def _mock_openai_stream():
) )
return httpx.MockTransport(handler) 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
+109
View File
@@ -1594,3 +1594,112 @@ class TestPlugins:
).scalars().all() ).scalars().all()
) )
assert len(audits) == 1 # 撤架前那次调用仍在审计链 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