import json import base64 import io import logging import re import time 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 _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 = [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_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 = [str(answer).strip() for answer in answers if str(answer).strip()] 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 _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 = 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 '未提供'} 只返回严格 JSON,不要 Markdown 或解释,格式:{{"questions":["问题1","问题2"]}}。问题应像真实买家提问,尽量不超过30字,不要虚构具体参数或功效。""" 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 = str(value).strip() 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 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 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 _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]