194 lines
10 KiB
Python
194 lines
10 KiB
Python
import json
|
||
import base64
|
||
import logging
|
||
import re
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
from fastapi import HTTPException, status
|
||
|
||
from .config import settings
|
||
from .schemas import QAGenerateRequest
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _prompt(payload: QAGenerateRequest) -> str:
|
||
style = "带真实体验的轻度种草语气" if payload.seed_style else "客观自然，不刻意种草"
|
||
return f"""你是电商问大家内容策划。请根据商品条件生成真实、具体、互不重复的消费者提问和参考回答。
|
||
商品名称：{payload.product_name or '未提供'}
|
||
商品品类：{payload.category or '未提供'}
|
||
商品功效：{payload.benefits or '未提供'}
|
||
商品卖点：{payload.selling_points or '未提供'}
|
||
语言风格：{payload.language_style}
|
||
目标人群：{payload.target_group}
|
||
使用场景：{payload.usage_scene}
|
||
购买渠道：{payload.purchase_channel}
|
||
回答要求：{payload.answer_requirement or '直接回应问题，避免夸大功效'}
|
||
回答语气：{style}
|
||
真实问答参考：{payload.reference_qa or '未提供'}
|
||
|
||
请生成 {payload.question_count} 个问题，每个问题生成 {payload.answers_per_question} 个不同角度的参考回答。
|
||
约束：问题尽量不超过20字，回答尽量不超过50字；不虚构成分、检测、认证和医疗功效；表达像真实买家；不要重复商品名称作为开头。
|
||
如果提供了真实问答参考，请优先学习其中的提问角度、真实语气和回答结构，但不要照抄原文，不要把参考中的其他商品信息误套到当前商品。
|
||
只返回严格 JSON，不要 Markdown 代码块或额外解释，格式必须是：
|
||
{{"items":[{{"question":"问题文本","answers":["回答1","回答2"]}}]}}"""
|
||
|
||
|
||
def _parse_content(content: str, expected_answers: int) -> list[dict]:
|
||
cleaned = content.strip()
|
||
if cleaned.startswith("```"):
|
||
cleaned = re.sub(r"^```(?:json)?\s*|\s*```$", "", cleaned, flags=re.I | re.S).strip()
|
||
try:
|
||
parsed = json.loads(cleaned)
|
||
except json.JSONDecodeError as exc:
|
||
start, end = cleaned.find("{"), cleaned.rfind("}")
|
||
if start < 0 or end <= start:
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="大模型返回格式无法解析") from exc
|
||
parsed = json.loads(cleaned[start:end + 1])
|
||
raw_items = parsed.get("items") if isinstance(parsed, dict) else parsed
|
||
if not isinstance(raw_items, list):
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="大模型未返回有效问答列表")
|
||
items = []
|
||
for index, raw in enumerate(raw_items):
|
||
if not isinstance(raw, dict) or not str(raw.get("question", "")).strip():
|
||
continue
|
||
answers = raw.get("answers", [])
|
||
if isinstance(answers, str):
|
||
answers = [answers]
|
||
answers = [str(answer).strip() for answer in answers if str(answer).strip()][:expected_answers]
|
||
if answers:
|
||
items.append({"id": f"Q{index + 1:03d}", "question": str(raw["question"]).strip(), "answers": answers, "source": "AI生成", "status": "待采纳", "accepted": False})
|
||
if not items:
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="大模型未返回有效问答内容")
|
||
return items
|
||
|
||
|
||
def generate_qa(payload: QAGenerateRequest) -> list[dict]:
|
||
if not settings.llm_api_key:
|
||
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="尚未配置大模型 API Key，请先设置 LLM_API_KEY")
|
||
url = settings.llm_base_url.rstrip("/") + "/chat/completions"
|
||
request_body = {"model": settings.llm_model, "temperature": 0.8, "messages": [{"role": "system", "content": "你只输出合法 JSON。"}, {"role": "user", "content": _prompt(payload)}]}
|
||
request = urllib.request.Request(url, data=json.dumps(request_body).encode("utf-8"), headers={"Content-Type": "application/json", "Authorization": f"Bearer {settings.llm_api_key}"}, method="POST")
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=settings.llm_timeout_seconds) as response:
|
||
result = json.loads(response.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as exc:
|
||
logger.error("LLM HTTP error: status=%s url=%s", exc.code, url)
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"大模型服务返回错误（HTTP {exc.code}）") from exc
|
||
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
|
||
logger.exception("LLM request failed: url=%s reason=%s", url, exc)
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="大模型服务暂时不可用，请稍后重试") from exc
|
||
try:
|
||
content = result["choices"][0]["message"]["content"]
|
||
except (KeyError, IndexError, TypeError) as exc:
|
||
logger.error("LLM response missing choices/message/content: keys=%s", list(result) if isinstance(result, dict) else type(result).__name__)
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="大模型响应缺少内容") from exc
|
||
return _parse_content(content, payload.answers_per_question)
|
||
|
||
|
||
def generate_answer_for_question(product_title: str, question: str) -> str:
|
||
"""Generate one concise answer. Falls back to a safe draft during the simulated-OCR MVP."""
|
||
if not settings.llm_api_key:
|
||
return f"结合{product_title}的使用体验来看，建议按自身肤质先少量试用，确认合适后再正常使用。"
|
||
url = settings.llm_base_url.rstrip("/") + "/chat/completions"
|
||
prompt = f"商品：{product_title}\n问题：{question}\n请生成一条50字以内、自然真实、不夸大功效的淘宝问大家回答。只输出回答正文。"
|
||
request_body = {"model": settings.llm_model, "temperature": 0.7, "messages": [{"role": "system", "content": "你是电商问大家回答助手。"}, {"role": "user", "content": prompt}]}
|
||
request = urllib.request.Request(url, data=json.dumps(request_body).encode("utf-8"), headers={"Content-Type": "application/json", "Authorization": f"Bearer {settings.llm_api_key}"}, method="POST")
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=settings.llm_timeout_seconds) as response:
|
||
result = json.loads(response.read().decode("utf-8"))
|
||
answer = str(result["choices"][0]["message"]["content"]).strip()
|
||
return answer[:300]
|
||
except Exception:
|
||
logger.exception("Answer generation failed for simulated OCR question")
|
||
return f"结合{product_title}的使用体验来看，建议按自身肤质先少量试用，确认合适后再正常使用。"
|
||
|
||
|
||
def _parse_ocr_content(content: str) -> list[dict]:
|
||
if isinstance(content, list):
|
||
content = "".join(str(item.get("text", "")) for item in content if isinstance(item, dict))
|
||
cleaned = str(content or "").strip()
|
||
if cleaned.startswith("```"):
|
||
cleaned = re.sub(r"^```(?:json)?\s*|\s*```$", "", cleaned, flags=re.I | re.S).strip()
|
||
try:
|
||
parsed = json.loads(cleaned)
|
||
except json.JSONDecodeError:
|
||
start, end = cleaned.find("{"), cleaned.rfind("}")
|
||
if start < 0 or end <= start:
|
||
return []
|
||
try:
|
||
parsed = json.loads(cleaned[start:end + 1])
|
||
except json.JSONDecodeError:
|
||
return []
|
||
raw_items = parsed.get("items") if isinstance(parsed, dict) else parsed
|
||
if not isinstance(raw_items, list):
|
||
return []
|
||
items = []
|
||
seen = set()
|
||
for raw in raw_items:
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
question = re.sub(r"^[\s\d.#、.)（()]+", "", str(raw.get("question", ""))).strip()
|
||
if len(question) < 4 or question in seen:
|
||
continue
|
||
if any(token in question for token in ("回答", "规则", "待回答", "已回答", "已提问")):
|
||
continue
|
||
seen.add(question)
|
||
items.append({
|
||
"product_title": str(raw.get("product_title", "待识别商品")).strip() or "待识别商品",
|
||
"question": question[:300],
|
||
})
|
||
return items[:100]
|
||
|
||
|
||
def extract_questions_from_images(images: list[tuple[bytes, str]]) -> list[dict]:
|
||
"""Extract consumer questions from one or more Taobao Q&A screenshots.
|
||
|
||
The configured OpenAI-compatible model must support image input. The
|
||
caller keeps a deterministic fallback when no vision model/API key is
|
||
configured, so uploading screenshots never blocks the task workflow.
|
||
"""
|
||
if not settings.llm_api_key or not images:
|
||
return []
|
||
content = [{
|
||
"type": "text",
|
||
"text": (
|
||
"请识别这些淘宝‘问答/待回答’页面截图中的消费者提问。只提取真正的问题句子，"
|
||
"不要提取导航、按钮、商品名、用户昵称、规则文字或页面标题。去除序号并保持原文。"
|
||
"多张截图合并去重。只返回严格 JSON："
|
||
'{"items":[{"product_title":"商品名或待识别商品","question":"问题原文"}]}'
|
||
),
|
||
}]
|
||
for index, (image_data, content_type) in enumerate(images, start=1):
|
||
encoded = base64.b64encode(image_data).decode("ascii")
|
||
content.append({"type": "text", "text": f"截图 {index}"})
|
||
content.append({
|
||
"type": "image_url",
|
||
"image_url": {"url": f"data:{content_type};base64,{encoded}"},
|
||
})
|
||
request_body = {
|
||
"model": settings.llm_vision_model,
|
||
"temperature": 0,
|
||
"messages": [
|
||
{"role": "system", "content": "你是严谨的中文电商截图 OCR 提取器，只输出合法 JSON。"},
|
||
{"role": "user", "content": content},
|
||
],
|
||
}
|
||
url = settings.llm_base_url.rstrip("/") + "/chat/completions"
|
||
request = urllib.request.Request(
|
||
url,
|
||
data=json.dumps(request_body).encode("utf-8"),
|
||
headers={"Content-Type": "application/json", "Authorization": f"Bearer {settings.llm_api_key}"},
|
||
method="POST",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=settings.llm_timeout_seconds) as response:
|
||
result = json.loads(response.read().decode("utf-8"))
|
||
content = result["choices"][0]["message"]["content"]
|
||
return _parse_ocr_content(content)
|
||
except Exception:
|
||
logger.exception("Question OCR failed")
|
||
return []
|