285 lines
11 KiB
Python
285 lines
11 KiB
Python
"""产品推介材料的受限版式规划。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from typing import Any
|
||
|
||
from app.service.agent.bootstrap import get_model_service
|
||
from app.service.model_gateway import DatabaseModelEndpointResolver
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
ALLOWED_COVER_LAYOUTS = {
|
||
"cover_left_photo_right",
|
||
"cover_visual_right",
|
||
"cover_full_bleed",
|
||
}
|
||
ALLOWED_CONTENT_LAYOUTS = {
|
||
"single_column",
|
||
"two_columns",
|
||
"chart_focus",
|
||
"metrics_focus",
|
||
"manager_profile",
|
||
"full_width_disclosure",
|
||
"visual_focus",
|
||
}
|
||
ALLOWED_POSTER_LAYOUTS = {
|
||
"poster_balanced",
|
||
"poster_chart_focus",
|
||
"poster_manager_focus",
|
||
}
|
||
ALLOWED_BACKGROUND_THEMES = {
|
||
"data_lines",
|
||
"geometric_grid",
|
||
"soft_wave",
|
||
"none",
|
||
}
|
||
ALLOWED_DESIGN_TONES = {"institutional_tech", "warm_editorial", "quiet_data"}
|
||
ALLOWED_CONTENT_STRUCTURES = {
|
||
"balanced_split",
|
||
"strategy_right_emphasis",
|
||
"strategy_left_emphasis",
|
||
}
|
||
ALLOWED_COLUMN_RATIOS = {"40_60", "45_55", "50_50", "55_45", "60_40"}
|
||
|
||
|
||
class PromotionLayoutPlanner:
|
||
"""调用文本模型规划有限枚举，模型不可输出坐标或改写材料内容。"""
|
||
|
||
async def plan(
|
||
self,
|
||
draft: dict[str, Any],
|
||
*,
|
||
has_photo: bool,
|
||
has_chart: bool,
|
||
) -> dict[str, Any]:
|
||
fallback = self.default_plan(draft, has_photo=has_photo, has_chart=has_chart)
|
||
prompt = self._prompt(draft, has_photo=has_photo, has_chart=has_chart)
|
||
try:
|
||
endpoints = await DatabaseModelEndpointResolver().resolve(
|
||
agent_type="promotion_material",
|
||
task_type="text_generation",
|
||
)
|
||
result = await get_model_service().generate(endpoints, prompt, max_attempts=2)
|
||
parsed = self._parse_json(result.text)
|
||
return self._validate(parsed, draft, fallback)
|
||
except Exception:
|
||
logger.warning("推介材料版式规划失败，退回确定性版式", exc_info=True)
|
||
return fallback
|
||
|
||
@staticmethod
|
||
def default_plan(
|
||
draft: dict[str, Any],
|
||
*,
|
||
has_photo: bool,
|
||
has_chart: bool,
|
||
) -> dict[str, Any]:
|
||
chapters = draft.get("chapters", [])
|
||
page_plans = []
|
||
for chapter in chapters:
|
||
title = str(chapter.get("title") or "")
|
||
density = PromotionLayoutPlanner.content_density(chapter)
|
||
if chapter.get("chart_index") is not None and has_chart:
|
||
layout = "chart_focus"
|
||
elif title == "管理人及投研团队" and has_photo:
|
||
layout = "manager_profile"
|
||
elif title == "风险揭示":
|
||
layout = "full_width_disclosure"
|
||
elif density == "low":
|
||
layout = "visual_focus"
|
||
elif title in {"产品基本信息", "投资范围、策略与限制", "费用结构"}:
|
||
layout = "two_columns"
|
||
else:
|
||
layout = "single_column"
|
||
page_plans.append({
|
||
"chapter_title": title,
|
||
"layout": layout,
|
||
"content_density": density,
|
||
})
|
||
return {
|
||
"cover_layout": "cover_left_photo_right" if has_photo else "cover_visual_right",
|
||
"poster_layout": "poster_chart_focus" if has_chart else "poster_balanced",
|
||
"background_theme": "data_lines",
|
||
"design_profile": PromotionLayoutPlanner._default_design_profile(
|
||
chapters, has_photo=has_photo, has_chart=has_chart
|
||
),
|
||
"page_plans": page_plans,
|
||
}
|
||
|
||
@staticmethod
|
||
def _default_design_profile(
|
||
chapters: list[dict[str, Any]], *, has_photo: bool, has_chart: bool
|
||
) -> dict[str, str]:
|
||
"""从参考样张提炼可组合的设计基因，而不是复制固定模板。"""
|
||
strategy = next(
|
||
(item for item in chapters if item.get("title") == "投资范围、策略与限制"),
|
||
{},
|
||
)
|
||
strategy_is_primary = len(str(strategy.get("body") or "")) >= 90
|
||
if has_chart and has_photo:
|
||
tone = "institutional_tech"
|
||
elif has_photo:
|
||
tone = "warm_editorial"
|
||
else:
|
||
tone = "quiet_data"
|
||
return {
|
||
"design_tone": tone,
|
||
"content_structure": (
|
||
"strategy_right_emphasis" if strategy_is_primary else "balanced_split"
|
||
),
|
||
"column_ratio": "40_60" if strategy_is_primary else "50_50",
|
||
"manager_layout": "profile_feature" if has_photo else "profile_compact",
|
||
"performance_layout": "chart_dominant" if has_chart else "metrics_first",
|
||
"background_role": "edge_texture" if has_chart else "soft_paper",
|
||
}
|
||
|
||
@staticmethod
|
||
def content_density(chapter: dict[str, Any]) -> str:
|
||
"""按原始内容和已有素材判断信息密度，不生成或修改业务内容。"""
|
||
body = str(chapter.get("body") or "").strip()
|
||
length = len(body.replace("\\n", ""))
|
||
signals = sum(
|
||
chapter.get("chart_index") is not None
|
||
if name == "chart_index"
|
||
else bool(chapter.get(name))
|
||
for name in ("chart_index", "metrics", "table", "photo")
|
||
)
|
||
if length < 90 and signals == 0:
|
||
return "low"
|
||
if length < 260 and signals <= 1:
|
||
return "medium"
|
||
return "high"
|
||
|
||
@staticmethod
|
||
def _prompt(
|
||
draft: dict[str, Any],
|
||
*,
|
||
has_photo: bool,
|
||
has_chart: bool,
|
||
) -> str:
|
||
chapters = [
|
||
{
|
||
"title": str(item.get("title") or ""),
|
||
"body_length": len(str(item.get("body") or "")),
|
||
"has_chart": item.get("chart_index") is not None,
|
||
"content_density": PromotionLayoutPlanner.content_density(item),
|
||
}
|
||
for item in draft.get("chapters", [])
|
||
]
|
||
request = {
|
||
"style_code": draft.get("style_code"),
|
||
"title": draft.get("title"),
|
||
"chapters": chapters,
|
||
"has_manager_photo": has_photo,
|
||
"has_performance_chart": has_chart,
|
||
}
|
||
return (
|
||
"你是金融产品推介材料的版式规划器。只根据以下结构化摘要选择版式，"
|
||
"不得改写、删减、补充任何材料文字、数字、表格、头像或业绩数据。"
|
||
"不得返回坐标、颜色值、HTML、SVG、图片提示词或新事实。"
|
||
"必须只返回一个 JSON 对象，字段为 cover_layout、poster_layout、"
|
||
"background_theme、design_profile、page_plans。"
|
||
"design_profile 含 design_tone、content_structure、column_ratio、manager_layout、"
|
||
"performance_layout、background_role；page_plans 的每项只含 chapter_title 和 layout。"
|
||
"可选 cover_layout：cover_left_photo_right、cover_visual_right、cover_full_bleed。"
|
||
"可选 poster_layout：poster_balanced、poster_chart_focus、poster_manager_focus。"
|
||
"可选 background_theme：data_lines、geometric_grid、soft_wave、none。"
|
||
"可选 layout：single_column、two_columns、chart_focus、metrics_focus、"
|
||
"manager_profile、full_width_disclosure、visual_focus。"
|
||
"低密度页面优先选择 visual_focus，"
|
||
"但不得删除或改写原始文字。content_density 只用于版式选择。"
|
||
"可选 design_tone：institutional_tech、warm_editorial、quiet_data。"
|
||
"可选 content_structure：balanced_split、strategy_right_emphasis、"
|
||
"strategy_left_emphasis。可选 column_ratio：40_60、45_55、50_50、55_45、60_40。"
|
||
"manager_layout 只能是 profile_feature 或 profile_compact；"
|
||
"performance_layout 只能是 chart_dominant 或 metrics_first；"
|
||
"background_role 只能是 edge_texture 或 soft_paper。"
|
||
"页面标题必须逐字复制输入中的 title。"
|
||
f"结构化摘要：{json.dumps(request, ensure_ascii=False)}"
|
||
)
|
||
|
||
@staticmethod
|
||
def _parse_json(text: str) -> dict[str, Any]:
|
||
value = text.strip()
|
||
if value.startswith("```"):
|
||
value = value.split("\n", 1)[1].rsplit("```", 1)[0].strip()
|
||
parsed = json.loads(value)
|
||
if not isinstance(parsed, dict):
|
||
raise ValueError("版式规划不是 JSON 对象")
|
||
return parsed
|
||
|
||
@classmethod
|
||
def _validate(
|
||
cls,
|
||
value: dict[str, Any],
|
||
draft: dict[str, Any],
|
||
fallback: dict[str, Any],
|
||
) -> dict[str, Any]:
|
||
if value.get("cover_layout") not in ALLOWED_COVER_LAYOUTS:
|
||
return fallback
|
||
if value.get("poster_layout") not in ALLOWED_POSTER_LAYOUTS:
|
||
return fallback
|
||
if value.get("background_theme") not in ALLOWED_BACKGROUND_THEMES:
|
||
return fallback
|
||
profile = value.get("design_profile")
|
||
if not isinstance(profile, dict):
|
||
return fallback
|
||
if profile.get("design_tone") not in ALLOWED_DESIGN_TONES:
|
||
return fallback
|
||
if profile.get("content_structure") not in ALLOWED_CONTENT_STRUCTURES:
|
||
return fallback
|
||
if profile.get("column_ratio") not in ALLOWED_COLUMN_RATIOS:
|
||
return fallback
|
||
if profile.get("manager_layout") not in {"profile_feature", "profile_compact"}:
|
||
return fallback
|
||
if profile.get("performance_layout") not in {"chart_dominant", "metrics_first"}:
|
||
return fallback
|
||
if profile.get("background_role") not in {"edge_texture", "soft_paper"}:
|
||
return fallback
|
||
expected_titles = [
|
||
str(item.get("title") or "") for item in draft.get("chapters", [])
|
||
]
|
||
raw_pages = value.get("page_plans")
|
||
if not isinstance(raw_pages, list):
|
||
return fallback
|
||
by_title: dict[str, str] = {}
|
||
for item in raw_pages:
|
||
if not isinstance(item, dict):
|
||
return fallback
|
||
title = item.get("chapter_title")
|
||
layout = item.get("layout")
|
||
if title not in expected_titles or layout not in ALLOWED_CONTENT_LAYOUTS:
|
||
return fallback
|
||
by_title[str(title)] = str(layout)
|
||
if set(by_title) != set(expected_titles):
|
||
return fallback
|
||
return {
|
||
"cover_layout": value["cover_layout"],
|
||
"poster_layout": value["poster_layout"],
|
||
"background_theme": value["background_theme"],
|
||
"design_profile": {name: str(profile[name]) for name in (
|
||
"design_tone", "content_structure", "column_ratio", "manager_layout",
|
||
"performance_layout", "background_role",
|
||
)},
|
||
"page_plans": [
|
||
{
|
||
"chapter_title": title,
|
||
"layout": by_title[title],
|
||
"content_density": next(
|
||
(
|
||
str(item.get("content_density"))
|
||
for item in raw_pages
|
||
if isinstance(item, dict) and item.get("chapter_title") == title
|
||
),
|
||
cls.content_density(next(
|
||
item for item in draft.get("chapters", [])
|
||
if str(item.get("title") or "") == title
|
||
)),
|
||
),
|
||
}
|
||
for title in expected_titles
|
||
],
|
||
}
|