okooooooo/Jellyfish
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor API router imports and remove unused task imports

Browse Source
This commit is contained in:
extreme committed 2026-03-26 11:04:16 +08:00
1 parent 129b713057
commit 44ad86d455
6 files changed
+1 -458
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/app/api/v1/routes/film/__init__.py
+1 -5
View File
@@ -2,16 +2,12 @@
from __future__ import annotations
import asyncio
import json
from fastapi import APIRouter
from app.api.v1.routes.film import extract, tasks_entities, generated_video, generated_image, tasks_images, task_status
from app.api.v1.routes.film import generated_video, generated_image, tasks_images, task_status
router = APIRouter()
router.include_router(extract.router)
router.include_router(tasks_entities.router)
router.include_router(generated_video.router)
router.include_router(generated_image.router)
router.include_router(tasks_images.router)
backend/app/api/v1/routes/film/extract.py
-65
View File
@@ -1,65 +0,0 @@
from __future__ import annotations
import json
from fastapi import APIRouter, Depends
from langchain_core.runnables import Runnable
from app.chains.agents import FilmEntityExtractorAgent, FilmShotlistStoryboarderAgent
from app.schemas.skills.film import FilmEntityExtractionResult, FilmShotlistResult
from app.dependencies import get_llm
from app.schemas.common import ApiResponse, success_response
from .common import EntityExtractRequest, ShotlistExtractRequest
router = APIRouter()
@router.post(
"/extract/entities",
response_model=ApiResponse[FilmEntityExtractionResult],
summary="关键信息抽取",
description="从小说文本中抽取人物、地点、道具，忠实原文、可追溯证据。",
)
def extract_entities(
body: EntityExtractRequest,
llm: Runnable = Depends(get_llm),
) -> ApiResponse[FilmEntityExtractionResult]:
"""FilmEntityExtractorAgent：人物/地点/道具抽取。"""
extractor = FilmEntityExtractorAgent(llm)
chunks_json = json.dumps(
[{"chunk_id": c.chunk_id, "text": c.text} for c in body.chunks],
ensure_ascii=False,
)
result = extractor.extract(
source_id=body.source_id,
language=body.language or "zh",
chunks_json=chunks_json,
)
return success_response(result)
@router.post(
"/extract/shotlist",
response_model=ApiResponse[FilmShotlistResult],
summary="分镜抽取",
description="将小说片段转为可拍镜头表（景别/机位/运镜/转场/VFX）。",
)
def extract_shotlist(
body: ShotlistExtractRequest,
llm: Runnable = Depends(get_llm),
) -> ApiResponse[FilmShotlistResult]:
"""FilmShotlistStoryboarderAgent：场景/镜头/转场抽取。"""
storyboarder = FilmShotlistStoryboarderAgent(llm)
chunks_json = json.dumps(
[{"chunk_id": c.chunk_id, "text": c.text} for c in body.chunks],
ensure_ascii=False,
)
result = storyboarder.extract(
source_id=body.source_id,
source_title=body.source_title or "",
language=body.language or "zh",
chunks_json=chunks_json,
)
return success_response(result)
backend/app/api/v1/routes/film/tasks_entities.py
-162
View File
@@ -1,162 +0,0 @@
from __future__ import annotations
import json
from fastapi import APIRouter, Depends
from langchain_core.runnables import Runnable
from sqlalchemy.ext.asyncio import AsyncSession
from app.chains.agents import FilmEntityExtractorAgent, FilmShotlistStoryboarderAgent
from app.core.db import async_session_maker
from app.core.task_manager import DeliveryMode, SqlAlchemyTaskStore, TaskManager
from app.core.task_manager.types import TaskStatus
from app.dependencies import get_db, get_llm
from app.schemas.common import ApiResponse, success_response
from .common import (
EntityExtractTaskRequest,
ShotlistExtractTaskRequest,
TaskCreated,
_CreateOnlyTask,
bind_task,
ensure_single_bind_target,
)
router = APIRouter()
@router.post(
"/tasks/entities",
response_model=ApiResponse[TaskCreated],
status_code=201,
summary="关键信息抽取（任务版）",
)
async def create_entity_extract_task(
body: EntityExtractTaskRequest,
llm: Runnable = Depends(get_llm),
db: AsyncSession = Depends(get_db),
) -> ApiResponse[TaskCreated]:
target_type, target_id = ensure_single_bind_target(body)
store = SqlAlchemyTaskStore(db)
tm = TaskManager(store=store, strategies={})
chunks_json = json.dumps([{"chunk_id": c.chunk_id, "text": c.text} for c in body.chunks], ensure_ascii=False)
task_record = await tm.create(
task=_CreateOnlyTask(),
mode=DeliveryMode.async_polling,
run_args={
"source_id": body.source_id,
"language": body.language or "zh",
"chunks_json": chunks_json,
},
)
await bind_task(
db,
task_id=task_record.id,
target_type=target_type,
target_id=target_id,
relation_type="entities",
)
async def _runner(task_id: str, input_dict: dict) -> None:
async with async_session_maker() as session:
try:
store2 = SqlAlchemyTaskStore(session)
await store2.set_status(task_id, TaskStatus.running)
await store2.set_progress(task_id, 10)
extractor = FilmEntityExtractorAgent(llm)
result = await extractor.aextract(**input_dict)
await store2.set_result(task_id, result.model_dump())
await store2.set_progress(task_id, 100)
await store2.set_status(task_id, TaskStatus.succeeded)
await session.commit()
except Exception as exc: # noqa: BLE001
await session.rollback()
async with async_session_maker() as s2:
store3 = SqlAlchemyTaskStore(s2)
await store3.set_error(task_id, str(exc))
await store3.set_status(task_id, TaskStatus.failed)
await s2.commit()
import asyncio
asyncio.create_task(
_runner(
task_record.id,
{"source_id": body.source_id, "language": body.language or "zh", "chunks_json": chunks_json},
)
)
return success_response(TaskCreated(task_id=task_record.id), code=201)
@router.post(
"/tasks/shotlist",
response_model=ApiResponse[TaskCreated],
status_code=201,
summary="分镜抽取（任务版）",
)
async def create_shotlist_task(
body: ShotlistExtractTaskRequest,
llm: Runnable = Depends(get_llm),
db: AsyncSession = Depends(get_db),
) -> ApiResponse[TaskCreated]:
target_type, target_id = ensure_single_bind_target(body)
store = SqlAlchemyTaskStore(db)
tm = TaskManager(store=store, strategies={})
chunks_json = json.dumps([{"chunk_id": c.chunk_id, "text": c.text} for c in body.chunks], ensure_ascii=False)
task_record = await tm.create(
task=_CreateOnlyTask(),
mode=DeliveryMode.async_polling,
run_args={
"source_id": body.source_id,
"source_title": body.source_title or "",
"language": body.language or "zh",
"chunks_json": chunks_json,
},
)
await bind_task(
db,
task_id=task_record.id,
target_type=target_type,
target_id=target_id,
relation_type="shotlist",
)
async def _runner(task_id: str, input_dict: dict) -> None:
async with async_session_maker() as session:
try:
store2 = SqlAlchemyTaskStore(session)
await store2.set_status(task_id, TaskStatus.running)
await store2.set_progress(task_id, 10)
storyboarder = FilmShotlistStoryboarderAgent(llm)
result = await storyboarder.aextract(**input_dict)
await store2.set_result(task_id, result.model_dump())
await store2.set_progress(task_id, 100)
await store2.set_status(task_id, TaskStatus.succeeded)
await session.commit()
except Exception as exc: # noqa: BLE001
await session.rollback()
async with async_session_maker() as s2:
store3 = SqlAlchemyTaskStore(s2)
await store3.set_error(task_id, str(exc))
await store3.set_status(task_id, TaskStatus.failed)
await s2.commit()
import asyncio
asyncio.create_task(
_runner(
task_record.id,
{
"source_id": body.source_id,
"source_title": body.source_title or "",
"language": body.language or "zh",
"chunks_json": chunks_json,
},
)
)
return success_response(TaskCreated(task_id=task_record.id), code=201)
backend/app/core/tasks/__init__.py
-8
View File
@@ -11,11 +11,6 @@ from app.core.tasks.image_generation_tasks import (
ImageGenerationResult,
ImageGenerationTask,
)
from app.core.tasks.agent_tasks import (
FilmEntityExtractionTask2,
FilmShotlistTask2,
ShotFramePromptTask,
)
__all__ = [
"ProviderConfig",
@@ -25,7 +20,4 @@ __all__ = [
"ImageGenerationInput",
"ImageGenerationResult",
"ImageGenerationTask",
"FilmEntityExtractionTask2",
"FilmShotlistTask2",
"ShotFramePromptTask",
]
backend/app/core/tasks/agent_tasks.py
-126
View File
@@ -1,126 +0,0 @@
"""将 chains/agents 的能力封装为 BaseTask 任务。
说明：
- 这些 Task 采用 async_result 模式：run() 返回 None，结果通过 get_result() 获取。
- 目的：让上层可以用统一的 TaskManager/存储模型编排生成任务。
"""
from __future__ import annotations
from typing import Any, AsyncIterator, Optional
from app.chains.agents import (
FilmEntityExtractorAgent,
FilmShotlistStoryboarderAgent,
ShotFirstFramePromptAgent,
ShotKeyFramePromptAgent,
ShotLastFramePromptAgent,
)
from app.schemas.skills.film import FilmEntityExtractionResult, FilmShotlistResult
from app.schemas.skills.shot_frame_prompt import ShotFramePromptResult
from app.core.task_manager.types import BaseTask
class _BaseAsyncResultTask(BaseTask):
"""通用基类：保存 _result/_error，并提供通用 status/is_done/get_result。"""
TASK_KEY: str = ""
def __init__(self) -> None:
self._error: str = ""
async def status(self) -> dict[str, Any]: # type: ignore[override]
return {
"task": self.TASK_KEY,
"done": await self.is_done(),
"has_result": (await self.get_result()) is not None,
"error": self._error,
}
async def is_done(self) -> bool: # type: ignore[override]
return (await self.get_result()) is not None or bool(self._error)
class FilmEntityExtractionTask2(_BaseAsyncResultTask):
"""人物/地点/道具抽取任务（基于 FilmEntityExtractorAgent）。"""
TASK_KEY = "film_entity_extraction"
def __init__(
self,
extractor: FilmEntityExtractorAgent,
*,
input_dict: dict[str, Any],
) -> None:
super().__init__()
self._extractor = extractor
self._input_dict = input_dict
self._result: FilmEntityExtractionResult | None = None
async def run(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any] | None: # type: ignore[override]
try:
self._result = await self._extractor.aextract(**self._input_dict)
except Exception as exc: # noqa: BLE001
self._error = str(exc)
self._result = None
return None
async def get_result(self) -> FilmEntityExtractionResult | None: # type: ignore[override]
return self._result
class FilmShotlistTask2(_BaseAsyncResultTask):
"""分镜/镜头表抽取任务（基于 FilmShotlistStoryboarderAgent）。"""
TASK_KEY = "film_shotlist"
def __init__(
self,
storyboarder: FilmShotlistStoryboarderAgent,
*,
input_dict: dict[str, Any],
) -> None:
super().__init__()
self._storyboarder = storyboarder
self._input_dict = input_dict
self._result: FilmShotlistResult | None = None
async def run(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any] | None: # type: ignore[override]
try:
self._result = await self._storyboarder.aextract(**self._input_dict)
except Exception as exc: # noqa: BLE001
self._error = str(exc)
self._result = None
return None
async def get_result(self) -> FilmShotlistResult | None: # type: ignore[override]
return self._result
class ShotFramePromptTask(_BaseAsyncResultTask):
"""镜头分镜帧提示词生成任务（首帧/尾帧/关键帧共用）。"""
TASK_KEY = "shot_frame_prompt"
def __init__(
self,
agent: ShotFirstFramePromptAgent | ShotLastFramePromptAgent | ShotKeyFramePromptAgent,
*,
input_dict: dict[str, Any],
) -> None:
super().__init__()
self._agent = agent
self._input_dict = input_dict
self._result: ShotFramePromptResult | None = None
async def run(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any] | None: # type: ignore[override]
try:
self._result = await self._agent.aextract(**self._input_dict)
except Exception as exc: # noqa: BLE001
self._error = str(exc)
self._result = None
return None
async def get_result(self) -> ShotFramePromptResult | None: # type: ignore[override]
return self._result
backend/tests/test_skills_runtime_tasks.py
-92
View File
@@ -1,92 +0,0 @@
from __future__ import annotations
import pytest
from app.core.tasks.agent_tasks import FilmEntityExtractionTask2, FilmShotlistTask2
from app.schemas.skills.film import FilmEntityExtractionResult, FilmShotlistResult
class _StubEntityAgent:
async def aextract(self, **kwargs): # noqa: ANN003
return FilmEntityExtractionResult.model_validate(
{
"source_id": kwargs.get("source_id", "novel_ch01"),
"language": kwargs.get("language", "zh"),
"chunks": [],
"characters": [],
"locations": [],
"props": [],
"notes": [],
"uncertainties": [],
}
)
class _StubShotlistAgent:
async def aextract(self, **kwargs): # noqa: ANN003
return FilmShotlistResult.model_validate(
{
"breakdown": {
"source_id": kwargs.get("source_id", "novel_ch01"),
"chunks": [],
"characters": [],
"locations": [],
"props": [],
"scenes": [],
"shots": [],
"transitions": [],
"notes": [],
"uncertainties": [],
}
}
)
@pytest.mark.asyncio
async def test_film_entity_extraction_task_async_result() -> None:
extractor = _StubEntityAgent()
task = FilmEntityExtractionTask2(
extractor,
input_dict={"source_id": "novel_ch01", "language": "zh", "chunks_json": "[]"},
)
assert await task.is_done() is False
assert await task.get_result() is None
await task.run()
assert await task.is_done() is True
result = await task.get_result()
assert result is not None
assert result.source_id == "novel_ch01"
st = await task.status()
assert st["done"] is True
assert st["has_result"] is True
assert st["error"] == ""
@pytest.mark.asyncio
async def test_film_shotlist_task_async_result() -> None:
storyboarder = _StubShotlistAgent()
task = FilmShotlistTask2(
storyboarder,
input_dict={"source_id": "novel_ch01", "source_title": "", "language": "zh", "chunks_json": "[]"},
)
await task.run()
result = await task.get_result()
assert result is not None
assert result.breakdown.source_id == "novel_ch01"
@pytest.mark.asyncio
async def test_task_records_error_when_skill_invalid() -> None:
extractor = _StubEntityAgent()
task = FilmEntityExtractionTask2(
extractor,
input_dict={"source_id": "novel_ch01", "language": "zh", "chunks_json": "[]"},
)
# 这里不再测试“无效 skill_id”，因为动态 skill 机制已被移除。
await task.run()
assert await task.get_result() is not None