Files
ewoooc/services/ppt_vision_service.py
OoO 1cf1fd01b1
All checks were successful
CD Pipeline / deploy (push) Successful in 1m30s
修正 Ollama fallback 與 PPT vision payload
2026-05-18 21:32:15 +08:00

609 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
services/ppt_vision_service.py
Operation Ollama-First v5.0 / Phase 14 — PPT 視覺自審
設計原則:
- 用 minicpm-vGCP Primary 已拉5.5GB)對 PPT 截圖做品質檢查
- 替代 qwen2-vl:7bOllama registry 暫無)
- 用途PPT 生成後自動跑視覺檢查,找:
1. 圖表 layout 異常(被切掉、重疊)
2. 文字溢出框
3. 空白區塊(資料未填滿)
4. 配色衝突
- feature flag PPT_VISION_ENABLED 預設 OFF
- 失敗自動 skip不阻擋 PPT 生成主流程)
"""
from __future__ import annotations
import os
import time
import base64
import logging
import shutil
import threading
from io import BytesIO
from dataclasses import dataclass, field
from typing import Optional, Dict, Any, List, Sequence
logger = logging.getLogger(__name__)
# ─────────────────────────────────────────────────────────────────────────────
# Feature flag + 配置
# ─────────────────────────────────────────────────────────────────────────────
PPT_VISION_MODEL = os.getenv('PPT_VISION_MODEL', 'minicpm-v:latest')
PPT_VISION_TIMEOUT = int(os.getenv('PPT_VISION_TIMEOUT', '45'))
PPT_VISION_IMAGE_MAX_EDGE = int(os.getenv('PPT_VISION_IMAGE_MAX_EDGE', '1280'))
PPT_VISION_IMAGE_QUALITY = int(os.getenv('PPT_VISION_IMAGE_QUALITY', '82'))
_AUDIT_LOCK = threading.Lock()
_LAST_AUDIT_RUN: Dict[str, Any] | None = None
def is_ppt_vision_enabled() -> bool:
"""Runtime check避免 import-time freeze"""
return os.getenv('PPT_VISION_ENABLED', 'false').strip().lower() in ('true', '1', 'yes', 'on')
def get_ppt_vision_runtime_status() -> Dict[str, Any]:
"""Expose why the PPT vision pipeline is or is not ready."""
env_value = os.getenv('PPT_VISION_ENABLED')
enabled = is_ppt_vision_enabled()
converter = shutil.which('libreoffice') or shutil.which('soffice')
blockers = []
if not enabled:
blockers.append('PPT_VISION_ENABLED 未設定為 true')
if not converter:
blockers.append('容器內缺少 LibreOffice無法轉換 PPT 做視覺審核')
return {
'enabled': enabled,
'env_value': env_value if env_value is not None else '未設定(預設 false',
'model': os.getenv('PPT_VISION_MODEL', PPT_VISION_MODEL),
'converter': converter,
'converter_ready': bool(converter),
'blockers': blockers,
'ready': enabled and bool(converter),
}
# ─────────────────────────────────────────────────────────────────────────────
# 結果容器
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class VisionResult:
success: bool
issues_found: List[str] = field(default_factory=list) # 問題清單
confidence: float = 0.0 # 0-1模型自評
raw_response: str = ''
duration_ms: int = 0
error: Optional[str] = None
# ─────────────────────────────────────────────────────────────────────────────
# Vision 檢查 prompt繁中強制
# ─────────────────────────────────────────────────────────────────────────────
PPT_VISION_SYSTEM_PROMPT = """你是 momo 電商 PPT 排版品質審核員。
【任務】檢查截圖找出視覺異常,回繁中清單格式:
- 圖表被切掉 / 元素重疊 / 文字溢出框 / 空白區塊(資料未填滿)/ 配色衝突
- 商品名稱顯示不完整 / 數字單位錯誤 / 標題遮擋
【輸出格式】
若無問題:回「✅ 無視覺異常」
若有問題:每行一個問題,格式「⚠️ <問題類型><具體描述>」
【限制】
- 只檢查視覺,不評估內容對錯
- 用繁體中文(台灣用語),絕對禁止簡體字
- 不要寫過多解釋,每個問題一行精簡描述
"""
class PPTVisionService:
"""minicpm-v 視覺檢查服務."""
def __init__(self, model: str = PPT_VISION_MODEL):
self.model = model
def is_available(self) -> bool:
return is_ppt_vision_enabled()
def _encode_image_for_vision(self, image_path: str) -> str:
"""Compress slide screenshots before sending to Ollama vision."""
try:
from PIL import Image
with Image.open(image_path) as im:
image = im.convert('RGB')
image.thumbnail((PPT_VISION_IMAGE_MAX_EDGE, PPT_VISION_IMAGE_MAX_EDGE))
buffer = BytesIO()
image.save(
buffer,
format='JPEG',
quality=max(50, min(PPT_VISION_IMAGE_QUALITY, 95)),
optimize=True,
)
return base64.b64encode(buffer.getvalue()).decode('ascii')
except Exception:
# Pillow is an optimization, not a hard dependency for the vision path.
with open(image_path, 'rb') as f:
return base64.b64encode(f.read()).decode('ascii')
def check_ppt_file(self, pptx_path: str, max_slides: int = 5) -> Dict[str, Any]:
"""檢查整份 .pptx — Phase 26 整合到 PPT 生成流程。
流程:
1. LibreOffice headless 轉 png每張 slide 一張)
2. 對前 N 張跑 check_image
3. 彙總 issues + 平均 confidence
4. fail-safeLibreOffice 不在 / 轉檔失敗 → 回 skip 不阻擋主流程
Returns:
{
'success': bool,
'slides_checked': int,
'total_issues': int,
'issues_by_slide': [(slide_num, [issues...]), ...],
'error': str | None,
}
"""
import os
import subprocess
import tempfile
result = {
'success': False, 'slides_checked': 0, 'total_issues': 0,
'issues_by_slide': [], 'error': None,
}
if not self.is_available():
result['error'] = 'PPT_VISION_ENABLED=false'
return result
if not os.path.isfile(pptx_path):
result['error'] = f'pptx not found: {pptx_path}'
return result
converter = shutil.which('libreoffice') or shutil.which('soffice')
if not converter:
result['error'] = 'libreoffice not installed (skip vision check)'
return result
def _finish_with_error(message: str, duration_ms: int = 0) -> Dict[str, Any]:
result['error'] = message
try:
self._persist_audit_result(
pptx_path=pptx_path,
result=result,
avg_confidence=0.0,
duration_ms=duration_ms,
)
except Exception as e:
logger.warning(f"[PPTVision] persist audit result failed: {e}")
return result
# 1. LibreOffice 轉 png
with tempfile.TemporaryDirectory() as tmpdir:
convert_started = time.monotonic()
try:
proc = subprocess.run(
[converter, '--headless', '--convert-to', 'png',
'--outdir', tmpdir, pptx_path],
capture_output=True, timeout=60,
)
if proc.returncode != 0:
return _finish_with_error(
f'libreoffice convert failed: {proc.stderr.decode()[:200]}',
int((time.monotonic() - convert_started) * 1000),
)
except FileNotFoundError:
result['error'] = 'libreoffice not installed (skip vision check)'
return result
except subprocess.TimeoutExpired:
return _finish_with_error(
'libreoffice convert timeout (60s)',
int((time.monotonic() - convert_started) * 1000),
)
except Exception as e:
return _finish_with_error(
f'{type(e).__name__}: {str(e)[:200]}',
int((time.monotonic() - convert_started) * 1000),
)
# LibreOffice 對 .pptx 預設只輸出第一頁;多頁需 --convert-to png:impress_png_Export
png_files = sorted([
os.path.join(tmpdir, f) for f in os.listdir(tmpdir)
if f.lower().endswith('.png')
])
if not png_files:
return _finish_with_error(
'libreoffice 未產出 png (可能需要 --convert-to png:impress_png_Export)',
int((time.monotonic() - convert_started) * 1000),
)
# 2. 對前 N 張跑 check_image
import time as _time
t0 = _time.monotonic()
confidences = []
slide_errors = []
for idx, png in enumerate(png_files[:max_slides]):
try:
vr = self.check_image(png)
if vr.success:
result['slides_checked'] += 1
confidences.append(vr.confidence)
if vr.issues_found:
result['total_issues'] += len(vr.issues_found)
result['issues_by_slide'].append((idx + 1, vr.issues_found))
else:
slide_errors.append(f"slide {idx + 1}: {vr.error or 'vision model failed'}")
except Exception as exc:
message = f"slide {idx + 1}: {type(exc).__name__}: {str(exc)[:160]}"
slide_errors.append(message)
logger.warning(f"[PPTVision] slide {idx+1} check failed: {exc}")
result['success'] = result['slides_checked'] > 0
if not result['success'] and slide_errors:
result['error'] = ''.join(slide_errors[:3])
if slide_errors:
result['slide_errors'] = slide_errors
duration_ms = int((_time.monotonic() - t0) * 1000)
# Phase 38寫入 ppt_audit_results 留歷史(失敗安全)
try:
self._persist_audit_result(
pptx_path=pptx_path,
result=result,
avg_confidence=(sum(confidences) / len(confidences)) if confidences else 0.0,
duration_ms=duration_ms,
)
except Exception as e:
logger.warning(f"[PPTVision] persist audit result failed: {e}")
return result
def _persist_audit_result(self, pptx_path: str, result: Dict[str, Any],
avg_confidence: float, duration_ms: int) -> None:
"""Phase 38: 把每次 audit 結果寫入 ppt_audit_results 表。
失敗安全DB 寫入失敗只 log warning不擋主流程。
"""
import os
from datetime import datetime as _dt
from sqlalchemy import text as _sa_text
from database.manager import get_session
# 推論 audit_status
if result.get('error'):
err = result['error']
if 'libreoffice not installed' in err or 'PPT_VISION_ENABLED' in err:
status = 'skipped'
else:
status = 'error'
elif result.get('total_issues', 0) > 0:
status = 'failed'
elif result.get('success'):
status = 'passed'
else:
status = 'error'
# issues_found JSONB 序列化
import json as _json
issues_json = _json.dumps([
{'slide': slide_num, 'issues': issues}
for slide_num, issues in result.get('issues_by_slide', [])
], ensure_ascii=False)
try:
size_kb = round(os.path.getsize(pptx_path) / 1024, 1) if os.path.isfile(pptx_path) else None
mtime = _dt.fromtimestamp(os.path.getmtime(pptx_path)) if os.path.isfile(pptx_path) else None
except OSError:
size_kb = None
mtime = None
session = get_session()
try:
session.execute(
_sa_text("""
INSERT INTO ppt_audit_results
(pptx_filename, pptx_size_kb, pptx_mtime, vision_enabled,
audit_status, issues_count, issues_found, confidence,
duration_ms, error_msg)
VALUES
(:fname, :sz, :mt, :ve, :st, :ic, CAST(:if AS JSONB),
:cf, :du, :em)
"""),
{
'fname': os.path.basename(pptx_path),
'sz': size_kb,
'mt': mtime,
've': True, # 進到這裡代表 vision 已 enabled
'st': status,
'ic': result.get('total_issues', 0),
'if': issues_json,
'cf': round(avg_confidence, 3),
'du': duration_ms,
'em': result.get('error', None),
},
)
session.commit()
finally:
session.close()
def check_image(self, image_path: str) -> VisionResult:
"""檢查單張 PPT 截圖。
Args:
image_path: 本地檔案路徑jpg/png
Returns:
VisionResult.issues_found 含問題清單;無問題則空 list + confidence=1.0
"""
start = time.monotonic()
if not self.is_available():
return VisionResult(
success=False,
error='PPT_VISION_ENABLED=false (Phase 14 預設 OFF)',
)
if not os.path.isfile(image_path):
return VisionResult(
success=False,
error=f'image not found: {image_path}',
)
# 讀檔並 base64 編碼;可用時先壓縮縮圖,避免 Ollama vision 被大圖拖慢。
try:
img_b64 = self._encode_image_for_vision(image_path)
except Exception as e:
return VisionResult(
success=False,
error=f'read image failed: {type(e).__name__}: {str(e)[:200]}',
)
try:
from services.ollama_service import OllamaService
ollama = OllamaService(model=self.model)
resp = ollama.generate(
prompt='請檢查這張 momo 電商 PPT 截圖,找出視覺異常。',
model=self.model,
system_prompt=PPT_VISION_SYSTEM_PROMPT,
temperature=0.2,
timeout=PPT_VISION_TIMEOUT,
keep_alive='5m',
options={'num_predict': 256},
images=[img_b64],
)
duration_ms = int((time.monotonic() - start) * 1000)
if not resp.success:
return VisionResult(
success=False, duration_ms=duration_ms,
error=resp.error or 'ollama vision failed',
)
raw = (resp.content or '').strip()
# 解析輸出:每行一個 ⚠️ 開頭的視為 issue✅ 無視覺異常則空 list
issues = []
for line in raw.split('\n'):
line = line.strip()
if line.startswith('⚠️') or line.startswith('warning:') or line.startswith('警告'):
issues.append(line)
if '' in raw and '無視覺異常' in raw and not issues:
# 確認是 OK
return VisionResult(
success=True, issues_found=[],
confidence=1.0, raw_response=raw,
duration_ms=duration_ms,
)
return VisionResult(
success=True, issues_found=issues,
confidence=0.85 if issues else 0.5,
raw_response=raw,
duration_ms=duration_ms,
)
except Exception as e:
duration_ms = int((time.monotonic() - start) * 1000)
return VisionResult(
success=False, duration_ms=duration_ms,
error=f'{type(e).__name__}: {str(e)[:200]}',
)
# 全域單例
ppt_vision_service = PPTVisionService()
def audit_recent_ppts(reports_dir: str | None = None, hours: int = 24,
max_files: int = 10,
filenames: Sequence[str] | None = None) -> Dict[str, Any]:
"""Phase 26 整合 hook — 每日 22:00 cron 跑:掃 reports/ 當天新增 .pptx 跑視覺檢查。
Args:
reports_dir: PPT 輸出目錄,未提供時改用 REPORTS_DIR 環境變數
hours: 掃過去 N 小時內的檔
max_files: 一次最多查 N 個檔(避免一次跑太久)
Returns:
{
'audited_files': [...],
'total_issues': int,
'errors': [...],
}
"""
import os
import time
summary = {'audited_files': [], 'total_issues': 0, 'errors': []}
if reports_dir is None:
reports_dir = os.environ.get('REPORTS_DIR', '/app/data/reports')
if not is_ppt_vision_enabled():
summary['errors'].append('PPT_VISION_ENABLED=false')
return summary
if not os.path.isdir(reports_dir):
summary['errors'].append(f'{reports_dir} not found')
return summary
requested_names = {
os.path.basename(str(name))
for name in (filenames or [])
if str(name).lower().endswith('.pptx')
}
# 掃當天新增 .pptx若指定 filenames直接審指定檔不受 hours 視窗限制。
cutoff = time.time() - hours * 3600
pptx_files = []
for f in os.listdir(reports_dir):
if not f.lower().endswith('.pptx'):
continue
if requested_names and f not in requested_names:
continue
full = os.path.join(reports_dir, f)
try:
if requested_names or os.path.getmtime(full) >= cutoff:
pptx_files.append((os.path.getmtime(full), full))
except OSError:
continue
if requested_names:
found_names = {os.path.basename(path) for _mtime, path in pptx_files}
for missing in sorted(requested_names - found_names):
summary['errors'].append(f'{missing}: file not found')
pptx_files.sort(reverse=True)
pptx_files = pptx_files[:max_files]
svc = PPTVisionService()
for mtime, path in pptx_files:
try:
result = svc.check_ppt_file(path)
entry = {
'path': path,
'slides_checked': result.get('slides_checked', 0),
'issues': result.get('total_issues', 0),
'issues_by_slide': result.get('issues_by_slide', []),
'error': result.get('error'),
}
summary['audited_files'].append(entry)
summary['total_issues'] += entry['issues']
if entry['error']:
summary['errors'].append(f"{path}: {entry['error']}")
except Exception as exc:
summary['errors'].append(f'{path}: {type(exc).__name__}: {str(exc)[:150]}')
return summary
def start_ppt_vision_audit_background(
*,
reports_dir: str | None = None,
hours: int = 24,
max_files: int = 10,
filenames: Sequence[str] | None = None,
) -> Dict[str, Any]:
"""Queue a non-blocking PPT vision audit run for the admin UI."""
global _LAST_AUDIT_RUN
if _AUDIT_LOCK.locked():
return {
'ok': True,
'status': 'already_running',
'message': 'PPT vision audit is already running.',
'last_run': _LAST_AUDIT_RUN,
}
clean_filenames = [
os.path.basename(str(name))
for name in (filenames or [])
if str(name).lower().endswith('.pptx')
]
def _run():
global _LAST_AUDIT_RUN
with _AUDIT_LOCK:
started_at = time.strftime('%Y-%m-%d %H:%M:%S')
try:
summary = audit_recent_ppts(
reports_dir=reports_dir,
hours=hours,
max_files=max_files,
filenames=clean_filenames or None,
)
_LAST_AUDIT_RUN = {
'ok': True,
'status': 'completed',
'started_at': started_at,
'finished_at': time.strftime('%Y-%m-%d %H:%M:%S'),
'summary': summary,
}
except Exception as exc:
_LAST_AUDIT_RUN = {
'ok': False,
'status': 'error',
'started_at': started_at,
'finished_at': time.strftime('%Y-%m-%d %H:%M:%S'),
'error': f'{type(exc).__name__}: {str(exc)[:200]}',
}
logger.error("[PPTVision] background audit failed: %s", exc, exc_info=True)
thread = threading.Thread(target=_run, name='ppt-vision-audit', daemon=True)
thread.start()
return {
'ok': True,
'status': 'queued',
'message': 'PPT vision audit queued.',
'filenames': clean_filenames,
'max_files': max_files,
}
def push_ppt_audit_to_telegram(summary: Dict[str, Any]) -> bool:
"""有 issues 才推 Telegram避免靜默報「無問題」洗版"""
if summary['total_issues'] <= 0:
return False
try:
from services.telegram_templates import _send_telegram_raw
except Exception:
return False
lines = [f"🔍 <b>PPT 視覺審核({len(summary['audited_files'])} 份)</b>"]
lines.append('' * 18)
for entry in summary['audited_files']:
if entry['issues'] > 0:
fname = os.path.basename(entry['path']) if hasattr(__import__('os'), 'path') else entry['path']
import os as _os
fname = _os.path.basename(entry['path'])
lines.append(f"\n📊 <code>{fname}</code> ({entry['slides_checked']} slides, "
f"<b>{entry['issues']} issues</b>)")
for slide_num, issues in entry['issues_by_slide'][:3]: # 每檔最多列 3 張
for iss in issues[:2]: # 每張 slide 最多列 2 個 issue
lines.append(f" Slide {slide_num}: {iss[:120]}")
msg = '\n'.join(lines)
try:
_send_telegram_raw(msg)
return True
except Exception:
return False
__all__ = [
'PPTVisionService',
'VisionResult',
'ppt_vision_service',
'is_ppt_vision_enabled',
'get_ppt_vision_runtime_status',
'PPT_VISION_SYSTEM_PROMPT',
'audit_recent_ppts',
'start_ppt_vision_audit_background',
'push_ppt_audit_to_telegram',
]