image_check.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import sys
import base64
import requests
import json
import mimetypes

def get_server_ip():
    ip = os.getenv("server_ip")
    if ip:
        return ip
    try:
        env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), '.env')
        if os.path.exists(env_path):
            with open(env_path, 'r', encoding='utf-8') as f:
                for line in f:
                    line = line.strip()
                    if line and not line.startswith('#') and '=' in line:
                        parts = line.split('=', 1)
                        key = parts[0].strip()
                        val = parts[1].strip().strip('\'"')
                        if key == "server_ip":
                            return val
    except Exception:
        pass
    return "10.112.5.79"

def check_image(image_path_or_url="/home/ivan/eagle/cat.jpg", prompt_text="照片裡面有什麼？請遵照以下格式做輸出 {'danger':0, 'reason':text_resopn}，如果是危險生物的話，danger 輸出為1，反之輸出為0，並在reason說明原因"):
    server_ip = get_server_ip()
    url = f"http://{server_ip}:4000/v1/chat/completions"
    
    headers = {
        "Content-Type": "application/json"
    }
    
    resolved_image_url = image_path_or_url
    is_local = False
    
    # Check if the path exists locally on this machine
    if os.path.exists(image_path_or_url):
        is_local = True
        try:
            # Guess MIME type (e.g. image/jpeg, image/png)
            mime_type, _ = mimetypes.guess_type(image_path_or_url)
            if not mime_type:
                mime_type = "image/jpeg"
                
            # Read local file and encode to base64
            with open(image_path_or_url, "rb") as image_file:
                base64_data = base64.b64encode(image_file.read()).decode("utf-8")
            
            resolved_image_url = f"data:{mime_type};base64,{base64_data}"
        except Exception as e:
            print(f"Warning: Failed to read local image file '{image_path_or_url}': {e}")
            print("Sending path as-is instead.")
            
    payload = {
        "model": "openbmb/MiniCPM-V-4.6",
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": resolved_image_url
                        }
                    },
                    {
                        "type": "text",
                        "text": prompt_text
                    }
                ]
            }
        ]
    }
    
    print(f"Sending request to {url}...")
    if is_local:
        print(f"Image Source: {image_path_or_url} (local file, encoded in base64)")
    else:
        print(f"Image Source: {image_path_or_url} (remote/server path)")
    print(f"Prompt: {prompt_text}\n")
    
    try:
        response = requests.post(url, headers=headers, json=payload, timeout=60)
        response.raise_for_status()  # Check for HTTP errors
        
        # Parse JSON response
        result = response.json()
        
        # Print raw response metadata
        print("=== Raw Response ===")
        print(json.dumps(result, indent=2, ensure_ascii=False))
        print("====================\n")
        
        # Extract assistant message content
        choices = result.get("choices", [])
        if choices:
            content = choices[0].get("message", {}).get("content", "")
            print("=== Extracted Content ===")
            print(content)
            print("=========================")
            return content
        else:
            print("Error: No choices returned in the response.")
            return None
            
    except requests.exceptions.Timeout:
        print("Error: Request timed out. Make sure the API server is responsive.")
        return None
    except requests.exceptions.ConnectionError:
        print(f"Error: Connection failed. Please check if the API server at {server_ip}:4000 is running.")
        return None
    except requests.exceptions.HTTPError as http_err:
        print(f"HTTP Error occurred: {http_err}")
        try:
            print("Response body:")
            print(response.text)
        except Exception:
            pass
        return None
    except Exception as err:
        print(f"An unexpected error occurred: {err}")
        return None

if __name__ == "__main__":
    # If users provide arguments, they can customize the image and prompt.
    # E.g., python3 image_check.py "picture.jpeg" "照片裡有什麼？"
    img_arg = sys.argv[1] if len(sys.argv) > 1 else "/home/ivan/eagle/cat.jpg"
    prompt_arg = sys.argv[2] if len(sys.argv) > 2 else "照片裡面有什麼？請遵照以下格式做輸出 {'danger':0, 'reason':text_resopn}，如果是危險生物的話，danger 輸出為1，反之輸出為0，並在reason說明原因"
    
    check_image(image_path_or_url=img_arg, prompt_text=prompt_arg)