Toolbox/dev_test_scripts/tools/tool_qa_mall_logos.py

# -*- coding: utf-8 -*-
# ============================================================
# tool_qa_mall_logos.py - 宝龙美食打卡 lab · logo 裁切抽检（方案 C）
#
# 用视觉大模型 qwen-vl-plus-latest（Rule 11 视觉/多模态选型）对方案 A 的裁切结果做抽检：
#   按行拼版（每行 11 格，格顶标注 id），连同 id->店名 清单一起发给模型，
#   让它逐格检查：logo 是否被切边 / 串入邻格 / 带店名文字条 / 基本空白 / 与店名明显不符。
# 模型只读抽检、不改图；结果落 report.json 供人工复核决定是否重裁。
#
# 凭据：从 ~/Toolbox_local_creds.env.local 读 DASHSCOPE_API_KEY / DASHSCOPE_API_HOST_FOR_OPENAI
#       （零硬编码，Rule 13）
# 用法：python tool_qa_mall_logos.py
# ============================================================
import base64
import json
import os
import sys
import urllib.request

import cv2
import numpy as np

HERE = os.path.dirname(os.path.abspath(__file__))
IMAGES_DIR = r'd:\Trae_Files\TRAE-Toolbox\public\tools\thought_lab\labs\mall_food_checkin\images'
SEED = r'd:\Trae_Files\TRAE-Toolbox\src\server\thought_lab\labs\mall_food_checkin\shops_seed.json'
REPORT = r'd:\Trae_Files\TRAE-Toolbox\dev_test_scripts\debug\mall_qa_report.json'
CREDS = os.path.join(os.path.expanduser('~'), 'Toolbox_local_creds.env.local')
# 视觉模型：Rule 11 推荐 qwen-vl-plus-latest，但本 key 实测 403；
# 与项目现网口径（plant_home/style_check/yuanzhupai）一致用 qwen-vl-plus，可用 --model 覆盖
MODEL = 'qwen-vl-plus'
ROWS, COLS = 7, 11


def load_creds():
    env = {}
    try:
        for line in open(CREDS, encoding='utf-8'):
            line = line.strip()
            if not line or line.startswith('#') or '=' not in line:
                continue
            k, v = line.split('=', 1)
            env[k.strip()] = v.strip().strip('"').strip("'")
    except OSError:
        pass
    return env


def imread_u(path):
    data = np.fromfile(path, dtype=np.uint8)
    return cv2.imdecode(data, cv2.IMREAD_COLOR)


def imencode_b64(img):
    ok, buf = cv2.imencode('.jpg', img, [cv2.IMWRITE_JPEG_QUALITY, 85])
    if not ok:
        return ''
    return base64.b64encode(buf.tobytes()).decode('ascii')


def build_row_sheet(ids, crops):
    cw, ch, lab_h = 400, 260, 26
    sheet = np.full((ch + lab_h, cw * len(ids), 3), 255, np.uint8)
    for i, sid in enumerate(ids):
        x = i * cw
        cv2.putText(sheet, sid, (x + 8, 19), cv2.FONT_HERSHEY_SIMPLEX, 0.65, (0, 0, 0), 2)
        c = crops.get(sid)
        if c is None:
            cv2.putText(sheet, 'MISS', (x + 150, 150), cv2.FONT_HERSHEY_SIMPLEX, 1.0, (0, 0, 255), 3)
            continue
        sc = min((cw - 10) / float(c.shape[1]), (ch - 10) / float(c.shape[0]))
        cc = cv2.resize(c, (max(1, int(c.shape[1] * sc)), max(1, int(c.shape[0] * sc))))
        sheet[lab_h:lab_h + cc.shape[0], x + 5:x + 5 + cc.shape[1]] = cc
    return sheet


def ask_vl(api_base, api_key, sheet, names_text):
    body = {
        'model': MODEL,
        'messages': [{
            'role': 'user',
            'content': [
                {'type': 'text', 'text': (
                    '图中是若干张饭店 logo 裁切图，从左到右每格顶部标注了 id。'
                    'id 与店名对照：' + names_text + '。'
                    '判定口径（严格，宁漏报勿误报）：允许轻微白边、允许画面偏暗或过曝、允许构图不完美；'
                    '只有以下明显缺陷才报告：cut(logo 主体被裁掉超过 1/4) / neighbor(明显串入邻格的另一张图) / '
                    'label(底部带白底黑字店名文字条) / blank(整格基本是空白板面) / mismatch(画面内容与店名完全对不上，'
                    '例如咖啡店格子里是火锅)。拿不准的一律视为合格。'
                    '只输出 JSON 数组，元素形如 {"id":"r0c1","problem":"cut"}；全部合格则输出 []。'
                    '不要输出任何其他文字。')},
                {'type': 'image_url', 'image_url': {'url': 'data:image/jpeg;base64,' + imencode_b64(sheet)}}
            ]
        }],
        'temperature': 0.1
    }
    req = urllib.request.Request(
        api_base.rstrip('/') + '/chat/completions',
        data=json.dumps(body).encode('utf-8'),
        headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + api_key},
        method='POST')
    with urllib.request.urlopen(req, timeout=120) as r:
        data = json.loads(r.read().decode('utf-8'))
    text = data['choices'][0]['message']['content']
    text = text.strip()
    if text.startswith('```'):
        text = text.strip('`')
        if text.startswith('json'):
            text = text[4:]
    return json.loads(text.strip())


def main():
    global MODEL
    if '--model' in sys.argv:
        MODEL = sys.argv[sys.argv.index('--model') + 1]
    creds = load_creds()
    api_key = creds.get('DASHSCOPE_API_KEY', '') or os.environ.get('DASHSCOPE_API_KEY', '')
    api_base = creds.get('DASHSCOPE_API_HOST_FOR_OPENAI', '') or os.environ.get('DASHSCOPE_API_HOST_FOR_OPENAI', '')
    if not api_key or not api_base:
        print('缺少 DASHSCOPE_API_KEY / DASHSCOPE_API_HOST_FOR_OPENAI（检查 ~/Toolbox_local_creds.env.local）')
        sys.exit(1)

    seed = json.load(open(SEED, encoding='utf-8'))
    name_of = {s['id']: s['name'] for s in seed['shops']}
    crops = {}
    for sid in name_of:
        p = os.path.join(IMAGES_DIR, sid + '.jpg')
        if os.path.exists(p):
            crops[sid] = imread_u(p)

    all_ids = ['r%dc%d' % (ri, ci) for ri in range(ROWS) for ci in range(COLS)]
    report = {'model': MODEL, 'batches': []}
    for bi in range(0, len(all_ids), 4):
        ids = all_ids[bi:bi + 4]
        names_text = '、'.join('%s=%s' % (i, name_of.get(i, '?')) for i in ids)
        sheet = build_row_sheet(ids, crops)
        try:
            bad = ask_vl(api_base, api_key, sheet, names_text)
        except Exception as e:  # noqa: BLE001
            print('%s 抽检失败: %s' % (ids[0], e))
            report['batches'].append({'ids': ids, 'error': str(e)})
            continue
        print('%s -> %s' % (ids[0], json.dumps(bad, ensure_ascii=False)))
        report['batches'].append({'ids': ids, 'bad': bad})
    json.dump(report, open(REPORT, 'w', encoding='utf-8'), ensure_ascii=False, indent=1)
    total = sum(len(b.get('bad', [])) for b in report['batches'] if isinstance(b.get('bad'), list))
    print('抽检完成，问题格合计 %d，报告: %s' % (total, REPORT))


if __name__ == '__main__':
    main()