feat(product-suite): confirm prior successful generations

This commit is contained in:
chengma
2026-07-17 10:47:44 +08:00
parent 9166cb7010
commit 9f55dcd4f0
8 changed files with 349 additions and 4 deletions
+60
View File
@@ -151,6 +151,16 @@ class ImageStudioHistoryRound:
is_legacy: bool
@dataclass(frozen=True)
class ImageStudioSuccessfulGenerationHistorySummary:
"""Successful persisted output summary for one active product project."""
project_id: int
successful_round_count: int
successful_image_count: int
latest_succeeded_at: Optional[str]
class ImageStudioError(RuntimeError):
"""Raised when the AI image studio service cannot complete an operation."""
@@ -1319,6 +1329,56 @@ def list_generation_rounds(project_id, *, limit=None, offset=0, path=None, conn=
return rounds
def get_successful_generation_history_summary(project_id, path=None, conn=None):
"""Summarize successful effective generation output for one active project."""
try:
project_id = int(project_id)
except (TypeError, ValueError) as exc:
raise db.DbError("商品项目无效") from exc
sql = """
WITH latest_round_jobs AS (
SELECT project_id,
generation_round_key,
generation_slot_index,
MAX(id) AS latest_id
FROM image_studio_jobs
WHERE generation_round_key IS NOT NULL
GROUP BY project_id, generation_round_key, generation_slot_index
),
effective_jobs AS (
SELECT jobs.*
FROM image_studio_jobs AS jobs
LEFT JOIN latest_round_jobs AS latest
ON latest.latest_id = jobs.id
WHERE jobs.generation_round_key IS NULL OR latest.latest_id IS NOT NULL
)
SELECT COUNT(DISTINCT CASE
WHEN jobs.status = 'succeeded'
THEN COALESCE(jobs.generation_round_key, '__legacy__')
END) AS successful_round_count,
SUM(CASE WHEN jobs.status = 'succeeded' THEN 1 ELSE 0 END)
AS successful_image_count,
MAX(CASE
WHEN jobs.status = 'succeeded'
THEN COALESCE(jobs.finished_at, jobs.updated_at, jobs.created_at)
END) AS latest_succeeded_at
FROM image_studio_projects AS projects
LEFT JOIN effective_jobs AS jobs ON jobs.project_id = projects.id
WHERE projects.id = ?
AND projects.deleted_at IS NULL
"""
with _connection(conn, path) as database:
row = database.execute(sql, (project_id,)).fetchone()
return ImageStudioSuccessfulGenerationHistorySummary(
project_id=project_id,
successful_round_count=int(row["successful_round_count"] or 0) if row else 0,
successful_image_count=int(row["successful_image_count"] or 0) if row else 0,
latest_succeeded_at=row["latest_succeeded_at"] if row else None,
)
def list_global_generation_rounds(
*,
account_query="",