pq-web-ai/app/providers/understand_image_provider.py

from openai import OpenAI
from ..schemas.base import DataResponse
from ..core.config import get_settings
from ..core.logger import get_logger
from openai.types.chat import ChatCompletionToolParam
import json

settings = get_settings()
logger = get_logger("understand_image_provider")

SYSTEM_PROMPT = """
<SystemPrompt>
    <角色>
        你是一名资深广告文案专家。你的任务是根据输入的一张广告图片中的文字内容，生成一句简洁有力的广告文案。
    </角色>

    <受众>
        目标用户：50岁以上中老年人。语言需亲切、直白、易理解，避免专业术语与复杂长句。
    </受众>

    <生成逻辑>
        1. 首先对用户提供的广告图片进行OCR识别，提取出所有文字信息。  
        2. 文案生成必须严格基于识别到的文本内容，不得编造图片中不存在的优惠、动作指令或承诺。  
        3. 若图片中没有明确的转化行为，则默认使用“长按二维码”作为行动指令。  
    </生成逻辑>

    <结构公式>
        [行动指令]，[低门槛/优惠承诺]，[核心价值/具体收益]；[紧迫感/稀缺性提醒]
    </结构公式>

    <转化策略>
        <条目>文案必须以「[行动指令] + [低门槛/优惠承诺]」连续开头，但允许等义表述，以提升多样性。</条目>
        <条目>可接受的等义表述示例：
            行动指令（任选其一，按素材出现）：“长按二维码”“扫码二维码”“识别二维码”“长按识别”；  
            低门槛/优惠承诺（任选其一，按素材出现）：“0元入群”“免费进群”“0元加入”“限时免费加入”。  
        </条目>
        <条目>若仅满足其中之一（真实存在且与素材一致），则仅前置该项；严禁捏造另一项。</条目>
        <条目>若两项均不在素材中出现，则改用图片中存在的真实动作入口与优惠描述（如“点击查看”“立即预约”“限时立减”等）。</条目>
        <条目>上述两要素需置于句首，越靠前越好；其后再写核心收益与稀缺提醒。</条目>
    </转化策略>

    <约束>
        1. 文案必须准确传达广告图片中的产品/服务信息，不得杜撰不存在的内容。  
        2. 加入紧迫感或稀缺性（如“限时”“名额有限”“马上领取”等），但不得虚构或夸大事实。  
        3. 避免医疗或功效的绝对化/保证性用语（如“治愈”“根治”“无副作用”“永久有效”）。  
        4. 不得包含违法、虚假、低俗、敏感、歧视性内容，不引导危险行为，不传播迷信。  
        5. 涉及健康/养生场景时，表述应为辅助/改善/建议性质，不承诺疗效。  
        6. 仅输出一句中文广告文案，简短醒目，适合作为宣传主标题。  
        7. 标点与短句分隔：动作、优惠承诺、核心收益之间用逗号分隔；紧迫感/稀缺性提醒用分号与前半部分隔开；全句≤50字（含标点）。  
    </约束>

    <示例 few-shot="true">
        长按二维码，0元入群，领取中医调理养生建议；名额有限，赶快行动吧
    </示例>
    <示例 few-shot="true">
        扫码二维码，免费进群，获取控糖日常饮食要点；数量有限，立即参加！
    </示例>
    <示例 few-shot="true">
        识别二维码，0元加入，领取适合中老年人的养生课程；限时开放，先到先得！
    </示例>

    <自检>
        在输出前自检：是否基于图片文字生成；是否以「行动指令 + 低门槛/优惠承诺」连续开头（或在素材不支持时作真实替换）；是否遵循标点与长度；是否包含真实的稀缺/紧迫提醒；是否合规且未承诺疗效。
    </自检>

    <输出要求>
        仅输出生成的一句广告文案，不要附加解释或其他信息。
    </输出要求>
</SystemPrompt>
"""

tools: list[ChatCompletionToolParam] = [
    {
        "type": "function",
        "function": {
            "name": "generate_ocr_text",
            "description": "生成一句适合中老年用户的广告文案（遵循结构公式与约束）",
            "parameters": {
                "type": "object",
                "properties": {
                    "ocr_text": {
                        "type": "string",
                        "description": "最终的一句广告文案（中文，简短醒目，合规）"
                    }
                },
                "required": ["ocr_text"],
                "additionalProperties": False
            }
        }
    }
]

class UnderstandImageProvider:
    print("UnderstandImageProvider called")
    def understand_image(self, image_url: str, *, model: str) -> DataResponse:

        client = OpenAI(
            api_key = settings.dashscope_api_key or "",
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        )

        if not client:
            logger.error("OpenAI client is not initialized.")
            return DataResponse(code=1, data=None, msg=f"OpenAI client is not initialized")

        completion = client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {
                    "role": "user", 
                    "content": [{ "type": "image_url", "image_url": { "url": image_url } }],
                },
            ],
            tools=tools,
            tool_choice={
                "type": "function",
                "function": {"name": "generate_ocr_text"}
            },
            temperature=0.5
        )

        msg = completion.choices[0].message
        # Safely parse tool call arguments (if any)
        ocr_text = ""
        try:
            tool_calls = getattr(msg, "tool_calls", None) or []
            if tool_calls:
                call = tool_calls[0]
                arg_str = getattr(getattr(call, "function", None), "arguments", None)
                if isinstance(arg_str, str) and arg_str.strip():
                    args = json.loads(arg_str)
                    if isinstance(args, dict):
                        ocr_text = str(args.get("ocr_text", "")).strip()
        except Exception as e:
            logger.error("parse tool call failed: %s", e, exc_info=True)
            return DataResponse(code=1, data=None, msg=f"parse tool call failed: {e}")

        # Fallback: if no tool-calls returned, try to read text content
        content = getattr(msg, "content", None)
        if not ocr_text and isinstance(content, str):
            ocr_text = content.strip()

        print("✅ OCR_TEXT:\n", ocr_text)

        return DataResponse(code=0, data=ocr_text, msg="success")