483 lines
25 KiB
Python
483 lines
25 KiB
Python
import json
|
||
import base64
|
||
import io
|
||
import logging
|
||
import re
|
||
import time
|
||
import unicodedata
|
||
import urllib.error
|
||
import urllib.request
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
|
||
from PIL import Image, ImageOps
|
||
|
||
from fastapi import HTTPException, status
|
||
|
||
from .config import settings
|
||
from .schemas import QABulkAnswerRequest, QAGenerateRequest
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _natural_question(value: object) -> str:
|
||
"""Keep generated questions conversational and free of forced punctuation."""
|
||
text = re.sub(r"\s+", " ", str(value or "")).strip()
|
||
return re.sub(r"[?？。.!！]+$", "", text).rstrip()
|
||
|
||
|
||
def _natural_answer(value: object) -> str:
|
||
text = re.sub(r"\s+", " ", str(value or "")).strip()
|
||
text = re.sub(r"^(?:回答|答复|参考回答)\s*[:：]\s*", "", text).strip()
|
||
while text and (unicodedata.category(text[-1]).startswith("P") or text[-1] in "~～"):
|
||
text = text[:-1].rstrip()
|
||
return text
|
||
|
||
|
||
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 or '客观自然'}
|
||
目标人群：{payload.target_group or '普通消费者'}
|
||
使用场景：{payload.usage_scene or '日常使用'}
|
||
购买渠道：{payload.purchase_channel or '电商平台'}
|
||
回答要求：{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 = [cleaned for answer in answers if (cleaned := _natural_answer(answer))][:expected_answers]
|
||
if answers:
|
||
question = _natural_question(raw["question"])
|
||
if question:
|
||
items.append({"id": f"Q{index + 1:03d}", "question": question, "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_batch(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_qa(payload: QAGenerateRequest) -> list[dict]:
|
||
"""Generate the requested amount in small batches so large requests are not truncated."""
|
||
target = payload.question_count
|
||
items: list[dict] = []
|
||
seen: set[str] = set()
|
||
attempts = 0
|
||
max_attempts = max((target + 19) // 20 + 2, 3)
|
||
while len(items) < target and attempts < max_attempts:
|
||
batch_size = min(20, target - len(items))
|
||
batch_payload = payload.model_copy(update={"question_count": batch_size})
|
||
for item in _generate_qa_batch(batch_payload):
|
||
key = item["question"].strip()
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
items.append(item)
|
||
if len(items) >= target:
|
||
break
|
||
attempts += 1
|
||
if len(items) < target:
|
||
logger.warning("LLM returned %s/%s unique QA items", len(items), target)
|
||
for index, item in enumerate(items[:target], start=1):
|
||
item["id"] = f"Q{index:03d}"
|
||
return items[:target]
|
||
|
||
|
||
def _bulk_answers_prompt(payload: QABulkAnswerRequest, questions: list[str]) -> str:
|
||
style = "带真实体验的轻度种草语气" if payload.seed_style else "客观自然，不刻意种草"
|
||
question_lines = "\n".join(f"{index}. {question}" for index, question in enumerate(questions, start=1))
|
||
return f"""你是电商问大家回答助手。请严格按照给定问题逐条生成参考回答，不要新增、删除、合并或改写问题。
|
||
商品名称：{payload.product_name or '未提供'}
|
||
商品品类：{payload.category or '未提供'}
|
||
商品功效：{payload.benefits or '未提供'}
|
||
商品卖点：{payload.selling_points or '未提供'}
|
||
语言风格：{payload.language_style or '客观自然'}
|
||
目标人群：{payload.target_group or '普通消费者'}
|
||
使用场景：{payload.usage_scene or '日常使用'}
|
||
购买渠道：{payload.purchase_channel or '电商平台'}
|
||
回答要求：{payload.answer_requirement or '直接回应问题，避免夸大功效'}
|
||
回答语气：{style}
|
||
|
||
问题列表：
|
||
{question_lines}
|
||
|
||
每个问题生成 {payload.answers_per_question} 条不同角度的回答。回答要像普通买家在手机上随手回复，简短口语化，可自然使用“我用着、感觉、还行、挺、不会”等日常表达；避免客服腔、说明书语气、广告词和整齐统一的句式。不同回答的开头、长短和语气要有变化。回答末尾一律不要加任何标点符号。回答尽量不超过50字，不虚构成分、检测、认证、医疗功效或未提供的使用事实。
|
||
只返回严格 JSON，不要 Markdown 或解释。必须按问题序号原顺序返回，格式：
|
||
{{"items":[{{"index":1,"answers":["回答1","回答2"]}}]}}"""
|
||
|
||
|
||
def _generate_bulk_answers_batch(payload: QABulkAnswerRequest, questions: list[str]) -> list[list[str]]:
|
||
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.75,
|
||
"messages": [
|
||
{"role": "system", "content": "你只输出合法 JSON，并严格保持输入问题的数量和顺序。"},
|
||
{"role": "user", "content": _bulk_answers_prompt(payload, questions)},
|
||
],
|
||
}
|
||
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 = str(result["choices"][0]["message"]["content"]).strip()
|
||
cleaned = re.sub(r"^```(?:json)?\s*|\s*```$", "", content, flags=re.I | re.S).strip()
|
||
parsed = json.loads(cleaned)
|
||
raw_items = parsed.get("items") if isinstance(parsed, dict) else parsed
|
||
if not isinstance(raw_items, list):
|
||
raise ValueError("items is not a list")
|
||
by_index: dict[int, list[str]] = {}
|
||
for raw in raw_items:
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
try:
|
||
index = int(raw.get("index"))
|
||
except (TypeError, ValueError):
|
||
continue
|
||
answers = raw.get("answers") or []
|
||
if isinstance(answers, str):
|
||
answers = [answers]
|
||
clean_answers = [cleaned for answer in answers if (cleaned := _natural_answer(answer))]
|
||
if len(clean_answers) >= payload.answers_per_question:
|
||
by_index[index] = clean_answers[:payload.answers_per_question]
|
||
if any(index not in by_index for index in range(1, len(questions) + 1)):
|
||
raise ValueError("answers are incomplete")
|
||
return [by_index[index] for index in range(1, len(questions) + 1)]
|
||
except urllib.error.HTTPError as exc:
|
||
logger.error("LLM bulk-answer 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, KeyError, IndexError, TypeError, ValueError) as exc:
|
||
logger.exception("LLM bulk-answer generation failed")
|
||
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]:
|
||
"""Generate answers for exact imported questions without rewriting them."""
|
||
items: list[dict] = []
|
||
# Keep each response small enough that the provider can return every
|
||
# requested answer without truncating the JSON payload.
|
||
batch_size = max(1, min(10, 30 // payload.answers_per_question))
|
||
for offset in range(0, len(questions), batch_size):
|
||
batch = questions[offset:offset + batch_size]
|
||
answer_groups = _generate_bulk_answers_batch(payload, batch)
|
||
for question, answers in zip(batch, answer_groups):
|
||
items.append({
|
||
"id": f"Q{len(items) + 1:03d}",
|
||
"question": question,
|
||
"answers": answers,
|
||
"source": "批量导入（AI回答）",
|
||
"status": "待采纳",
|
||
"accepted": False,
|
||
})
|
||
return items
|
||
|
||
|
||
def _questions_prompt(payload: dict, count: int, reference_qa: str = "") -> str:
|
||
return f"""你是电商问大家提问策划。请根据商品信息生成 {count} 个真实、具体、互不重复的消费者提问。
|
||
商品名称：{payload.get('product_name') or '未提供'}
|
||
商品品类：{payload.get('category') or '未提供'}
|
||
商品功效：{payload.get('benefits') or '未提供'}
|
||
商品卖点：{payload.get('selling_points') or '未提供'}
|
||
语言风格：{payload.get('language_style') or '客观自然'}
|
||
目标人群：{payload.get('target_group') or '普通消费者'}
|
||
使用场景：{payload.get('usage_scene') or '日常使用'}
|
||
购买渠道：{payload.get('purchase_channel') or '电商平台'}
|
||
真实问答参考：{reference_qa or '未提供'}
|
||
提问必须像买家随手输入：简短、口语化，允许省略主语；不要每句都用“请问、是否、怎么样、如何、能不能”这类标准句式；句式和长短要有变化；末尾不要加问号或其他标点。尽量不超过30字，不要虚构具体参数、功效或使用事实。
|
||
只返回严格 JSON，不要 Markdown 或解释，格式：{{"questions":["问题1","问题2"]}}。"""
|
||
|
||
|
||
def _generate_questions_batch(payload: dict, count: int, reference_qa: str = "") -> list[str]:
|
||
"""Generate only consumer questions for the ask-task workflow."""
|
||
if not settings.llm_api_key:
|
||
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="尚未配置大模型 API Key，无法生成真实提问")
|
||
prompt = _questions_prompt(payload, count, reference_qa)
|
||
url = settings.llm_base_url.rstrip("/") + "/chat/completions"
|
||
body = {"model": settings.llm_model, "temperature": 0.85, "messages": [{"role": "system", "content": "你只输出合法 JSON。"}, {"role": "user", "content": prompt}]}
|
||
request = urllib.request.Request(url, data=json.dumps(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"]
|
||
cleaned = re.sub(r"^```(?:json)?\s*|\s*```$", "", str(content).strip(), flags=re.I | re.S).strip()
|
||
parsed = json.loads(cleaned)
|
||
questions = parsed.get("questions") if isinstance(parsed, dict) else parsed
|
||
if not isinstance(questions, list):
|
||
raise ValueError("questions is not a list")
|
||
output = []
|
||
for value in questions:
|
||
text = _natural_question(value)
|
||
if text and text not in output:
|
||
output.append(text[:200])
|
||
if not output:
|
||
raise ValueError("empty questions")
|
||
return output[:count]
|
||
except urllib.error.HTTPError as exc:
|
||
logger.error("LLM question generation 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, KeyError, IndexError, TypeError, ValueError) as exc:
|
||
logger.exception("LLM question generation failed")
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="大模型提问生成失败，请稍后重试") from exc
|
||
|
||
|
||
def generate_questions(payload: dict, count: int, reference_qa: str = "") -> list[str]:
|
||
"""Generate questions in batches and merge them until the requested count is reached."""
|
||
target = int(count)
|
||
output: list[str] = []
|
||
seen: set[str] = set()
|
||
attempts = 0
|
||
max_attempts = max((target + 19) // 20 + 2, 3)
|
||
while len(output) < target and attempts < max_attempts:
|
||
batch = _generate_questions_batch(payload, min(20, target - len(output)), reference_qa)
|
||
for question in batch:
|
||
if question not in seen:
|
||
seen.add(question)
|
||
output.append(question)
|
||
if len(output) >= target:
|
||
break
|
||
attempts += 1
|
||
if len(output) < target:
|
||
logger.warning("LLM returned %s/%s unique questions", len(output), target)
|
||
return output[:target]
|
||
|
||
|
||
def generate_answer_for_question(product_title: str, question: str) -> str:
|
||
"""Generate one concise answer for an OCR-extracted question."""
|
||
if not settings.llm_api_key:
|
||
return "我用着还行，建议先少量试试，适合自己再继续用"
|
||
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 = _natural_answer(result["choices"][0]["message"]["content"])
|
||
return answer[:300]
|
||
except Exception:
|
||
logger.exception("Answer generation failed for OCR question")
|
||
return "我用着还行，建议先少量试试，适合自己再继续用"
|
||
|
||
|
||
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 _compress_ocr_image(image_data: bytes, content_type: str) -> tuple[bytes, str]:
|
||
"""Normalize screenshots and keep each vision request reasonably small."""
|
||
try:
|
||
with Image.open(io.BytesIO(image_data)) as source:
|
||
image = ImageOps.exif_transpose(source)
|
||
image.thumbnail((2200, 2200), Image.Resampling.LANCZOS)
|
||
if image.mode in {"RGBA", "LA"} or "transparency" in image.info:
|
||
rgba = image.convert("RGBA")
|
||
background = Image.new("RGB", rgba.size, "white")
|
||
background.paste(rgba, mask=rgba.getchannel("A"))
|
||
image = background
|
||
else:
|
||
image = image.convert("RGB")
|
||
|
||
for quality in (86, 76, 66):
|
||
output = io.BytesIO()
|
||
image.save(output, format="JPEG", quality=quality, optimize=True)
|
||
compressed = output.getvalue()
|
||
if len(compressed) <= 1_500_000:
|
||
return compressed, "image/jpeg"
|
||
|
||
# Very long/high-detail screenshots get one final size reduction.
|
||
image.thumbnail((1700, 1700), Image.Resampling.LANCZOS)
|
||
output = io.BytesIO()
|
||
image.save(output, format="JPEG", quality=72, optimize=True)
|
||
return output.getvalue(), "image/jpeg"
|
||
except Exception:
|
||
logger.warning("Could not compress OCR image; using original payload", exc_info=True)
|
||
return image_data, content_type
|
||
|
||
|
||
def _extract_questions_from_image(image_data: bytes, content_type: str) -> list[dict]:
|
||
encoded = base64.b64encode(image_data).decode("ascii")
|
||
content = [
|
||
{
|
||
"type": "text",
|
||
"text": (
|
||
"请识别这张淘宝‘问答/待回答’页面截图中的所有消费者提问。只提取真正的问题句子，"
|
||
"不要提取导航、按钮、商品名、用户昵称、规则文字或页面标题。去除序号并保持原文。"
|
||
"只返回严格 JSON："
|
||
'{"items":[{"product_title":"商品名或待识别商品","question":"问题原文"}]}'
|
||
),
|
||
},
|
||
{
|
||
"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",
|
||
)
|
||
with urllib.request.urlopen(request, timeout=settings.llm_timeout_seconds) as response:
|
||
result = json.loads(response.read().decode("utf-8"))
|
||
return _parse_ocr_content(result["choices"][0]["message"]["content"])
|
||
|
||
|
||
def _recognize_one_image(index: int, image: tuple[bytes, str]) -> tuple[int, list[dict]]:
|
||
image_data, content_type = _compress_ocr_image(*image)
|
||
for attempt in range(2):
|
||
try:
|
||
items = _extract_questions_from_image(image_data, content_type)
|
||
if items:
|
||
return index, items
|
||
raise ValueError("vision model returned no valid questions")
|
||
except Exception:
|
||
logger.warning(
|
||
"Question OCR failed for image %s on attempt %s/2",
|
||
index + 1,
|
||
attempt + 1,
|
||
exc_info=True,
|
||
)
|
||
if attempt == 0:
|
||
time.sleep(0.5)
|
||
return index, []
|
||
|
||
|
||
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. Each
|
||
screenshot is compressed and recognized independently. Failed image
|
||
requests are retried once, then successful results are merged and deduped.
|
||
"""
|
||
if not settings.llm_api_key or not images:
|
||
return []
|
||
recognized: dict[int, list[dict]] = {}
|
||
with ThreadPoolExecutor(max_workers=min(3, len(images))) as executor:
|
||
futures = {
|
||
executor.submit(_recognize_one_image, index, image): index
|
||
for index, image in enumerate(images)
|
||
}
|
||
for future in as_completed(futures):
|
||
index, items = future.result()
|
||
recognized[index] = items
|
||
|
||
merged = []
|
||
seen = set()
|
||
for index in range(len(images)):
|
||
for item in recognized.get(index, []):
|
||
key = re.sub(r"[\s，。！？、,.!?]+", "", item["question"]).lower()
|
||
if not key or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
merged.append({**item, "source_index": index})
|
||
return merged[:100]
|