image_tools.py

"""
وحدة تحليل الصور
- استلام صور من تيليجرام
- تحليلها عبر:
  1. Z.ai vision API (إذا كان ZAI_API_KEY متاح)
  2. Hugging Face Inference API (بديل مجاني)
"""
import base64
import logging
from typing import Optional

import httpx

from config import config

logger = logging.getLogger(__name__)


async def download_telegram_file(bot, file_id: str) -> bytes:
    """تحميل ملف من تيليجرام"""
    tg_file = await bot.get_file(file_id)
    # استخدام download_to_memory لتجنب الكتابة على القرص
    import io
    buf = io.BytesIO()
    await tg_file.download_to_memory(buf)
    return buf.getvalue()


async def analyze_image_with_zai(
    image_bytes: bytes,
    prompt: str = "صف هذه الصورة بالتفصيل.",
) -> str:
    """
    تحليل صورة عبر Z.ai vision API (glm-4v).
    يتطلب ZAI_API_KEY.
    """
    if not config.api_keys_for_use:
        return "❌ ZAI_API_KEY غير مضبوط. لا يمكن تحليل الصورة."

    b64 = base64.b64encode(image_bytes).decode("utf-8")
    url = f"{config.ZAI_API_BASE}/chat/completions"
    headers = {
        "Authorization": f"Bearer {config.api_keys_for_use[0]}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": "glm-4v-plus",
        "messages": [
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}},
                ],
            }
        ],
        "max_tokens": 1024,
    }

    try:
        async with httpx.AsyncClient(timeout=60) as client:
            resp = await client.post(url, headers=headers, json=payload)
            if resp.status_code != 200:
                # محاولة بنموذج بديل
                payload["model"] = "glm-4v"
                resp = await client.post(url, headers=headers, json=payload)
                if resp.status_code != 200:
                    return f"❌ فشل التحليل (HTTP {resp.status_code}): {resp.text[:200]}"
            data = resp.json()
            return data.get("choices", [{}])[0].get("message", {}).get("content", "").strip()
    except Exception as e:
        logger.error(f"Z.ai vision failed: {e}", exc_info=True)
        return f"❌ خطأ في التحليل: {e}"


async def analyze_image_with_hf(
    image_bytes: bytes,
    hf_token: str,
) -> str:
    """
    تحليل صورة عبر Hugging Face Inference API (مجاني).
    يستخدم نموذج captioning مثل Salesforce/blip2-base.
    """
    if not hf_token:
        return "❌ HF_TOKEN غير مضبوط."
    try:
        # تجربة نموذج captioning
        url = "https://api-inference.huggingface.co/models/Salesforce/blip-image-captioning-base"
        headers = {"Authorization": f"Bearer {hf_token}"}
        async with httpx.AsyncClient(timeout=60) as client:
            resp = await client.post(url, headers=headers, content=image_bytes)
            if resp.status_code != 200:
                return f"❌ HF inference فشل (HTTP {resp.status_code})"
            data = resp.json()
            if isinstance(data, list) and data:
                caption = data[0].get("generated_text", "")
                return f"**وصف الصورة:** {caption}"
            return f"```json\n{data}\n```"
    except Exception as e:
        logger.error(f"HF image analysis failed: {e}", exc_info=True)
        return f"❌ خطأ: {e}"


async def analyze_image(
    image_bytes: bytes,
    prompt: str = "صف هذه الصورة بالتفصيل.",
    hf_token: str = "",
) -> str:
    """تحليل صورة مع fallback تلقائي"""
    # جرّب Z.ai أولاً (أفضل جودة)
    if config.ZAI_API_KEY:
        result = await analyze_image_with_zai(image_bytes, prompt)
        if not result.startswith("❌"):
            return result
        logger.warning(f"Z.ai vision failed, trying HF: {result}")

    # fallback إلى HF
    if hf_token:
        return await analyze_image_with_hf(image_bytes, hf_token)

    return "❌ لا يمكن تحليل الصورة: ZAI_API_KEY غير متاح و HF_TOKEN غير متاح."