"""章节封面是否可入队(与 Celery 任务共享,避免循环 import)。""" from __future__ import annotations from typing import Any from app.features.memoir.asset_resolver import parse_asset_refs from app.features.memoir.memoir_images.schema import ( IMAGE_STATUS_FAILED, IMAGE_STATUS_PENDING, ) # 正文内 ![...](asset://...) 数量需 **大于** 此值才生成/展示章节封面(与故事头图、正文配图任务独立) MIN_INLINE_BODY_IMAGES_FOR_CHAPTER_COVER = 3 def chapter_has_story_links(chapter: Any) -> bool: return any( getattr(link, "story", None) for link in getattr(chapter, "story_links", None) or [] ) def count_chapter_inline_body_images(chapter: Any) -> int: """统计章节 canonical_markdown 中正文插图(asset:// 图片引用)次数。""" md = getattr(chapter, "canonical_markdown", None) or "" return len(parse_asset_refs(md)) def chapter_eligible_for_cover_by_inline_body_image_count(chapter: Any) -> bool: """仅当正文内插图数量 > MIN_INLINE_BODY_IMAGES_FOR_CHAPTER_COVER 时才生成/展示章节封面。""" return ( count_chapter_inline_body_images(chapter) > MIN_INLINE_BODY_IMAGES_FOR_CHAPTER_COVER ) def primary_chapter_memoir_image(chapter: Any) -> Any | None: """章节级 MemoirImage(封面槽位):按 order_index 最小取第一条。""" imgs = sorted( getattr(chapter, "images", None) or [], key=lambda m: getattr(m, "order_index", 0), ) return imgs[0] if imgs else None def chapter_needs_cover_enqueue(chapter) -> bool: """尚无 cover_asset、有正文、且正文内 asset 插图多于阈值时,可派发 generate_chapter_cover。""" if not chapter: return False if not chapter_has_story_links(chapter): return False if getattr(chapter, "cover_asset_id", None): return False md = (getattr(chapter, "canonical_markdown", None) or "").strip() if not md: return False return chapter_eligible_for_cover_by_inline_body_image_count(chapter) def chapter_has_cover_to_generate(chapter) -> bool: """章节是否有待生成的封面图(任一条 chapter 级 MemoirImage 为 pending/failed)。""" for m in getattr(chapter, "images", None) or []: status = (m.status or "").strip() if status in (IMAGE_STATUS_PENDING, IMAGE_STATUS_FAILED): return True return False def cover_memoir_image_pending_or_failed(chapter: Any) -> Any | None: """用于补图任务:按 order_index 找到第一条 pending/failed 的章节配图行。""" images = sorted( getattr(chapter, "images", None) or [], key=lambda m: getattr(m, "order_index", 0), ) for m in images: st = (m.status or "").strip() if st in (IMAGE_STATUS_PENDING, IMAGE_STATUS_FAILED): return m return None