修复问答助手生成回答报错的bug

This commit is contained in:
db_huang 2026-09-22 17:57:35 +08:00
parent ee0f7d0a68
commit 0b83663f77
7 changed files with 537 additions and 19 deletions

View File

@ -194,6 +194,11 @@ def _generate_bulk_answers_batch(payload: QABulkAnswerRequest, questions: list[s
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="批量回答生成不完整,请重试") from exc raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="批量回答生成不完整,请重试") from exc
def generate_answers_batch(payload: QABulkAnswerRequest, questions: list[str]) -> list[list[str]]:
"""Generate one bounded batch for the persistent background job worker."""
return _generate_bulk_answers_batch(payload, questions)
def generate_answers_for_questions(payload: QABulkAnswerRequest, questions: list[str]) -> list[dict]: def generate_answers_for_questions(payload: QABulkAnswerRequest, questions: list[str]) -> list[dict]:
"""Generate answers for exact imported questions without rewriting them.""" """Generate answers for exact imported questions without rewriting them."""
items: list[dict] = [] items: list[dict] = []

View File

@ -4,6 +4,7 @@ import base64
import csv import csv
import hashlib import hashlib
import json import json
import logging
import random import random
import re import re
import secrets import secrets
@ -19,13 +20,13 @@ from pathlib import Path
from fastapi import BackgroundTasks, Body, Depends, FastAPI, File, Form, HTTPException, Request, Response, UploadFile, status from fastapi import BackgroundTasks, Body, Depends, FastAPI, File, Form, HTTPException, Request, Response, UploadFile, status
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import func, inspect, literal, or_, select, text, union_all from sqlalchemy import func, inspect, literal, or_, select, text, union_all, update
from sqlalchemy.orm import Session, defer, noload, selectinload from sqlalchemy.orm import Session, defer, noload, selectinload
from .config import settings from .config import settings
from .database import Base, SessionLocal, engine, get_db from .database import Base, SessionLocal, engine, get_db
from .llm import extract_questions_from_images, generate_answer_for_question, generate_answers_for_questions, generate_questions, generate_qa from .llm import extract_questions_from_images, generate_answer_for_question, generate_answers_batch, generate_answers_for_questions, generate_questions, generate_qa
from .models import AdminLoginLog, AdminUser, AnswerRecognizedQuestion, AnswerReviewDecision, AnswerScreenshot, AnswerSubmission, AnswerUser, AskAssignment, AskAssignmentBatch, AskAssignmentBatchReview, AskAssignmentBatchScreenshot, AskAssignmentQuestion, AskAssignmentReview, AskAssignmentScreenshot, AskAssignmentUsedQuestion, AskQuestion, AskTask, QALibraryEntry, RefundUploadSession from .models import AdminLoginLog, AdminUser, AnswerRecognizedQuestion, AnswerReviewDecision, AnswerScreenshot, AnswerSubmission, AnswerUser, AskAssignment, AskAssignmentBatch, AskAssignmentBatchReview, AskAssignmentBatchScreenshot, AskAssignmentQuestion, AskAssignmentReview, AskAssignmentScreenshot, AskAssignmentUsedQuestion, AskQuestion, AskTask, QABulkGenerationItem, QABulkGenerationJob, QALibraryEntry, RefundUploadSession
from .schemas import AdminUserCreate, AdminUserUpdate, LoginRequest, QABulkAnswerRequest, QAGenerateRequest, RegisterRequest from .schemas import AdminUserCreate, AdminUserUpdate, LoginRequest, QABulkAnswerRequest, QAGenerateRequest, RegisterRequest
from .security import create_session_token, decode_session_token, hash_password, verify_password from .security import create_session_token, decode_session_token, hash_password, verify_password
@ -47,6 +48,7 @@ ASK_UPLOAD_TIMEOUT = timedelta(minutes=30)
# Bumping this prefix invalidates every previously issued public-user token. # Bumping this prefix invalidates every previously issued public-user token.
# Only Taobao-ID login can issue a token accepted by the current user portals. # Only Taobao-ID login can issue a token accepted by the current user portals.
PUBLIC_USER_SESSION_PREFIX = "tb2_" PUBLIC_USER_SESSION_PREFIX = "tb2_"
QA_BULK_MAX_ATTEMPTS = 3
def _new_public_user_token() -> str: def _new_public_user_token() -> str:
@ -3573,6 +3575,298 @@ def qa_generate_answers(payload: QABulkAnswerRequest, _: AdminUser = Depends(cur
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="批量回答生成服务异常,请稍后重试") from exc raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="批量回答生成服务异常,请稍后重试") from exc
def _qa_bulk_job_data(job: QABulkGenerationJob, include_items: bool = False) -> dict:
processed = job.completed_count + job.failed_count
data = {
"id": job.public_id,
"status": job.status,
"total": job.total_count,
"completed": job.completed_count,
"failed": job.failed_count,
"pending": max(0, job.total_count - processed),
"progress_percent": round(processed * 100 / job.total_count) if job.total_count else 100,
"answers_per_question": job.answers_per_question,
"created_at": job.created_at.isoformat(),
"updated_at": job.updated_at.isoformat(),
}
if include_items:
succeeded = []
failed = []
for item in sorted(job.items, key=lambda value: value.sequence_no):
if item.status == "succeeded":
try:
answers = json.loads(item.answers_json or "[]")
except (TypeError, ValueError):
answers = []
succeeded.append({
"id": f"Q{item.sequence_no:03d}",
"question": item.question_text,
"answers": answers,
"source": "批量导入(AI回答)",
"status": "待采纳",
"accepted": False,
})
elif item.status == "failed":
failed.append({
"sequence_no": item.sequence_no,
"question": item.question_text,
"attempts": item.attempts,
"error": item.error_message or "生成失败",
})
data["items"] = succeeded
data["failed_items"] = failed
return data
def _qa_bulk_job_for_user(public_id: str, user: AdminUser, db: Session) -> QABulkGenerationJob:
job = db.scalar(select(QABulkGenerationJob).where(QABulkGenerationJob.public_id == public_id))
if not job or (user.role_code != "super_admin" and job.owner_id != user.id):
raise HTTPException(status_code=404, detail="批量生成任务不存在")
return job
def _refresh_qa_bulk_job_counts(db: Session, job: QABulkGenerationJob, final: bool = False) -> None:
# SessionLocal disables autoflush. Flush item status updates first so the
# aggregate query always reports the progress that was just persisted.
db.flush()
job.completed_count = int(db.scalar(
select(func.count(QABulkGenerationItem.id)).where(
QABulkGenerationItem.job_id == job.id,
QABulkGenerationItem.status == "succeeded",
)
) or 0)
job.failed_count = int(db.scalar(
select(func.count(QABulkGenerationItem.id)).where(
QABulkGenerationItem.job_id == job.id,
QABulkGenerationItem.status == "failed",
)
) or 0)
if final:
if job.failed_count:
job.status = "partial_failed" if job.completed_count else "failed"
else:
job.status = "completed"
job.updated_at = datetime.utcnow()
def _qa_bulk_error_message(exc: Exception) -> str:
if isinstance(exc, HTTPException):
return str(exc.detail)[:500]
return "大模型生成异常,请重试失败问题"
def _process_qa_bulk_job(job_id: int) -> None:
logger = logging.getLogger(__name__)
with SessionLocal() as db:
claimed = db.execute(
update(QABulkGenerationJob)
.where(QABulkGenerationJob.id == job_id, QABulkGenerationJob.status == "queued")
.values(status="processing", updated_at=datetime.utcnow())
)
if claimed.rowcount != 1:
db.rollback()
return
# A process can stop after marking a group as processing. A later
# retry should safely resume those records instead of leaving them
# permanently stuck.
db.execute(
update(QABulkGenerationItem)
.where(
QABulkGenerationItem.job_id == job_id,
QABulkGenerationItem.status == "processing",
)
.values(status="pending")
)
db.commit()
try:
job = db.get(QABulkGenerationJob, job_id)
if not job:
return
payload = QABulkAnswerRequest(**json.loads(job.payload_json))
pending_ids = list(db.scalars(
select(QABulkGenerationItem.id)
.where(
QABulkGenerationItem.job_id == job_id,
QABulkGenerationItem.status == "pending",
)
.order_by(QABulkGenerationItem.sequence_no)
).all())
batch_size = max(1, min(10, 30 // payload.answers_per_question))
for offset in range(0, len(pending_ids), batch_size):
item_ids = pending_ids[offset:offset + batch_size]
batch_items = list(db.scalars(
select(QABulkGenerationItem)
.where(QABulkGenerationItem.id.in_(item_ids))
.order_by(QABulkGenerationItem.sequence_no)
).all())
questions = [item.question_text for item in batch_items]
last_error = "批量回答生成不完整,请重试"
for attempt in range(QA_BULK_MAX_ATTEMPTS):
db.execute(
update(QABulkGenerationItem)
.where(QABulkGenerationItem.id.in_(item_ids))
.values(
status="processing",
attempts=QABulkGenerationItem.attempts + 1,
error_message=None,
updated_at=datetime.utcnow(),
)
)
job.updated_at = datetime.utcnow()
db.commit()
try:
answer_groups = generate_answers_batch(payload, questions)
if len(answer_groups) != len(batch_items):
raise ValueError("answer group count mismatch")
if any(
not isinstance(answers, list)
or len([answer for answer in answers if str(answer).strip()]) < payload.answers_per_question
for answers in answer_groups
):
raise ValueError("answer count mismatch")
except Exception as exc:
last_error = _qa_bulk_error_message(exc)
logger.warning(
"QA bulk job %s batch %s failed on attempt %s/%s: %s",
job.public_id,
offset // batch_size + 1,
attempt + 1,
QA_BULK_MAX_ATTEMPTS,
last_error,
)
if attempt + 1 < QA_BULK_MAX_ATTEMPTS:
time.sleep(2 ** attempt)
continue
db.execute(
update(QABulkGenerationItem)
.where(QABulkGenerationItem.id.in_(item_ids))
.values(
status="failed",
error_message=last_error,
updated_at=datetime.utcnow(),
)
)
break
else:
for item, answers in zip(batch_items, answer_groups):
stored = db.get(QABulkGenerationItem, item.id)
stored.answers_json = json.dumps(answers, ensure_ascii=False)
stored.status = "succeeded"
stored.error_message = None
stored.updated_at = datetime.utcnow()
break
_refresh_qa_bulk_job_counts(db, job)
db.commit()
_refresh_qa_bulk_job_counts(db, job, final=True)
db.commit()
except Exception:
db.rollback()
logger.exception("QA bulk background job %s crashed", job_id)
job = db.get(QABulkGenerationJob, job_id)
if not job:
return
db.execute(
update(QABulkGenerationItem)
.where(
QABulkGenerationItem.job_id == job_id,
QABulkGenerationItem.status.in_(["pending", "processing"]),
)
.values(
status="failed",
error_message="后台生成任务异常,请重试失败问题",
updated_at=datetime.utcnow(),
)
)
_refresh_qa_bulk_job_counts(db, job, final=True)
db.commit()
@app.post("/api/admin/qa/generate-answer-jobs", status_code=status.HTTP_202_ACCEPTED)
def create_qa_bulk_generation_job(
payload: QABulkAnswerRequest,
background_tasks: BackgroundTasks,
user: AdminUser = Depends(current_user),
db: Session = Depends(get_db),
):
intent_error = _product_intent_error([payload.product_name, payload.category, payload.benefits, payload.selling_points])
if intent_error:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=intent_error)
questions = _dedupe_import_questions(payload.questions)
if not questions:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请至少导入一个有效问题")
normalized_payload = payload.model_dump()
normalized_payload["questions"] = questions
job = QABulkGenerationJob(
public_id=uuid.uuid4().hex,
owner_id=user.id,
payload_json=json.dumps(normalized_payload, ensure_ascii=False),
answers_per_question=payload.answers_per_question,
status="queued",
total_count=len(questions),
)
db.add(job)
db.flush()
db.add_all([
QABulkGenerationItem(job_id=job.id, sequence_no=index, question_text=question)
for index, question in enumerate(questions, start=1)
])
db.commit()
db.refresh(job)
background_tasks.add_task(_process_qa_bulk_job, job.id)
return {"success": True, "data": _qa_bulk_job_data(job)}
@app.get("/api/admin/qa/generate-answer-jobs/{job_public_id}")
def get_qa_bulk_generation_job(
job_public_id: str,
include_items: bool = False,
user: AdminUser = Depends(current_user),
db: Session = Depends(get_db),
):
job = _qa_bulk_job_for_user(job_public_id, user, db)
return {"success": True, "data": _qa_bulk_job_data(job, include_items=include_items)}
@app.post("/api/admin/qa/generate-answer-jobs/{job_public_id}/retry", status_code=status.HTTP_202_ACCEPTED)
def retry_qa_bulk_generation_job(
job_public_id: str,
background_tasks: BackgroundTasks,
user: AdminUser = Depends(current_user),
db: Session = Depends(get_db),
):
job = db.scalar(
select(QABulkGenerationJob)
.where(QABulkGenerationJob.public_id == job_public_id)
.with_for_update()
)
if not job or (user.role_code != "super_admin" and job.owner_id != user.id):
raise HTTPException(status_code=404, detail="批量生成任务不存在")
if job.status in {"queued", "processing"}:
raise HTTPException(status_code=409, detail="批量生成任务仍在处理中")
if job.failed_count < 1:
raise HTTPException(status_code=409, detail="当前没有需要重试的失败问题")
db.execute(
update(QABulkGenerationItem)
.where(
QABulkGenerationItem.job_id == job.id,
QABulkGenerationItem.status == "failed",
)
.values(status="pending", error_message=None, updated_at=datetime.utcnow())
)
job.status = "queued"
job.failed_count = 0
job.updated_at = datetime.utcnow()
db.commit()
db.refresh(job)
background_tasks.add_task(_process_qa_bulk_job, job.id)
return {"success": True, "data": _qa_bulk_job_data(job)}
def _qa_library_data(entry: QALibraryEntry) -> dict: def _qa_library_data(entry: QALibraryEntry) -> dict:
try: try:
answers = json.loads(entry.answers_json) answers = json.loads(entry.answers_json)

View File

@ -67,6 +67,54 @@ class QALibraryEntry(Base):
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
class QABulkGenerationJob(Base):
"""Persistent progress for a bulk AI-answer generation request."""
__tablename__ = "qa_bulk_generation_jobs"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
public_id: Mapped[str] = mapped_column(String(64), unique=True, index=True)
# Keep this as an indexed ownership value without a database FK so the
# table remains compatible with existing MySQL installations whose admin
# primary key may use a different integer width.
owner_id: Mapped[int] = mapped_column(Integer, index=True)
payload_json: Mapped[str] = mapped_column(Text)
answers_per_question: Mapped[int] = mapped_column(Integer)
status: Mapped[str] = mapped_column(String(32), default="queued", index=True)
total_count: Mapped[int] = mapped_column(Integer, default=0)
completed_count: Mapped[int] = mapped_column(Integer, default=0)
failed_count: Mapped[int] = mapped_column(Integer, default=0)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
items: Mapped[list["QABulkGenerationItem"]] = relationship(
back_populates="job",
cascade="all, delete-orphan",
lazy="selectin",
order_by="QABulkGenerationItem.sequence_no",
)
class QABulkGenerationItem(Base):
__tablename__ = "qa_bulk_generation_items"
__table_args__ = (
UniqueConstraint("job_id", "sequence_no", name="uq_qa_bulk_generation_item_sequence"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
job_id: Mapped[int] = mapped_column(ForeignKey("qa_bulk_generation_jobs.id"), index=True)
sequence_no: Mapped[int] = mapped_column(Integer)
question_text: Mapped[str] = mapped_column(Text)
answers_json: Mapped[str | None] = mapped_column(Text, nullable=True)
status: Mapped[str] = mapped_column(String(24), default="pending", index=True)
attempts: Mapped[int] = mapped_column(Integer, default=0)
error_message: Mapped[str | None] = mapped_column(String(500), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
job: Mapped[QABulkGenerationJob] = relationship(back_populates="items")
class AnswerUser(Base): class AnswerUser(Base):
__tablename__ = "answer_users" __tablename__ = "answer_users"

View File

@ -37,3 +37,38 @@ CREATE TABLE IF NOT EXISTS admin_login_logs (
-- 提问任务归属运营账号;历史数据允许为空,超级管理员仍可查看全部任务。 -- 提问任务归属运营账号;历史数据允许为空,超级管理员仍可查看全部任务。
-- 新环境建表时请将 owner_id、shop_name 一并加入 ask_tasks;已有环境由应用启动时自动补列。 -- 新环境建表时请将 owner_id、shop_name 一并加入 ask_tasks;已有环境由应用启动时自动补列。
-- public_token 是对外分享链接标识,不应使用 task_no 拼接用户链接。 -- public_token 是对外分享链接标识,不应使用 task_no 拼接用户链接。
CREATE TABLE IF NOT EXISTS qa_bulk_generation_jobs (
id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
public_id VARCHAR(64) NOT NULL,
owner_id BIGINT NULL,
payload_json LONGTEXT NOT NULL,
answers_per_question INT NOT NULL,
status VARCHAR(24) NOT NULL DEFAULT 'queued',
total_count INT NOT NULL DEFAULT 0,
completed_count INT NOT NULL DEFAULT 0,
failed_count INT NOT NULL DEFAULT 0,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id),
UNIQUE KEY uk_qa_bulk_generation_jobs_public_id (public_id),
KEY idx_qa_bulk_generation_jobs_owner (owner_id),
KEY idx_qa_bulk_generation_jobs_status (status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS qa_bulk_generation_items (
id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
job_id BIGINT UNSIGNED NOT NULL,
sequence_no INT NOT NULL,
question_text TEXT NOT NULL,
answers_json LONGTEXT NULL,
status VARCHAR(24) NOT NULL DEFAULT 'pending',
attempts INT NOT NULL DEFAULT 0,
error_message VARCHAR(500) NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id),
UNIQUE KEY uk_qa_bulk_generation_items_sequence (job_id, sequence_no),
KEY idx_qa_bulk_generation_items_status (job_id, status),
CONSTRAINT fk_qa_bulk_generation_items_job FOREIGN KEY (job_id) REFERENCES qa_bulk_generation_jobs(id) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

View File

@ -131,6 +131,55 @@ def test_imported_questions_can_generate_configured_answer_count(monkeypatch):
assert all(len(item["answers"]) == 3 for item in items) assert all(len(item["answers"]) == 3 for item in items)
def test_bulk_answer_generation_persists_progress_and_retries_only_failed_items(monkeypatch):
state = {"allow_first_batch": False, "calls": []}
def fake_batch_answers(payload, questions):
state["calls"].append(tuple(questions))
if questions[0] == "问题1?" and not state["allow_first_batch"]:
raise RuntimeError("temporary model failure")
return [[f"{question}回答{i}" for i in range(1, payload.answers_per_question + 1)] for question in questions]
monkeypatch.setattr(main_module, "generate_answers_batch", fake_batch_answers)
monkeypatch.setattr(main_module.time, "sleep", lambda _: None)
payload = {
"product_name": "持久化批量测试商品",
"questions": [f"问题{i}?" for i in range(1, 8)],
"answers_per_question": 5,
}
with TestClient(app) as client:
assert client.post(
"/api/admin/auth/login",
json={"username": "operator", "password": "operator123456"},
).status_code == 200
created = client.post("/api/admin/qa/generate-answer-jobs", json=payload)
assert created.status_code == 202
job_id = created.json()["data"]["id"]
failed = client.get(f"/api/admin/qa/generate-answer-jobs/{job_id}?include_items=true")
assert failed.status_code == 200
failed_data = failed.json()["data"]
assert failed_data["status"] == "partial_failed"
assert failed_data["completed"] == 1
assert failed_data["failed"] == 6
assert len(failed_data["items"]) == 1
assert len(failed_data["failed_items"]) == 6
assert state["calls"].count(("问题7?",)) == 1
state["allow_first_batch"] = True
retried = client.post(f"/api/admin/qa/generate-answer-jobs/{job_id}/retry")
assert retried.status_code == 202
completed = client.get(f"/api/admin/qa/generate-answer-jobs/{job_id}?include_items=true")
assert completed.status_code == 200
completed_data = completed.json()["data"]
assert completed_data["status"] == "completed"
assert completed_data["completed"] == 7
assert completed_data["failed"] == 0
assert len(completed_data["items"]) == 7
assert state["calls"].count(("问题7?",)) == 1
def test_accepted_qa_is_saved_to_library_without_duplicates(): def test_accepted_qa_is_saved_to_library_without_duplicates():
unique_product = f"问答库测试商品-{uuid4().hex[:8]}" unique_product = f"问答库测试商品-{uuid4().hex[:8]}"
payload = { payload = {

View File

@ -570,6 +570,55 @@
color: #8296ae; color: #8296ae;
font-size: 11px; font-size: 11px;
} }
.assistant-import-progress {
display: grid;
gap: 9px;
margin-top: 14px;
padding: 13px 14px;
border: 1px solid #bcd3f4;
border-radius: 8px;
background: #f5f9ff;
}
.assistant-import-progress.has-failures {
border-color: #f0c2c8;
background: #fff7f8;
}
.assistant-import-progress header,
.assistant-import-progress p {
display: flex;
align-items: center;
justify-content: space-between;
gap: 10px;
margin: 0;
}
.assistant-import-progress header {
color: #21466f;
font-size: 12px;
}
.assistant-import-progress header b {
color: #1764e8;
}
.assistant-import-progress-track {
height: 7px;
overflow: hidden;
border-radius: 4px;
background: #dce8f7;
}
.assistant-import-progress-track i {
display: block;
height: 100%;
border-radius: inherit;
background: #1764e8;
transition: width .25s ease;
}
.assistant-import-progress p {
flex-wrap: wrap;
color: #7187a3;
font-size: 11px;
}
.assistant-import-progress p b {
color: #173b66;
}
.qa-empty-answer { .qa-empty-answer {
color: #c07a00; color: #c07a00;
font-style: italic; font-style: italic;

View File

@ -1066,11 +1066,15 @@ function AssistantQuestionImportModal({ form, onClose, onImport }) {
); );
const [busy, setBusy] = useState(false); const [busy, setBusy] = useState(false);
const [error, setError] = useState(""); const [error, setError] = useState("");
const [jobProgress, setJobProgress] = useState(null);
const [retryJobId, setRetryJobId] = useState("");
async function selectFile(nextFile) { async function selectFile(nextFile) {
setFile(nextFile || null); setFile(nextFile || null);
setQuestions([]); setQuestions([]);
setError(""); setError("");
setJobProgress(null);
setRetryJobId("");
if (!nextFile) return; if (!nextFile) return;
setBusy(true); setBusy(true);
try { try {
@ -1096,9 +1100,16 @@ function AssistantQuestionImportModal({ form, onClose, onImport }) {
setBusy(true); setBusy(true);
setError(""); setError("");
try { try {
await onImport(questions, generateAnswers, answerCount); await onImport(questions, generateAnswers, answerCount, {
jobId: retryJobId,
onProgress: setJobProgress,
});
} catch (err) { } catch (err) {
setError(err.message || "批量导入失败"); setError(err.message || "批量导入失败");
if (err.bulkJob) {
setJobProgress(err.bulkJob);
setRetryJobId(err.bulkJob.id || "");
}
setBusy(false); setBusy(false);
} }
} }
@ -1126,18 +1137,25 @@ function AssistantQuestionImportModal({ form, onClose, onImport }) {
<fieldset className="assistant-import-options"> <fieldset className="assistant-import-options">
<legend>是否自动生成回答</legend> <legend>是否自动生成回答</legend>
<div className="assistant-import-mode" role="group" aria-label="是否自动生成回答"> <div className="assistant-import-mode" role="group" aria-label="是否自动生成回答">
<button type="button" className={!generateAnswers ? "active" : ""} onClick={() => setGenerateAnswers(false)}>不生成回答</button> <button type="button" disabled={busy || !!retryJobId} className={!generateAnswers ? "active" : ""} onClick={() => setGenerateAnswers(false)}>不生成回答</button>
<button type="button" className={generateAnswers ? "active" : ""} onClick={() => setGenerateAnswers(true)}>自动生成回答</button> <button type="button" disabled={busy || !!retryJobId} className={generateAnswers ? "active" : ""} onClick={() => setGenerateAnswers(true)}>自动生成回答</button>
</div> </div>
{generateAnswers && ( {generateAnswers && (
<label> <label>
每个问题生成回答条数 每个问题生成回答条数
<input type="number" min="1" max="20" step="1" value={answersPerQuestion} onChange={(event) => setAnswersPerQuestion(event.target.value)} /> <input type="number" disabled={busy || !!retryJobId} min="1" max="20" step="1" value={answersPerQuestion} onChange={(event) => setAnswersPerQuestion(event.target.value)} />
<small>可填写 1-20 条,默认 2 条</small> <small>可填写 1-20 条,默认 2 条</small>
</label> </label>
)} )}
</fieldset> </fieldset>
{busy && !questions.length && <div className="ask-import-status">正在解析文件…</div>} {busy && !questions.length && <div className="ask-import-status">正在解析文件…</div>}
{jobProgress && (
<div className={`assistant-import-progress ${jobProgress.failed ? "has-failures" : ""}`}>
<header><span>{jobProgress.status === "completed" ? "回答生成完成" : jobProgress.status === "partial_failed" || jobProgress.status === "failed" ? "部分问题生成失败" : "正在生成回答"}</span><b>{jobProgress.progress_percent || 0}%</b></header>
<div className="assistant-import-progress-track"><i style={{ width: `${jobProgress.progress_percent || 0}%` }} /></div>
<p><span>共 {jobProgress.total || 0} 条</span><span>已完成 <b>{jobProgress.completed || 0}</b></span><span>失败 <b>{jobProgress.failed || 0}</b></span><span>待处理 <b>{jobProgress.pending || 0}</b></span></p>
</div>
)}
{error && <div className="form-error">{error}</div>} {error && <div className="form-error">{error}</div>}
{questions.length > 0 && ( {questions.length > 0 && (
<div className="ask-import-preview"> <div className="ask-import-preview">
@ -1149,7 +1167,7 @@ function AssistantQuestionImportModal({ form, onClose, onImport }) {
<div className="manual-modal-actions"> <div className="manual-modal-actions">
<button type="button" disabled={busy} onClick={onClose}>取消</button> <button type="button" disabled={busy} onClick={onClose}>取消</button>
<button type="submit" className="primary-button" disabled={!questions.length || busy}> <button type="submit" className="primary-button" disabled={!questions.length || busy}>
{busy ? (generateAnswers ? "正在生成回答…" : "正在导入…") : `导入 ${questions.length || 0} 条问题`} {busy ? (jobProgress ? `正在生成回答(${jobProgress.completed || 0}/${jobProgress.total || questions.length})…` : generateAnswers ? "正在生成回答…" : "正在导入…") : retryJobId ? `重试 ${jobProgress?.failed || 0} 条失败问题` : `导入 ${questions.length || 0} 条问题`}
</button> </button>
</div> </div>
</form> </form>
@ -1867,10 +1885,10 @@ function AssistantPageFinal() {
]); ]);
setManualOpen(false); setManualOpen(false);
}; };
async function addImportedQuestions(questions, generateAnswers, answersPerQuestion) { async function addImportedQuestions(questions, generateAnswers, answersPerQuestion, jobOptions = {}) {
const existing = new Set(items.map((item) => String(item.question || "").trim().toLocaleLowerCase())); const existing = new Set(items.map((item) => String(item.question || "").trim().toLocaleLowerCase()));
const newQuestions = questions.filter((question) => !existing.has(String(question || "").trim().toLocaleLowerCase())); const newQuestions = questions.filter((question) => !existing.has(String(question || "").trim().toLocaleLowerCase()));
if (!newQuestions.length) { if (!newQuestions.length && !jobOptions.jobId) {
setSavedMessage("导入的问题已全部存在于当前列表中"); setSavedMessage("导入的问题已全部存在于当前列表中");
setImportOpen(false); setImportOpen(false);
return; return;
@ -1884,7 +1902,12 @@ function AssistantPageFinal() {
form.selling_points, form.selling_points,
]); ]);
if (intentError) throw new Error(`自动生成回答前,${intentError}`); if (intentError) throw new Error(`自动生成回答前,${intentError}`);
const result = await api("/api/admin/qa/generate-answers", { let job;
if (jobOptions.jobId) {
const retryResult = await api(`/api/admin/qa/generate-answer-jobs/${jobOptions.jobId}/retry`, { method: "POST" });
job = retryResult.data;
} else {
const createResult = await api("/api/admin/qa/generate-answer-jobs", {
method: "POST", method: "POST",
body: JSON.stringify({ body: JSON.stringify({
...form, ...form,
@ -1892,7 +1915,22 @@ function AssistantPageFinal() {
answers_per_question: answersPerQuestion, answers_per_question: answersPerQuestion,
}), }),
}); });
importedItems = result.data.items || []; job = createResult.data;
}
jobOptions.onProgress?.(job);
while (["queued", "processing"].includes(job.status)) {
await new Promise((resolve) => window.setTimeout(resolve, 1200));
const progressResult = await api(`/api/admin/qa/generate-answer-jobs/${job.id}`);
job = progressResult.data;
jobOptions.onProgress?.(job);
}
if (job.status !== "completed") {
const failure = new Error(`已完成 ${job.completed} 条,${job.failed} 条生成失败。点击下方按钮仅重试失败问题。`);
failure.bulkJob = job;
throw failure;
}
const detailResult = await api(`/api/admin/qa/generate-answer-jobs/${job.id}?include_items=true`);
importedItems = detailResult.data.items || [];
} else { } else {
importedItems = newQuestions.map((question) => ({ importedItems = newQuestions.map((question) => ({
question, question,