803 lines
33 KiB
Python
803 lines
33 KiB
Python
"""SQLAlchemy-backed task-analysis store.
|
|
|
|
Mirrors the memory implementation's normalisation: JSON payloads
|
|
(chain_snapshot / stages / seat_results / blocks / direction_notes) are stored
|
|
as LONGTEXT JSON strings; datetimes are ISO strings in dict form.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import delete, func, select, update
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from deerflow.persistence.task_analysis.base import TaskAnalysisStore
|
|
from deerflow.persistence.task_analysis.model import (
|
|
TaskAnalysisBindingRow,
|
|
TaskAnalysisContentVersionRow,
|
|
TaskAnalysisRunMessageRow,
|
|
TaskAnalysisRunRow,
|
|
TaskAnalysisSettingRow,
|
|
TaskAnalysisTaskRow,
|
|
TaskDataBlockDefRow,
|
|
TaskDataBlockRow,
|
|
TaskDataPeriodRow,
|
|
TaskDataSubmissionRow,
|
|
)
|
|
|
|
|
|
def _iso(value: datetime | str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
return value.isoformat()
|
|
return str(value)
|
|
|
|
|
|
def _utcnow() -> datetime:
|
|
return datetime.now(UTC)
|
|
|
|
|
|
def _parse_dt(value: Any) -> datetime | None:
|
|
if value is None or isinstance(value, datetime):
|
|
return value
|
|
try:
|
|
return datetime.fromisoformat(str(value))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _dump(value: Any) -> str | None:
|
|
return None if value is None else json.dumps(value, ensure_ascii=False)
|
|
|
|
|
|
def _load(value: str | None, default: Any) -> Any:
|
|
if value is None or value == "":
|
|
return default
|
|
try:
|
|
return json.loads(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def _task_dict(row: TaskAnalysisTaskRow) -> dict[str, Any]:
|
|
data = row.to_dict()
|
|
data["created_at"] = _iso(row.created_at)
|
|
data["updated_at"] = _iso(row.updated_at)
|
|
return data
|
|
|
|
|
|
def _binding_dict(row: TaskAnalysisBindingRow) -> dict[str, Any]:
|
|
data = row.to_dict()
|
|
data["chain_snapshot"] = _load(row.chain_snapshot, {})
|
|
data["direction_notes"] = _load(row.direction_notes, [])
|
|
for key in ("next_run_at", "last_run_at", "created_at", "updated_at"):
|
|
data[key] = _iso(getattr(row, key))
|
|
return data
|
|
|
|
|
|
def _run_dict(row: TaskAnalysisRunRow) -> dict[str, Any]:
|
|
data = row.to_dict()
|
|
data["scheduled_for"] = _iso(row.scheduled_for)
|
|
data["started_at"] = _iso(row.started_at)
|
|
data["finished_at"] = _iso(row.finished_at)
|
|
data["created_at"] = _iso(row.created_at)
|
|
data["updated_at"] = _iso(row.updated_at)
|
|
for key, default in (
|
|
("stages_snapshot", []),
|
|
("stage_goals_snapshot", []),
|
|
("seat_names_snapshot", {}),
|
|
("seat_results", {}),
|
|
):
|
|
data[key] = _load(getattr(row, key), default)
|
|
return data
|
|
|
|
|
|
def _message_dict(row: TaskAnalysisRunMessageRow) -> dict[str, Any]:
|
|
data = row.to_dict()
|
|
data["created_at"] = _iso(row.created_at)
|
|
return data
|
|
|
|
|
|
def _version_dict(row: TaskAnalysisContentVersionRow) -> dict[str, Any]:
|
|
data = row.to_dict()
|
|
data["blocks"] = _load(row.blocks, {})
|
|
data["created_at"] = _iso(row.created_at)
|
|
return data
|
|
|
|
|
|
def _block_def_dict(row: TaskDataBlockDefRow) -> dict[str, Any]:
|
|
data = row.to_dict()
|
|
data["created_at"] = _iso(row.created_at)
|
|
data["updated_at"] = _iso(row.updated_at)
|
|
return data
|
|
|
|
|
|
def _period_dict(row: TaskDataPeriodRow) -> dict[str, Any]:
|
|
data = row.to_dict()
|
|
data["opened_at"] = _iso(row.opened_at)
|
|
data["created_at"] = _iso(row.created_at)
|
|
return data
|
|
|
|
|
|
def _task_block_dict(row: TaskDataBlockRow) -> dict[str, Any]:
|
|
data = row.to_dict()
|
|
data["payload"] = _load(row.payload, {})
|
|
data["updated_at"] = _iso(row.updated_at)
|
|
data["created_at"] = _iso(row.created_at)
|
|
return data
|
|
|
|
|
|
def _submission_dict(row: TaskDataSubmissionRow) -> dict[str, Any]:
|
|
data = row.to_dict()
|
|
data["payload"] = _load(row.payload, {})
|
|
data["created_at"] = _iso(row.created_at)
|
|
return data
|
|
|
|
|
|
class TaskAnalysisRepository(TaskAnalysisStore):
|
|
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
|
|
self._sf = session_factory
|
|
|
|
# ── tasks ───────────────────────────────────────────────────────────────
|
|
|
|
async def create_task(self, data: dict[str, Any]) -> dict[str, Any]:
|
|
source = str(data["source_task_id"])
|
|
async with self._sf() as session:
|
|
existing = await session.execute(
|
|
select(TaskAnalysisTaskRow).where(TaskAnalysisTaskRow.source_task_id == source)
|
|
)
|
|
if existing.scalar_one_or_none() is not None:
|
|
raise ValueError(f"source_task_id already exists: {source}")
|
|
row = TaskAnalysisTaskRow(
|
|
id=str(data.get("id") or f"ta-{source}"),
|
|
source_task_id=source,
|
|
name=str(data.get("name") or ""),
|
|
description=data.get("description"),
|
|
)
|
|
session.add(row)
|
|
# 建任务即种 6 条数据块定义（块 key 定死，标题后续可改）。
|
|
from deerflow.persistence.task_analysis.data_blocks import default_block_defs
|
|
|
|
for definition in default_block_defs():
|
|
session.add(
|
|
TaskDataBlockDefRow(
|
|
id=f"tdbd-{row.id}-{definition['block_key']}",
|
|
task_id=row.id,
|
|
block_key=definition["block_key"],
|
|
title=definition["title"],
|
|
sort_order=definition["sort_order"],
|
|
)
|
|
)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return _task_dict(row)
|
|
|
|
async def get_task(self, task_id: str) -> dict[str, Any] | None:
|
|
async with self._sf() as session:
|
|
row = await session.get(TaskAnalysisTaskRow, task_id)
|
|
return _task_dict(row) if row else None
|
|
|
|
async def get_task_by_source(self, source_task_id: str) -> dict[str, Any] | None:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskAnalysisTaskRow).where(TaskAnalysisTaskRow.source_task_id == source_task_id)
|
|
)
|
|
row = result.scalar_one_or_none()
|
|
return _task_dict(row) if row else None
|
|
|
|
async def list_tasks(self) -> list[dict[str, Any]]:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskAnalysisTaskRow).order_by(TaskAnalysisTaskRow.created_at.desc())
|
|
)
|
|
tasks = [_task_dict(row) for row in result.scalars()]
|
|
bindings = await session.execute(select(TaskAnalysisBindingRow))
|
|
by_task: dict[str, dict[str, Any]] = {}
|
|
for row in bindings.scalars():
|
|
by_task[row.external_task_id] = _binding_dict(row)
|
|
for task in tasks:
|
|
task["binding"] = by_task.get(task["id"])
|
|
return tasks
|
|
|
|
async def delete_task_cascade(self, task_id: str) -> bool:
|
|
async with self._sf() as session:
|
|
row = await session.get(TaskAnalysisTaskRow, task_id)
|
|
if row is None:
|
|
return False
|
|
run_ids = (
|
|
(
|
|
await session.execute(
|
|
select(TaskAnalysisRunRow.id).where(TaskAnalysisRunRow.external_task_id == task_id)
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
await session.execute(
|
|
delete(TaskAnalysisBindingRow).where(TaskAnalysisBindingRow.external_task_id == task_id)
|
|
)
|
|
await session.execute(
|
|
delete(TaskDataBlockDefRow).where(TaskDataBlockDefRow.task_id == task_id)
|
|
)
|
|
await session.execute(
|
|
delete(TaskDataPeriodRow).where(TaskDataPeriodRow.task_id == task_id)
|
|
)
|
|
await session.execute(
|
|
delete(TaskDataBlockRow).where(TaskDataBlockRow.task_id == task_id)
|
|
)
|
|
await session.execute(
|
|
delete(TaskDataSubmissionRow).where(TaskDataSubmissionRow.task_id == task_id)
|
|
)
|
|
if run_ids:
|
|
await session.execute(
|
|
delete(TaskAnalysisRunMessageRow).where(TaskAnalysisRunMessageRow.run_id.in_(run_ids))
|
|
)
|
|
await session.execute(
|
|
delete(TaskAnalysisContentVersionRow).where(
|
|
TaskAnalysisContentVersionRow.run_id.in_(run_ids)
|
|
)
|
|
)
|
|
await session.execute(
|
|
delete(TaskAnalysisRunRow).where(TaskAnalysisRunRow.external_task_id == task_id)
|
|
)
|
|
await session.delete(row)
|
|
await session.commit()
|
|
return True
|
|
|
|
# ── bindings ────────────────────────────────────────────────────────────
|
|
|
|
async def get_binding(self, external_task_id: str) -> dict[str, Any] | None:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskAnalysisBindingRow).where(
|
|
TaskAnalysisBindingRow.external_task_id == external_task_id
|
|
)
|
|
)
|
|
row = result.scalar_one_or_none()
|
|
return _binding_dict(row) if row else None
|
|
|
|
async def get_binding_by_id(self, binding_id: str) -> dict[str, Any] | None:
|
|
async with self._sf() as session:
|
|
row = await session.get(TaskAnalysisBindingRow, binding_id)
|
|
return _binding_dict(row) if row else None
|
|
|
|
async def upsert_binding(self, external_task_id: str, data: dict[str, Any]) -> dict[str, Any]:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskAnalysisBindingRow).where(
|
|
TaskAnalysisBindingRow.external_task_id == external_task_id
|
|
)
|
|
)
|
|
row = result.scalar_one_or_none()
|
|
if row is None:
|
|
row = TaskAnalysisBindingRow(
|
|
id=str(data.get("id") or f"tab-{external_task_id}"),
|
|
external_task_id=external_task_id,
|
|
chain_id=str(data["chain_id"]),
|
|
created_by=data.get("created_by"),
|
|
)
|
|
session.add(row)
|
|
row.chain_id = str(data.get("chain_id") or row.chain_id)
|
|
row.chain_snapshot = _dump(data.get("chain_snapshot") or {})
|
|
row.cron_expr = str(data.get("cron_expr") or row.cron_expr or "")
|
|
row.timezone = str(data.get("timezone") or row.timezone or "Asia/Shanghai")
|
|
if "schedule_type" in data:
|
|
row.schedule_type = str(data.get("schedule_type") or "cron")
|
|
row.model = data.get("model")
|
|
if "enabled" in data:
|
|
row.enabled = bool(data["enabled"])
|
|
if "next_run_at" in data:
|
|
row.next_run_at = _parse_dt(data.get("next_run_at"))
|
|
row.coordinator_prompt = data.get("coordinator_prompt")
|
|
if "direction_notes" in data:
|
|
row.direction_notes = _dump(list(data["direction_notes"] or []))
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return _binding_dict(row)
|
|
|
|
async def delete_binding(self, external_task_id: str) -> bool:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
delete(TaskAnalysisBindingRow).where(
|
|
TaskAnalysisBindingRow.external_task_id == external_task_id
|
|
)
|
|
)
|
|
await session.commit()
|
|
return bool(result.rowcount)
|
|
|
|
async def due_bindings(self, now: datetime, *, limit: int = 20) -> list[dict[str, Any]]:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskAnalysisBindingRow)
|
|
.where(
|
|
TaskAnalysisBindingRow.enabled.is_(True),
|
|
TaskAnalysisBindingRow.next_run_at.is_not(None),
|
|
TaskAnalysisBindingRow.next_run_at <= now,
|
|
)
|
|
.order_by(TaskAnalysisBindingRow.next_run_at.asc())
|
|
.limit(limit)
|
|
)
|
|
return [_binding_dict(row) for row in result.scalars()]
|
|
|
|
async def update_binding(self, binding_id: str, patch: dict[str, Any]) -> dict[str, Any] | None:
|
|
allowed = {
|
|
"chain_id", "cron_expr", "timezone", "schedule_type", "enabled", "next_run_at",
|
|
"last_run_at", "last_status", "coordinator_prompt", "model",
|
|
}
|
|
values: dict[str, Any] = {k: v for k, v in patch.items() if k in allowed}
|
|
if "next_run_at" in values:
|
|
values["next_run_at"] = _parse_dt(values["next_run_at"])
|
|
if "last_run_at" in values:
|
|
values["last_run_at"] = _parse_dt(values["last_run_at"])
|
|
if "direction_notes" in patch:
|
|
values["direction_notes"] = _dump(list(patch["direction_notes"] or []))
|
|
if not values:
|
|
row = await self._get_binding_row(binding_id)
|
|
return _binding_dict(row) if row else None
|
|
async with self._sf() as session:
|
|
await session.execute(
|
|
update(TaskAnalysisBindingRow).where(TaskAnalysisBindingRow.id == binding_id).values(**values)
|
|
)
|
|
await session.commit()
|
|
row = await session.get(TaskAnalysisBindingRow, binding_id)
|
|
return _binding_dict(row) if row else None
|
|
|
|
async def _get_binding_row(self, binding_id: str) -> TaskAnalysisBindingRow | None:
|
|
async with self._sf() as session:
|
|
return await session.get(TaskAnalysisBindingRow, binding_id)
|
|
|
|
async def last_bound_chain_id(self) -> str | None:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskAnalysisBindingRow.chain_id)
|
|
.order_by(TaskAnalysisBindingRow.updated_at.desc())
|
|
.limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
# ── runs ────────────────────────────────────────────────────────────────
|
|
|
|
async def create_run_once(self, data: dict[str, Any]) -> dict[str, Any]:
|
|
binding_id = str(data["binding_id"])
|
|
scheduled_for = _parse_dt(data["scheduled_for"]) or datetime.now(UTC)
|
|
import uuid as _uuid
|
|
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskAnalysisRunRow).where(
|
|
TaskAnalysisRunRow.binding_id == binding_id,
|
|
TaskAnalysisRunRow.scheduled_for == scheduled_for,
|
|
)
|
|
)
|
|
existing = result.scalar_one_or_none()
|
|
if existing is not None:
|
|
return _run_dict(existing)
|
|
row = TaskAnalysisRunRow(
|
|
id=str(data.get("id") or f"tar-{_uuid.uuid4()}"),
|
|
binding_id=binding_id,
|
|
external_task_id=str(data["external_task_id"]),
|
|
trigger=str(data.get("trigger") or "schedule"),
|
|
scheduled_for=scheduled_for,
|
|
status=str(data.get("status") or "pending"),
|
|
stages_snapshot=_dump(data.get("stages_snapshot") or []),
|
|
stage_goals_snapshot=_dump(data.get("stage_goals_snapshot") or []),
|
|
seat_names_snapshot=_dump(data.get("seat_names_snapshot") or {}),
|
|
seat_results=_dump(data.get("seat_results") or {}),
|
|
)
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return _run_dict(row)
|
|
|
|
async def get_run(self, run_id: str) -> dict[str, Any] | None:
|
|
async with self._sf() as session:
|
|
row = await session.get(TaskAnalysisRunRow, run_id)
|
|
return _run_dict(row) if row else None
|
|
|
|
async def list_runs(self, external_task_id: str, *, limit: int = 50) -> list[dict[str, Any]]:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskAnalysisRunRow)
|
|
.where(TaskAnalysisRunRow.external_task_id == external_task_id)
|
|
.order_by(TaskAnalysisRunRow.started_at.desc())
|
|
.limit(limit)
|
|
)
|
|
return [_run_dict(row) for row in result.scalars()]
|
|
|
|
async def update_run(self, run_id: str, patch: dict[str, Any]) -> dict[str, Any] | None:
|
|
allowed = {
|
|
"trigger", "status", "finished_at", "error", "retry_count",
|
|
"stages_snapshot", "stage_goals_snapshot", "seat_names_snapshot", "seat_results",
|
|
}
|
|
values: dict[str, Any] = {}
|
|
for key, value in patch.items():
|
|
if key not in allowed:
|
|
continue
|
|
if key == "finished_at":
|
|
values[key] = _parse_dt(value)
|
|
elif key in {"stages_snapshot", "stage_goals_snapshot", "seat_names_snapshot", "seat_results"}:
|
|
values[key] = _dump(value)
|
|
else:
|
|
values[key] = value
|
|
if not values:
|
|
row = await self._get_run_row(run_id)
|
|
return _run_dict(row) if row else None
|
|
async with self._sf() as session:
|
|
await session.execute(
|
|
update(TaskAnalysisRunRow).where(TaskAnalysisRunRow.id == run_id).values(**values)
|
|
)
|
|
await session.commit()
|
|
row = await session.get(TaskAnalysisRunRow, run_id)
|
|
return _run_dict(row) if row else None
|
|
|
|
async def _get_run_row(self, run_id: str) -> TaskAnalysisRunRow | None:
|
|
async with self._sf() as session:
|
|
return await session.get(TaskAnalysisRunRow, run_id)
|
|
|
|
async def stuck_runs(self, older_than: datetime, *, limit: int = 20) -> list[dict[str, Any]]:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskAnalysisRunRow)
|
|
.where(
|
|
TaskAnalysisRunRow.status.in_(("pending", "running")),
|
|
TaskAnalysisRunRow.started_at < older_than,
|
|
)
|
|
.order_by(TaskAnalysisRunRow.started_at.asc())
|
|
.limit(limit)
|
|
)
|
|
return [_run_dict(row) for row in result.scalars()]
|
|
|
|
# ── messages ────────────────────────────────────────────────────────────
|
|
|
|
async def append_run_messages(self, run_id: str, messages: list[dict[str, Any]]) -> int:
|
|
import uuid as _uuid
|
|
|
|
async with self._sf() as session:
|
|
existing = (
|
|
(
|
|
await session.execute(
|
|
select(TaskAnalysisRunMessageRow.id).where(
|
|
TaskAnalysisRunMessageRow.run_id == run_id
|
|
)
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
base_seq = len(existing)
|
|
rows = [
|
|
TaskAnalysisRunMessageRow(
|
|
id=str(_uuid.uuid4()),
|
|
run_id=run_id,
|
|
seq=base_seq + index,
|
|
seat_agent_id=str(message.get("seat_agent_id") or ""),
|
|
stage_index=int(message.get("stage_index") or 0),
|
|
event_type=str(message.get("event_type") or "llm.ai"),
|
|
content=message.get("content"),
|
|
)
|
|
for index, message in enumerate(messages)
|
|
]
|
|
session.add_all(rows)
|
|
await session.commit()
|
|
return len(rows)
|
|
|
|
async def list_run_messages(
|
|
self, run_id: str, *, seat: str | None = None, limit: int = 500
|
|
) -> list[dict[str, Any]]:
|
|
async with self._sf() as session:
|
|
stmt = (
|
|
select(TaskAnalysisRunMessageRow)
|
|
.where(TaskAnalysisRunMessageRow.run_id == run_id)
|
|
.order_by(TaskAnalysisRunMessageRow.seq.asc())
|
|
.limit(limit)
|
|
)
|
|
if seat is not None:
|
|
stmt = stmt.where(TaskAnalysisRunMessageRow.seat_agent_id == seat)
|
|
result = await session.execute(stmt)
|
|
return [_message_dict(row) for row in result.scalars()]
|
|
|
|
# ── versions ────────────────────────────────────────────────────────────
|
|
|
|
async def create_version(self, data: dict[str, Any]) -> dict[str, Any]:
|
|
import uuid as _uuid
|
|
|
|
run_id = str(data["run_id"])
|
|
async with self._sf() as session:
|
|
await session.execute(
|
|
delete(TaskAnalysisContentVersionRow).where(
|
|
TaskAnalysisContentVersionRow.run_id == run_id
|
|
)
|
|
)
|
|
row = TaskAnalysisContentVersionRow(
|
|
id=str(_uuid.uuid4()),
|
|
external_task_id=str(data["external_task_id"]),
|
|
run_id=run_id,
|
|
blocks=_dump(data.get("blocks") or {}),
|
|
)
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return _version_dict(row)
|
|
|
|
async def list_versions(
|
|
self, external_task_id: str, *, limit: int = 50, offset: int = 0
|
|
) -> list[dict[str, Any]]:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskAnalysisContentVersionRow)
|
|
.where(TaskAnalysisContentVersionRow.external_task_id == external_task_id)
|
|
.order_by(TaskAnalysisContentVersionRow.created_at.desc())
|
|
.offset(offset)
|
|
.limit(limit)
|
|
)
|
|
return [_version_dict(row) for row in result.scalars()]
|
|
|
|
# ── data delivery: block defs / periods / submissions ────────────────────
|
|
|
|
async def ensure_block_defs(self, task_id: str) -> list[dict[str, Any]]:
|
|
from deerflow.persistence.task_analysis.data_blocks import default_block_defs
|
|
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskDataBlockDefRow).where(TaskDataBlockDefRow.task_id == task_id)
|
|
)
|
|
existing_keys = {row.block_key for row in result.scalars()}
|
|
missing = [
|
|
definition
|
|
for definition in default_block_defs()
|
|
if definition["block_key"] not in existing_keys
|
|
]
|
|
if not missing:
|
|
return await self._ordered_defs(session, task_id)
|
|
for definition in missing:
|
|
session.add(
|
|
TaskDataBlockDefRow(
|
|
id=f"tdbd-{task_id}-{definition['block_key']}",
|
|
task_id=task_id,
|
|
block_key=definition["block_key"],
|
|
title=definition["title"],
|
|
sort_order=definition["sort_order"],
|
|
)
|
|
)
|
|
try:
|
|
await session.commit()
|
|
except IntegrityError:
|
|
# 并发补种：另一个请求先种好了，沿用其结果。
|
|
await session.rollback()
|
|
return await self._ordered_defs(session, task_id)
|
|
|
|
@staticmethod
|
|
async def _ordered_defs(session: AsyncSession, task_id: str) -> list[dict[str, Any]]:
|
|
result = await session.execute(
|
|
select(TaskDataBlockDefRow)
|
|
.where(TaskDataBlockDefRow.task_id == task_id)
|
|
.order_by(TaskDataBlockDefRow.sort_order.asc(), TaskDataBlockDefRow.block_key.asc())
|
|
)
|
|
return [_block_def_dict(row) for row in result.scalars()]
|
|
|
|
async def list_block_defs(self, task_id: str) -> list[dict[str, Any]]:
|
|
async with self._sf() as session:
|
|
return await self._ordered_defs(session, task_id)
|
|
|
|
async def update_block_defs(self, task_id: str, updates: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskDataBlockDefRow).where(TaskDataBlockDefRow.task_id == task_id)
|
|
)
|
|
rows = {row.block_key: row for row in result.scalars()}
|
|
touched = False
|
|
for item in updates:
|
|
block_key = str(item.get("block_key") or "")
|
|
row = rows.get(block_key)
|
|
if row is None:
|
|
raise ValueError(f"未知数据块：{block_key}")
|
|
patch: dict[str, Any] = {}
|
|
if item.get("title") is not None:
|
|
patch["title"] = str(item["title"])
|
|
if item.get("sort_order") is not None:
|
|
patch["sort_order"] = int(item["sort_order"])
|
|
if patch:
|
|
await session.execute(
|
|
update(TaskDataBlockDefRow)
|
|
.where(TaskDataBlockDefRow.id == row.id)
|
|
.values(**patch)
|
|
)
|
|
touched = True
|
|
if touched:
|
|
await session.commit()
|
|
return await self._ordered_defs(session, task_id)
|
|
|
|
async def open_period(
|
|
self,
|
|
task_id: str,
|
|
*,
|
|
source: str,
|
|
run_id: str | None = None,
|
|
opened_at: datetime | None = None,
|
|
) -> dict[str, Any]:
|
|
import uuid as _uuid
|
|
|
|
for _attempt in range(3):
|
|
async with self._sf() as session:
|
|
if run_id:
|
|
existing = await session.execute(
|
|
select(TaskDataPeriodRow).where(TaskDataPeriodRow.run_id == run_id)
|
|
)
|
|
row = existing.scalar_one_or_none()
|
|
if row is not None:
|
|
return _period_dict(row)
|
|
max_index = (
|
|
await session.execute(
|
|
select(func.max(TaskDataPeriodRow.period_index)).where(
|
|
TaskDataPeriodRow.task_id == task_id
|
|
)
|
|
)
|
|
).scalar()
|
|
row = TaskDataPeriodRow(
|
|
id=f"tdp-{_uuid.uuid4()}",
|
|
task_id=task_id,
|
|
period_index=(max_index or 0) + 1,
|
|
opened_at=opened_at or _utcnow(),
|
|
trigger=str(source or "cron"),
|
|
run_id=run_id,
|
|
)
|
|
session.add(row)
|
|
try:
|
|
await session.commit()
|
|
except IntegrityError:
|
|
# 并发开期撞了期号唯一键（或 run_id 幂等锚点）——重查/重试。
|
|
await session.rollback()
|
|
continue
|
|
await session.refresh(row)
|
|
return _period_dict(row)
|
|
raise RuntimeError(f"open_period 连续冲突，task={task_id} source={source}")
|
|
|
|
async def get_period(self, task_id: str, period_index: int) -> dict[str, Any] | None:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskDataPeriodRow).where(
|
|
TaskDataPeriodRow.task_id == task_id,
|
|
TaskDataPeriodRow.period_index == period_index,
|
|
)
|
|
)
|
|
row = result.scalar_one_or_none()
|
|
return _period_dict(row) if row else None
|
|
|
|
async def current_period(self, task_id: str) -> dict[str, Any] | None:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskDataPeriodRow)
|
|
.where(TaskDataPeriodRow.task_id == task_id)
|
|
.order_by(TaskDataPeriodRow.period_index.desc())
|
|
.limit(1)
|
|
)
|
|
row = result.scalar_one_or_none()
|
|
return _period_dict(row) if row else None
|
|
|
|
async def list_periods(self, task_id: str, *, limit: int = 500) -> list[dict[str, Any]]:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskDataPeriodRow)
|
|
.where(TaskDataPeriodRow.task_id == task_id)
|
|
.order_by(TaskDataPeriodRow.period_index.desc())
|
|
.limit(limit)
|
|
)
|
|
rows = [_period_dict(row) for row in result.scalars()]
|
|
rows.reverse() # 时间轴语义：按期号升序返回
|
|
return rows
|
|
|
|
async def append_submission(
|
|
self,
|
|
task_id: str,
|
|
period_index: int,
|
|
block_key: str,
|
|
payload: dict[str, Any],
|
|
*,
|
|
submitted_by: str | None = None,
|
|
) -> dict[str, Any]:
|
|
import uuid as _uuid
|
|
|
|
async with self._sf() as session:
|
|
base_seq = (
|
|
(
|
|
await session.execute(
|
|
select(func.count(TaskDataSubmissionRow.id)).where(
|
|
TaskDataSubmissionRow.task_id == task_id,
|
|
TaskDataSubmissionRow.period_index == period_index,
|
|
TaskDataSubmissionRow.block_key == block_key,
|
|
)
|
|
)
|
|
)
|
|
.scalar_one()
|
|
)
|
|
row = TaskDataSubmissionRow(
|
|
id=f"tds-{_uuid.uuid4()}",
|
|
task_id=task_id,
|
|
period_index=period_index,
|
|
block_key=block_key,
|
|
payload=_dump(payload),
|
|
submitted_by=submitted_by,
|
|
seq=int(base_seq),
|
|
)
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return _submission_dict(row)
|
|
|
|
async def list_submissions(
|
|
self, task_id: str, period_index: int, block_key: str, *, limit: int = 500
|
|
) -> list[dict[str, Any]]:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskDataSubmissionRow)
|
|
.where(
|
|
TaskDataSubmissionRow.task_id == task_id,
|
|
TaskDataSubmissionRow.period_index == period_index,
|
|
TaskDataSubmissionRow.block_key == block_key,
|
|
)
|
|
.order_by(TaskDataSubmissionRow.seq.asc())
|
|
.limit(limit)
|
|
)
|
|
return [_submission_dict(row) for row in result.scalars()]
|
|
|
|
async def upsert_period_block(
|
|
self, task_id: str, period_index: int, block_key: str, payload: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
import uuid as _uuid
|
|
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskDataBlockRow).where(
|
|
TaskDataBlockRow.task_id == task_id,
|
|
TaskDataBlockRow.period_index == period_index,
|
|
TaskDataBlockRow.block_key == block_key,
|
|
)
|
|
)
|
|
row = result.scalar_one_or_none()
|
|
if row is None:
|
|
row = TaskDataBlockRow(
|
|
id=f"tdb-{_uuid.uuid4()}",
|
|
task_id=task_id,
|
|
period_index=period_index,
|
|
block_key=block_key,
|
|
payload=_dump(payload),
|
|
)
|
|
session.add(row)
|
|
else:
|
|
row.payload = _dump(payload)
|
|
row.updated_at = _utcnow()
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return _task_block_dict(row)
|
|
|
|
async def list_task_blocks(self, task_id: str, *, limit: int = 2000) -> list[dict[str, Any]]:
|
|
async with self._sf() as session:
|
|
result = await session.execute(
|
|
select(TaskDataBlockRow)
|
|
.where(TaskDataBlockRow.task_id == task_id)
|
|
.order_by(TaskDataBlockRow.period_index.asc(), TaskDataBlockRow.block_key.asc())
|
|
.limit(limit)
|
|
)
|
|
return [_task_block_dict(row) for row in result.scalars()]
|
|
|
|
# ── settings ────────────────────────────────────────────────────────────
|
|
|
|
async def get_setting(self, key: str) -> str | None:
|
|
async with self._sf() as session:
|
|
row = await session.get(TaskAnalysisSettingRow, key)
|
|
return row.value if row else None
|
|
|
|
async def set_setting(self, key: str, value: str) -> None:
|
|
async with self._sf() as session:
|
|
row = await session.get(TaskAnalysisSettingRow, key)
|
|
if row is None:
|
|
session.add(TaskAnalysisSettingRow(key=key, value=value))
|
|
else:
|
|
row.value = value
|
|
await session.commit()
|