chengma/cmbot
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ai-outfit): 左栏新增「标题生成」(看图→提示词→AI 文字标题→写回A列) (§19.18)

Browse Source 
AI 穿搭页左栏新增独立的「生成标题」流程：用户写标题提示词，AI 看该行
衣服图（目录行取首图）生成电商标题，逐行立即写回 Excel A 列并刷新明细表；
完成后重载 Excel，紧接「开始生成」跑图即用新标题。移除原「最终生成要求预览」
腾出版面（docs/11 §17）。

- ai_text_service.py：AiTextClient 复用图像服务 HTTP 管道做文本输出；
  chat/gemini 带图视觉，images/images_edits 明确报错；extract_text 取首条标题
- ai_title.py + TitleResult：单行编排，never raises
- excel_service.write_title_result：只写 A 列、不动 D/E/F
- config_service：title_model 默认 + load/save_title_prompt + 默认标题话术
- 面板：标题生成组（提示词+标题模型下拉+保存+生成标题）置于话术组上方；
  _TitleWorker 顺序逐行+立即回填+刷新；与「开始生成」互斥；删预览相关组件
- 测试：文本解析/payload、generate_title（单文件/目录首图/失败）、
  write_title_result、面板标题组存在且无预览；全套 py37 通过

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
admin
2026-06-23 11:04:38 +08:00
co-authored by Claude Opus 4.8
parent 313e71b774
commit 2fa488bd21
12 changed files with 921 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app/widgets/ai_outfit_panel.py
+262 -84
View File
@@ -43,7 +43,9 @@ from services.config_service import (
DEFAULT_OUTFIT_PROMPT,
load_ai_models,
load_outfit_prompts,
load_title_prompt,
save_outfit_prompts,
save_title_prompt,
)
logger = logging.getLogger(__name__)
@@ -136,6 +138,84 @@ class _OutfitWorker(QObject):
self.finished.emit(summary)
class _TitleWorker(QObject):
"""Generates titles row-by-row on a QThread (docs/11 §17); queued signals.
Sequential by row: each row's garment image + the title prompt -> one title,
written back to that row's A column immediately. Failures are logged and skipped.
"""
tasks_loaded = Signal(object) # List[OutfitTask]
log = Signal(str)
progress = Signal(int, int, object) # completed, total, TitleResult
finished = Signal(int, int) # success_count, fail_count
failed = Signal(str) # fatal pre-run error (e.g. Excel locked)
def __init__(self, excel_path, model_config, prompt, request_interval):
super().__init__()
self._excel_path = excel_path
self._model_config = model_config
self._prompt = prompt
self._interval = float(request_interval or 0.0)
self._stop = False
def stop(self):
self._stop = True
def run(self):
import time
from core.ai_title import generate_title
from core.models import TitleResult
from services.ai_text_service import AiTextClient
from services.excel_service import (
ensure_excel_writable,
read_all_rows,
write_title_result,
)
try:
ensure_excel_writable(self._excel_path)
rows = read_all_rows(self._excel_path)
client = AiTextClient(self._model_config) # one client for the run
except Exception as exc: # noqa: BLE001 - report to UI
self.failed.emit(str(exc))
return
self.tasks_loaded.emit(rows)
total = len(rows)
if total == 0:
self.finished.emit(0, 0)
return
success = fail = 0
for index, task in enumerate(rows, start=1):
if self._stop:
break
if index > 1 and self._interval > 0:
time.sleep(self._interval)
result = generate_title(task, self._prompt, self._model_config,
api_client=client)
if result.success:
try:
write_title_result(self._excel_path, task.row_index,
result.generated_title)
except Exception as exc: # noqa: BLE001 - keep going
result = TitleResult(task=task, success=False,
error="写回失败：{}".format(exc), attempts=1)
if result.success:
success += 1
self.log.emit("第 {} 行标题：{}".format(
task.row_index, result.generated_title))
else:
fail += 1
self.log.emit("第 {} 行标题失败：{}".format(
task.row_index, result.error))
self.progress.emit(index, total, result)
self.finished.emit(success, fail)
# ---------------------------------------------------------------------------
# Main panel
# ---------------------------------------------------------------------------
@@ -151,6 +231,9 @@ class AiOutfitPanel(QWidget):
self._models = [] # list of model dicts from ai_models.json
self._thread = None
self._worker = None
self._title_thread = None # 标题生成线程（§17）
self._title_worker = None
self._title_fail = 0 # running failure count for title stats
self._row_to_table = {} # excel row_index -> table row
self._failures = [] # list of OutfitResult (failed)
self._last_resolution = "" # for "value actually changed" check (§10.2)
@@ -241,7 +324,7 @@ class AiOutfitPanel(QWidget):
col.setContentsMargins(12, 12, 12, 12)
col.setSpacing(10)
# 数据源 / 输出：行内一行（标签 + 路径 + 浏览），省纵向空间给话术/预览
# 数据源 / 输出：行内一行（标签 + 路径 + 浏览），省纵向空间给标题/话术
self._excel_edit = QLineEdit()
self._excel_edit.setPlaceholderText("选择商品表 .xlsx")
col.addLayout(self._inline_path_row("Excel", self._excel_edit, self._browse_excel))
@@ -249,6 +332,9 @@ class AiOutfitPanel(QWidget):
self._output_edit.setPlaceholderText("默认：程序旁的「穿搭图片」")
col.addLayout(self._inline_path_row("输出", self._output_edit, self._browse_output))
# 标题生成（§17）：看衣服图 + 提示词 → AI 文字标题 → 写回 Excel A 列
col.addWidget(self._build_title_group(), stretch=2)
# 穿搭生成话术（多套模板；加大，随窗口高度拉伸）
prm = QGroupBox()
pv = QVBoxLayout(prm)
@@ -271,7 +357,6 @@ class AiOutfitPanel(QWidget):
pv.addLayout(trow)
self._prompt_edit = QPlainTextEdit()
self._prompt_edit.setMinimumHeight(150)
self._prompt_edit.textChanged.connect(self._refresh_preview)
pv.addWidget(self._prompt_edit)
prow = QHBoxLayout()
insert_title_btn = QPushButton("插入标题")
@@ -282,31 +367,37 @@ class AiOutfitPanel(QWidget):
save_btn.clicked.connect(self._save_prompt)
prow.addWidget(save_btn)
pv.addLayout(prow)
col.addWidget(prm, stretch=2)
# 最终生成要求预览（加大；内嵌、实时；替换样本行占位符 + 附加输出要求）
prev = QGroupBox()
pvw = QVBoxLayout(prev)
preview_header = QHBoxLayout()
preview_header.addWidget(QLabel("最终生成要求预览"))
self._sample_combo = QComboBox()
self._compact_combo(self._sample_combo)
self._sample_combo.currentIndexChanged.connect(self._refresh_preview)
preview_header.addWidget(self._sample_combo, stretch=1)
pvw.addLayout(preview_header)
self._preview_warn = QLabel("")
self._preview_warn.setStyleSheet("color:#b87a00;")
self._preview_warn.setVisible(False)
pvw.addWidget(self._preview_warn)
self._preview_view = QPlainTextEdit()
self._preview_view.setReadOnly(True)
self._preview_view.setMinimumHeight(200)
pvw.addWidget(self._preview_view)
col.addWidget(prev, stretch=3)
col.addWidget(prm, stretch=3)
scroll.setWidget(inner)
return scroll
def _build_title_group(self):
"""标题生成组（§17）：提示词 + 标题模型下拉 + 保存 + 生成标题。"""
box = QGroupBox()
v = QVBoxLayout(box)
v.addWidget(QLabel("标题生成提示词"))
self._title_prompt_edit = QPlainTextEdit()
self._title_prompt_edit.setMinimumHeight(110)
v.addWidget(self._title_prompt_edit)
model_row = QHBoxLayout()
model_row.addWidget(QLabel("标题模型"))
self._title_model_combo = QComboBox()
self._compact_combo(self._title_model_combo)
model_row.addWidget(self._title_model_combo, stretch=1)
v.addLayout(model_row)
btn_row = QHBoxLayout()
save_title_btn = QPushButton("保存")
save_title_btn.clicked.connect(self._save_title_prompt)
btn_row.addWidget(save_title_btn)
self._title_btn = QPushButton("生成标题")
self._title_btn.clicked.connect(self._start_title)
btn_row.addWidget(self._title_btn)
v.addLayout(btn_row)
return box
def _create_settings_group(self):
"""生成设置 group (lives in the right run column); two params per row."""
gen = QGroupBox()
@@ -331,7 +422,6 @@ class AiOutfitPanel(QWidget):
self._retry_count.setRange(0, 10)
self._resolution = QComboBox()
self._resolution.addItems(_RESOLUTIONS)
self._resolution.currentIndexChanged.connect(self._refresh_preview)
# activated = user click only; programmatic sets won't pop (docs/11 §10.2)
self._resolution.activated.connect(self._on_resolution_activated)
self._quality = QComboBox()
@@ -490,24 +580,29 @@ class AiOutfitPanel(QWidget):
self._rebuild_prompt_combo()
self._apply_prompt(name)
# 标题生成提示词（单份，§17.3）
self._title_prompt_edit.setPlainText(load_title_prompt())
self._models = load_ai_models()
self._model_combo.clear()
if not self._models:
self._model_combo.addItem("（未配置模型，请在 ai_models.json 添加）")
self._model_combo.setEnabled(False)
else:
self._model_combo.setEnabled(True)
for m in self._models:
self._model_combo.addItem(m.get("name") or m.get("model") or "(未命名)")
self._set_combo(self._model_combo, config.get("outfit_model", ""))
self._fill_model_combo(self._model_combo, config.get("outfit_model", ""))
self._fill_model_combo(self._title_model_combo, config.get("title_model", ""))
# Snapshot current dropdown values so a later user re-select of the same
# item doesn't trigger the info popup (§10.2).
self._last_resolution = self._resolution.currentText()
self._last_model = self._model_combo.currentText() if self._models else ""
# Fill the preview's sample-row dropdown from the remembered Excel.
self._reload_sample_rows()
def _fill_model_combo(self, combo, selected_name):
"""Fill a model dropdown from self._models (shared by 图片/标题 model, §17.3)."""
combo.clear()
if not self._models:
combo.addItem("（未配置模型，请在 ai_models.json 添加）")
combo.setEnabled(False)
return
combo.setEnabled(True)
for m in self._models:
combo.addItem(m.get("name") or m.get("model") or "(未命名)")
self._set_combo(combo, selected_name)
def _set_combo(self, combo, value):
idx = combo.findText(str(value))
@@ -527,6 +622,7 @@ class AiOutfitPanel(QWidget):
"outfit_quality": self._quality.currentText(),
"outfit_retry_failed": self._retry_failed_chk.isChecked(),
"outfit_prompt_name": self._current_prompt_name,
"title_model": self._title_model_combo.currentText() if self._models else "",
})
# -- left actions ---------------------------------------------------
@@ -537,7 +633,6 @@ class AiOutfitPanel(QWidget):
if path:
self._excel_edit.setText(path)
self._emit_config()
self._reload_sample_rows()
def _browse_output(self):
path = QFileDialog.getExistingDirectory(
@@ -574,7 +669,7 @@ class AiOutfitPanel(QWidget):
self._current_prompt_name = name
self._saved_text = self._prompt_text(name)
self._select_prompt_in_combo(name)
self._prompt_edit.setPlainText(self._saved_text) # fires _refresh_preview
self._prompt_edit.setPlainText(self._saved_text)
def _store_current_text(self):
"""Save the editor text into the current template + persist to disk."""
@@ -684,54 +779,133 @@ class AiOutfitPanel(QWidget):
self._apply_prompt(self._prompts[min(idx, len(self._prompts) - 1)]["name"])
self._emit_config()
# -- inline prompt preview ------------------------------------------
# -- 标题生成（§17）-------------------------------------------------
def _fill_sample_combo(self, tasks):
"""Populate the sample-row dropdown (None data = no sample)."""
self._sample_combo.blockSignals(True)
self._sample_combo.clear()
if tasks:
for t in tasks:
pid = t.product_id if (t.product_id and t.product_id.strip()) else "(无货号)"
self._sample_combo.addItem(
"第 {} 行 · {} · {}".format(t.row_index, pid, t.title), t)
else:
self._sample_combo.addItem("（选 Excel 后显示替换效果）", None)
self._sample_combo.blockSignals(False)
self._refresh_preview()
def _save_title_prompt(self, silent=False):
save_title_prompt(self._title_prompt_edit.toPlainText())
if not silent:
self.statusBar_message("标题提示词已保存")
def _reload_sample_rows(self):
"""Best-effort: read the chosen Excel to fill the sample dropdown.
def _selected_title_model_config(self):
if not self._models:
QMessageBox.warning(
self, "未配置模型",
"尚未配置 AI 模型。请在 ~/.cmbot/config/ai_models.json 添加后重试。")
return None
idx = self._title_model_combo.currentIndex()
if idx < 0 or idx >= len(self._models):
QMessageBox.warning(self, "未选择标题模型", "请先选择标题模型。")
return None
from services.ai_image_service import AiModelConfig, api_config_errors
data = self._models[idx]
errors = api_config_errors(data)
if errors:
QMessageBox.warning(self, "模型配置有误", "；".join(errors))
return None
return AiModelConfig.from_dict(data)
Uses read_all_rows (status-independent) so preview still works after the
whole sheet is 完成 (docs/11 §10.3); generation still uses load_outfit_tasks.
"""
excel = self._excel_edit.text().strip()
rows = []
if excel:
try:
from services.excel_service import read_all_rows
rows = read_all_rows(excel)
except Exception as exc: # noqa: BLE001 - silent for preview
logger.info("Sample rows unavailable: %s", exc)
self._fill_sample_combo(rows)
def _refresh_preview(self):
if not hasattr(self, "_preview_view"):
def _start_title(self):
if self._thread is not None or self._title_thread is not None:
return
from core.ai_outfit import build_output_requirements, render_prompt
template = self._prompt_edit.toPlainText()
self._preview_warn.setVisible("{title}" not in template)
if "{title}" not in template:
self._preview_warn.setText("⚠ 话术缺少 {title} 占位符")
# Mirror what actually gets sent: append the output-requirements block
# for the currently selected resolution (docs/11 §7.1).
resolution = self._resolution.currentText() if hasattr(self, "_resolution") else None
task = self._sample_combo.currentData() if hasattr(self, "_sample_combo") else None
if task is None:
self._preview_view.setPlainText(template + build_output_requirements(resolution))
excel = self._excel_edit.text().strip()
if not excel:
QMessageBox.information(self, "提示", "请先选择 Excel 文件。")
return
model_config = self._selected_title_model_config()
if model_config is None:
return
prompt = self._title_prompt_edit.toPlainText().strip()
if not prompt:
QMessageBox.information(self, "提示", "请先填写标题生成提示词。")
return
self._save_title_prompt(silent=True)
self._emit_config()
# reset run state (shared with image generation)
self._table.setRowCount(0)
self._row_to_table = {}
self._title_fail = 0
self._progress.setValue(0)
self._log.clear()
self._title_worker = _TitleWorker(
excel, model_config, prompt, self._interval.value())
self._title_thread = QThread(self)
self._title_worker.moveToThread(self._title_thread)
self._title_thread.started.connect(self._title_worker.run)
self._title_worker.tasks_loaded.connect(self._on_title_tasks_loaded)
self._title_worker.log.connect(self._append_log)
self._title_worker.progress.connect(self._on_title_progress)
self._title_worker.finished.connect(self._on_title_finished)
self._title_worker.failed.connect(self._on_title_failed)
self._title_worker.finished.connect(self._title_thread.quit)
self._title_worker.failed.connect(self._title_thread.quit)
self._title_thread.finished.connect(self._cleanup_title_thread)
self._title_thread.start()
self._set_title_running(True)
def _on_title_tasks_loaded(self, tasks):
self._populate_table(tasks)
self._update_stats(0, 0, len(tasks))
self._append_log("标题生成：已加载 {} 行".format(len(tasks)))
def _on_title_progress(self, completed, total, result):
row = self._row_to_table.get(result.task.row_index)
if row is not None:
if result.success:
self._set_cell(row, 1, result.generated_title) # refresh 标题 列
self._set_cell(row, 4, "完成")
self._set_cell(row, 5, "标题已生成")
else:
self._set_cell(row, 4, "失败")
self._set_cell(row, 5, result.error)
if not result.success:
self._title_fail += 1
self._progress.setValue(completed)
self._update_stats(completed - self._title_fail, self._title_fail, total - completed)
def _on_title_finished(self, success, fail):
self._set_title_running(False)
msg = "标题生成结束：成功 {}，失败 {}".format(success, fail)
self._append_log(msg)
self._reload_after_titles()
if success or fail:
QMessageBox.information(
self, "标题生成",
msg + "。\n标题已写回 Excel，可点「开始生成」生成穿搭图。")
else:
self._preview_view.setPlainText(render_prompt(template, task, resolution))
QMessageBox.information(self, "标题生成", "该表没有可处理的行。")
def _on_title_failed(self, message):
self._set_title_running(False)
self._append_log("标题生成无法开始：" + message)
QMessageBox.warning(self, "无法开始", message)
def _cleanup_title_thread(self):
self._title_thread = None
self._title_worker = None
def _reload_after_titles(self):
"""Re-read Excel so the table + later image generation use new titles."""
excel = self._excel_edit.text().strip()
if not excel:
return
try:
from services.excel_service import read_all_rows
rows = read_all_rows(excel)
except Exception as exc: # noqa: BLE001 - best effort
logger.info("Reload after titles failed: %s", exc)
return
self._populate_table(rows)
self._update_stats(0, 0, len(rows))
def _set_title_running(self, running):
self._title_btn.setEnabled(not running)
self._title_btn.setText("生成中…" if running else "生成标题")
self._start_btn.setEnabled(not running) # mutually exclusive with 开始生成
self._excel_edit.setEnabled(not running)
self._title_model_combo.setEnabled(not running and bool(self._models))
# -- switch info popups (user-only; §10.2) --------------------------
@@ -767,7 +941,7 @@ class AiOutfitPanel(QWidget):
# -- run control ----------------------------------------------------
def _start(self):
if self._thread is not None:
if self._thread is not None or self._title_thread is not None:
return
excel = self._excel_edit.text().strip()
if not excel:
@@ -856,11 +1030,14 @@ class AiOutfitPanel(QWidget):
self._stop_btn.setEnabled(running)
self._excel_edit.setEnabled(not running)
self._model_combo.setEnabled(not running and bool(self._models))
self._title_btn.setEnabled(not running) # mutually exclusive with 生成标题
# -- worker callbacks (UI thread) -----------------------------------
def _on_tasks_loaded(self, tasks):
def _populate_table(self, tasks):
"""Fill the detail table from a task list (shared by image/title runs)."""
self._table.setRowCount(len(tasks))
self._row_to_table = {}
for row, task in enumerate(tasks):
self._row_to_table[task.row_index] = row
self._set_cell(row, 0, str(task.row_index))
@@ -870,9 +1047,10 @@ class AiOutfitPanel(QWidget):
self._set_cell(row, 4, "待处理")
self._set_cell(row, 5, "—")
self._progress.setMaximum(max(1, len(tasks)))
def _on_tasks_loaded(self, tasks):
self._populate_table(tasks)
self._update_stats(0, 0, len(tasks))
# Preview sample rows are owned by read_all_rows (§10.3/§10.4); don't reset
# them to the run's task list (which is empty when the sheet is all 完成).
self._append_log("已加载 {} 行待处理任务".format(len(tasks)))
def _on_progress(self, completed, total, result):
src/core/ai_title.py
+49
View File
@@ -0,0 +1,49 @@
"""AI 标题生成单行编排（docs/11 §17）。
看该行衣服图（目录行取首图）+ 用户标题提示词 → 调 AI 文本服务生成电商标题。
返回 TitleResult；never raises（异常聚合进结果）。标题写回 Excel A 列由调用方做。
"""
import logging
from core.ai_outfit import list_directory_images, looks_like_directory
from core.models import OutfitTask, TitleResult
from services.ai_text_service import AiTextClient
logger = logging.getLogger(__name__)
def render_title_prompt(template, task):
"""Replace {title}/{product_id} in the title prompt (no image-output tail)."""
return str(template).replace("{title}", task.title).replace(
"{product_id}", task.product_id)
def _reference_image(garment_path):
"""Pick the vision reference: the file itself, or a directory's first image."""
if looks_like_directory(garment_path):
images = list_directory_images(garment_path)
return str(images[0]) if images else None
return garment_path
def generate_title(task, prompt_template, model_config, api_client=None):
"""Generate one title for an Excel row and return TitleResult. Never raises."""
if not isinstance(task, OutfitTask):
raise TypeError("task must be OutfitTask")
try:
image_path = _reference_image(task.garment_path)
if not image_path:
return TitleResult(task=task, success=False, attempts=1,
error="目录内没有图片：{}".format(task.garment_path))
prompt = render_title_prompt(prompt_template, task)
client = api_client or AiTextClient(model_config)
title = client.generate_text(prompt, image_path)
if not title:
return TitleResult(task=task, success=False, attempts=1,
error="AI 未返回标题")
logger.info("Generated title row %s -> %s", task.row_index, title)
return TitleResult(task=task, success=True, generated_title=title, attempts=1)
except Exception as exc: # noqa: BLE001 - aggregate
logger.exception("Title generation failed for row %s", task.row_index)
return TitleResult(task=task, success=False, error=str(exc), attempts=1)
src/core/models.py
+13
View File
@@ -182,6 +182,19 @@ class OutfitResult:
output_paths: List[str] = field(default_factory=list)
@dataclass
class TitleResult:
"""AI 标题生成单行结果（docs/11 §17）。
generated_title: 成功时为 AI 生成、清洗后的单行标题（写回 Excel A 列）。
"""
task: OutfitTask
success: bool
generated_title: str = ""
error: str = ""
attempts: int = 0
# ---------------------------------------------------------------------------
# 合成结果
# ---------------------------------------------------------------------------
src/services/ai_text_service.py
+185
View File
@@ -0,0 +1,185 @@
"""AI 文本服务（docs/11 §17）：复用图像服务的 HTTP 管道，输出文字（标题）。
与 `ai_image_service.ImageApiClient` 平行：同一份 `AiModelConfig`、传图、鉴权、
超时、session 全部复用，只把「解析图片」换成「解析文字」。仅 chat / gemini 这类
能返回文字的接口可用；纯图片接口（images / images_edits）会明确报错。
"""
import logging
import re
import requests
from services.ai_image_service import (
API_CHAT,
API_GEMINI,
API_IMAGES,
API_IMAGES_EDITS,
_coerce_config,
_split_data_url,
detect_api_type,
image_to_data_url,
normalize_api_url,
resolution_timeout,
validate_api_config,
)
logger = logging.getLogger(__name__)
class AiTextServiceError(RuntimeError):
"""Raised when AI text (title) generation fails."""
# Strip a leading list marker (1. / 2) / - / • / ①…) from a title line.
_TITLE_LEAD = re.compile(r"^\s*(?:\d+\s*[\.\)、:：]|[-*•])\s*")
_TITLE_CIRCLED = "①②③④⑤⑥⑦⑧⑨⑩"
_TITLE_QUOTES = "\"'「」『』“”‘’"
def build_text_payload(config, prompt, image_data_url=None):
"""Build a JSON body that asks a chat/gemini model for TEXT output.
image_data_url optional: when given, the garment image is sent as a vision
reference (docs/11 §17 看图生成). images/images_edits are text-incapable.
"""
cfg = _coerce_config(config)
api_type = detect_api_type(cfg.url, cfg.api_type)
if api_type == API_CHAT:
content = [{"type": "text", "text": prompt}]
if image_data_url:
content.append({"type": "image_url", "image_url": {"url": image_data_url}})
payload = {
"model": cfg.model,
"messages": [{"role": "user", "content": content}],
"stream": False,
}
elif api_type == API_GEMINI:
parts = [{"text": prompt}]
if image_data_url:
mime_type, data = _split_data_url(image_data_url)
parts.append({"inlineData": {"mimeType": mime_type, "data": data}})
payload = {
"contents": [{"parts": parts}],
"generationConfig": {"responseModalities": ["TEXT"]},
}
else:
raise AiTextServiceError(
"该模型是图片接口（{}），不能生成文字标题，请改选能返回文字的模型"
"（chat 或 gemini）".format(api_type))
payload.update(cfg.extra_body)
return payload
def _content_to_text(content):
"""Flatten an OpenAI chat message 'content' (str or parts list) to text."""
if isinstance(content, str):
return content
if isinstance(content, list):
texts = []
for part in content:
if isinstance(part, dict) and isinstance(part.get("text"), str):
texts.append(part["text"])
elif isinstance(part, str):
texts.append(part)
return "\n".join(texts)
return ""
def _extract_raw_text(data):
"""Pull the model's text out of a chat or gemini JSON response."""
if not isinstance(data, dict):
return ""
choices = data.get("choices")
if isinstance(choices, list) and choices and isinstance(choices[0], dict):
message = choices[0].get("message")
if isinstance(message, dict):
text = _content_to_text(message.get("content"))
if text.strip():
return text
# Some relays use the legacy completion shape choices[0].text.
legacy = choices[0].get("text")
if isinstance(legacy, str) and legacy.strip():
return legacy
candidates = data.get("candidates")
if isinstance(candidates, list) and candidates and isinstance(candidates[0], dict):
content = candidates[0].get("content")
if isinstance(content, dict) and isinstance(content.get("parts"), list):
texts = [p.get("text") for p in content["parts"]
if isinstance(p, dict) and isinstance(p.get("text"), str)]
joined = "\n".join(t for t in texts if t)
if joined.strip():
return joined
return ""
def _clean_title(text):
"""Return the first non-empty line as a single clean title.
Drops list numbering/bullets and wrapping quotes; even if the prompt asked
for several titles, only the first is used (docs/11 §17.1).
"""
if not text:
return ""
for line in str(text).splitlines():
stripped = line.strip()
if not stripped:
continue
stripped = _TITLE_LEAD.sub("", stripped)
stripped = stripped.lstrip(_TITLE_CIRCLED).strip()
stripped = stripped.strip(_TITLE_QUOTES).strip()
if stripped:
return stripped
return ""
def extract_text_from_response(data):
"""Return the first clean title text from an AI JSON response ('' if none)."""
return _clean_title(_extract_raw_text(data))
class AiTextClient:
"""HTTP client for AI text (title) generation; mirrors ImageApiClient."""
def __init__(self, config, session=None):
self.config = _coerce_config(config)
self.session = session or requests.Session()
if hasattr(self.session, "trust_env"):
self.session.trust_env = False
def generate_text(self, prompt, image_path=None, resolution="1K"):
validate_api_config(self.config)
api_type = detect_api_type(self.config.url, self.config.api_type)
if api_type in (API_IMAGES, API_IMAGES_EDITS):
raise AiTextServiceError(
"该模型是图片接口（{}），不能生成文字标题，请改选能返回文字的模型"
"（chat 或 gemini）".format(api_type))
url = normalize_api_url(self.config.url, api_type)
if api_type == API_GEMINI:
url = url.replace("{model}", self.config.model)
headers = {
"Authorization": "Bearer {}".format(self.config.api_key),
"Content-Type": "application/json",
}
read_timeout = (
self.config.timeout_seconds
if self.config.timeout_seconds > 0
else resolution_timeout(resolution)
)
timeout = (self.config.connect_timeout_seconds, read_timeout)
data_url = image_to_data_url(image_path) if image_path else None
payload = build_text_payload(self.config, prompt, data_url)
response = self.session.post(url, headers=headers, json=payload, timeout=timeout)
response.raise_for_status()
title = extract_text_from_response(response.json())
if not title:
raise AiTextServiceError("AI 响应中未找到文字标题")
return title
src/services/config_service.py
+36
View File
@@ -27,12 +27,14 @@ DEFAULT_CONFIG = {
"outfit_quality": "均衡",
"outfit_retry_failed": False,
"outfit_prompt_name": "默认", # last-selected 话术模板 name (docs/11 §7.2)
"title_model": "", # last-selected 标题模型 name (docs/11 §17.3)
}
_CONFIG_FILENAME = "app_config.json"
_AI_MODELS_FILENAME = "ai_models.json"
_OUTFIT_PROMPT_FILENAME = "outfit_prompt.txt" # legacy single prompt (migrated)
_OUTFIT_PROMPTS_FILENAME = "outfit_prompts.json" # multi named templates (§7.2)
_TITLE_PROMPT_FILENAME = "title_prompt.txt" # single 标题生成提示词 (§17.3)
DEFAULT_OUTFIT_PROMPT_NAME = "默认"
# Default outfit prompt (docs/11 §7). Seeded into outfit_prompts.json on first run.
@@ -42,6 +44,12 @@ DEFAULT_OUTFIT_PROMPT = (
"电商主图风格，不加文字与促销标签。"
)
# Default title prompt (docs/11 §17.3). Used by 标题生成 when title_prompt.txt absent.
DEFAULT_TITLE_PROMPT = (
"请根据这件女装的款式、版型、颜色与印花特点，生成一条适合台湾蝦皮电商的中文商品标题："
"突出卖点与适穿场景，控制在 30 字以内。只输出标题本身一行，不要序号、引号、表情或促销词。"
)
def load_config():
"""
@@ -173,6 +181,34 @@ def save_outfit_prompt(text):
logger.error("Failed to save outfit prompt to %s: %s", prompt_file, exc)
def load_title_prompt():
"""Return the saved 标题生成提示词, or the built-in default (docs/11 §17.3)."""
from services.file_service import get_config_path
prompt_file = get_config_path(_TITLE_PROMPT_FILENAME)
if not prompt_file.exists():
return DEFAULT_TITLE_PROMPT
try:
with open(str(prompt_file), encoding="utf-8-sig") as f:
text = f.read()
return text if text.strip() else DEFAULT_TITLE_PROMPT
except OSError as exc:
logger.warning("Title prompt unreadable (%s): %s", exc, prompt_file)
return DEFAULT_TITLE_PROMPT
def save_title_prompt(text):
"""Persist the 标题生成提示词 (utf-8, no BOM). Does not raise."""
from services.file_service import get_config_path
prompt_file = get_config_path(_TITLE_PROMPT_FILENAME)
try:
prompt_file.parent.mkdir(parents=True, exist_ok=True)
with open(str(prompt_file), "w", encoding="utf-8") as f:
f.write(text)
logger.info("Title prompt saved to %s", prompt_file)
except OSError as exc:
logger.error("Failed to save title prompt to %s: %s", prompt_file, exc)
def _normalize_prompts(data):
"""Keep only valid {name, text} entries (non-empty name, string text)."""
if not isinstance(data, list):
src/services/excel_service.py
+16
View File
@@ -173,6 +173,22 @@ def write_outfit_source_excel(excel_path, rows):
logger.info("Wrote outfit source Excel: %s (%d rows)", excel_path, len(rows))
def write_title_result(excel_path, row_index, title):
"""Write a generated title into column A (标题) and save (docs/11 §17).
Only touches the title cell; D/E/F (image-generation status) are untouched.
"""
path = Path(excel_path)
workbook = load_workbook(str(path))
try:
sheet = workbook.worksheets[0]
sheet.cell(row_index, COL_TITLE).value = title
workbook.save(str(path))
logger.info("Outfit row %s title written: %s", row_index, title)
finally:
workbook.close()
def write_outfit_result(excel_path, result):
"""Write one outfit result to columns D/E/F and save immediately."""
if not isinstance(result, OutfitResult):