feat: add AI service (Claude + OpenAI) and image processor with EXIF geo-tagging
AIService wraps both Anthropic Claude and OpenAI APIs for product content generation. ImageProcessor handles EXIF geo-tagging with company info and GPS coordinates using piexif. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsinghpal
@@ -1,2 +1,3 @@
|
||||
# Library helpers will be imported here
|
||||
from .woo_api_client import WooApiClient
|
||||
from .ai_service import AIService
|
||||
from .image_processor import ImageProcessor
|
||||
|
||||
171
fusion-woo-odoo/fusion_woocommerce/lib/ai_service.py
Normal file
171
fusion-woo-odoo/fusion_woocommerce/lib/ai_service.py
Normal file
@@ -0,0 +1,171 @@
|
||||
import json
|
||||
import logging
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AIService:
|
||||
"""AI content generation service supporting Claude and OpenAI."""
|
||||
|
||||
def __init__(self, provider, api_key, model=None):
|
||||
"""
|
||||
Args:
|
||||
provider: 'claude' or 'openai'
|
||||
api_key: API key for the chosen provider
|
||||
model: Model name (defaults to claude-sonnet-4-5-20250514 or gpt-4o)
|
||||
"""
|
||||
self.provider = provider
|
||||
self.api_key = api_key
|
||||
if model:
|
||||
self.model = model
|
||||
else:
|
||||
self.model = 'claude-sonnet-4-5-20250514' if provider == 'claude' else 'gpt-4o'
|
||||
self._client = None
|
||||
|
||||
def _get_client(self):
|
||||
if self._client:
|
||||
return self._client
|
||||
if self.provider == 'claude':
|
||||
try:
|
||||
import anthropic
|
||||
self._client = anthropic.Anthropic(api_key=self.api_key)
|
||||
except ImportError:
|
||||
raise RuntimeError("anthropic package not installed. Run: pip install anthropic")
|
||||
elif self.provider == 'openai':
|
||||
try:
|
||||
import openai
|
||||
self._client = openai.OpenAI(api_key=self.api_key)
|
||||
except ImportError:
|
||||
raise RuntimeError("openai package not installed. Run: pip install openai")
|
||||
return self._client
|
||||
|
||||
def generate(self, system_prompt, user_message, max_tokens=2000):
|
||||
"""Generate text using the configured AI provider."""
|
||||
client = self._get_client()
|
||||
try:
|
||||
if self.provider == 'claude':
|
||||
response = client.messages.create(
|
||||
model=self.model,
|
||||
max_tokens=max_tokens,
|
||||
system=system_prompt,
|
||||
messages=[{"role": "user", "content": user_message}],
|
||||
)
|
||||
return response.content[0].text
|
||||
elif self.provider == 'openai':
|
||||
response = client.chat.completions.create(
|
||||
model=self.model,
|
||||
max_tokens=max_tokens,
|
||||
messages=[
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_message},
|
||||
],
|
||||
)
|
||||
return response.choices[0].message.content
|
||||
except Exception as e:
|
||||
_logger.error("AI generation failed (%s): %s", self.provider, str(e))
|
||||
raise
|
||||
|
||||
def generate_product_content(self, product_info, prompts):
|
||||
"""Generate all product content at once.
|
||||
|
||||
Args:
|
||||
product_info: dict with keys like name, category, features, raw_description
|
||||
prompts: dict with keys: title, short_desc, long_desc, meta_title, meta_desc, keywords
|
||||
|
||||
Returns:
|
||||
dict with generated content for each field
|
||||
"""
|
||||
context = json.dumps(product_info, indent=2)
|
||||
system = (
|
||||
"You are an expert e-commerce copywriter and SEO specialist. "
|
||||
"You create compelling, SEO-optimized product content for online stores. "
|
||||
"Always respond with valid JSON containing the requested fields. "
|
||||
"HTML descriptions should use proper semantic HTML tags."
|
||||
)
|
||||
|
||||
user_msg = f"""Based on this product information:
|
||||
{context}
|
||||
|
||||
Generate the following content as a JSON object with these exact keys:
|
||||
|
||||
1. "title": {prompts.get('title', 'SEO-optimized product title in Title Case')}
|
||||
2. "short_description": {prompts.get('short_desc', 'Compelling 2-3 sentence HTML summary')}
|
||||
3. "long_description": {prompts.get('long_desc', 'Detailed HTML product description with headings and lists')}
|
||||
4. "meta_title": {prompts.get('meta_title', 'SEO meta title under 60 characters')}
|
||||
5. "meta_description": {prompts.get('meta_desc', 'SEO meta description under 160 characters')}
|
||||
6. "keywords": {prompts.get('keywords', 'Comma-separated SEO keywords')}
|
||||
|
||||
Respond ONLY with the JSON object, no markdown formatting."""
|
||||
|
||||
try:
|
||||
raw = self.generate(system, user_msg, max_tokens=3000)
|
||||
# Try to parse JSON from the response
|
||||
# Strip any markdown code fences
|
||||
cleaned = raw.strip()
|
||||
if cleaned.startswith('```'):
|
||||
cleaned = cleaned.split('\n', 1)[1]
|
||||
if cleaned.endswith('```'):
|
||||
cleaned = cleaned[:-3]
|
||||
cleaned = cleaned.strip()
|
||||
return json.loads(cleaned)
|
||||
except json.JSONDecodeError:
|
||||
_logger.warning("AI returned non-JSON response, returning raw text")
|
||||
return {
|
||||
'title': raw[:200] if raw else '',
|
||||
'short_description': '',
|
||||
'long_description': raw or '',
|
||||
'meta_title': '',
|
||||
'meta_description': '',
|
||||
'keywords': '',
|
||||
}
|
||||
|
||||
def generate_single_field(self, product_info, prompt, field_name):
|
||||
"""Generate a single field using the given prompt."""
|
||||
context = json.dumps(product_info, indent=2)
|
||||
system = (
|
||||
"You are an expert e-commerce copywriter and SEO specialist. "
|
||||
"Respond with ONLY the requested content, no explanations or formatting."
|
||||
)
|
||||
user_msg = f"Product info:\n{context}\n\nTask: {prompt}"
|
||||
|
||||
result = self.generate(system, user_msg, max_tokens=1500)
|
||||
return result.strip() if result else ''
|
||||
|
||||
def generate_image_metadata(self, product_name, product_category, prompt_alt, prompt_caption):
|
||||
"""Generate SEO metadata for a product image.
|
||||
|
||||
Returns:
|
||||
dict with: alt_text, caption, title, description
|
||||
"""
|
||||
system = (
|
||||
"You are an SEO specialist for e-commerce product images. "
|
||||
"Generate metadata that helps with image SEO and accessibility. "
|
||||
"Respond ONLY with a JSON object."
|
||||
)
|
||||
user_msg = f"""Product: {product_name}
|
||||
Category: {product_category}
|
||||
|
||||
Generate image metadata as JSON:
|
||||
- "alt_text": {prompt_alt} (under 125 characters)
|
||||
- "caption": {prompt_caption}
|
||||
- "title": SEO-optimized image title
|
||||
- "description": Descriptive image text for SEO
|
||||
|
||||
Respond ONLY with the JSON object."""
|
||||
|
||||
try:
|
||||
raw = self.generate(system, user_msg, max_tokens=500)
|
||||
cleaned = raw.strip()
|
||||
if cleaned.startswith('```'):
|
||||
cleaned = cleaned.split('\n', 1)[1]
|
||||
if cleaned.endswith('```'):
|
||||
cleaned = cleaned[:-3]
|
||||
cleaned = cleaned.strip()
|
||||
return json.loads(cleaned)
|
||||
except (json.JSONDecodeError, Exception):
|
||||
return {
|
||||
'alt_text': product_name,
|
||||
'caption': product_name,
|
||||
'title': product_name,
|
||||
'description': product_name,
|
||||
}
|
||||
104
fusion-woo-odoo/fusion_woocommerce/lib/image_processor.py
Normal file
104
fusion-woo-odoo/fusion_woocommerce/lib/image_processor.py
Normal file
@@ -0,0 +1,104 @@
|
||||
import base64
|
||||
import io
|
||||
import logging
|
||||
import struct
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ImageProcessor:
|
||||
"""Process product images: EXIF geo-tagging, metadata, optimization."""
|
||||
|
||||
@staticmethod
|
||||
def geo_tag_image(image_b64, company_name, address, phone, lat, lng):
|
||||
"""Write EXIF metadata with company info and GPS coordinates.
|
||||
|
||||
Args:
|
||||
image_b64: base64-encoded image data
|
||||
company_name: company name for Copyright/Artist tags
|
||||
address: company address for ImageDescription
|
||||
phone: company phone
|
||||
lat: GPS latitude (float)
|
||||
lng: GPS longitude (float)
|
||||
|
||||
Returns:
|
||||
base64-encoded image with EXIF data
|
||||
"""
|
||||
try:
|
||||
import piexif
|
||||
except ImportError:
|
||||
_logger.warning("piexif not installed — skipping geo-tagging. Run: pip install piexif")
|
||||
return image_b64
|
||||
|
||||
try:
|
||||
image_data = base64.b64decode(image_b64)
|
||||
|
||||
# Check if it's a JPEG (piexif only works with JPEG)
|
||||
if not image_data[:2] == b'\xff\xd8':
|
||||
_logger.info("Image is not JPEG — skipping EXIF geo-tagging")
|
||||
return image_b64
|
||||
|
||||
# Try to load existing EXIF
|
||||
try:
|
||||
exif_dict = piexif.load(image_data)
|
||||
except Exception:
|
||||
exif_dict = {"0th": {}, "Exif": {}, "GPS": {}, "1st": {}}
|
||||
|
||||
# Set 0th IFD tags
|
||||
exif_dict["0th"][piexif.ImageIFD.ImageDescription] = address.encode('utf-8') if address else b''
|
||||
exif_dict["0th"][piexif.ImageIFD.Copyright] = (
|
||||
f"Copyright {company_name}".encode('utf-8') if company_name else b''
|
||||
)
|
||||
exif_dict["0th"][piexif.ImageIFD.Artist] = company_name.encode('utf-8') if company_name else b''
|
||||
|
||||
# Set GPS tags if coordinates provided
|
||||
if lat and lng:
|
||||
lat_deg = ImageProcessor._decimal_to_dms(abs(lat))
|
||||
lng_deg = ImageProcessor._decimal_to_dms(abs(lng))
|
||||
|
||||
exif_dict["GPS"] = {
|
||||
piexif.GPSIFD.GPSLatitudeRef: b'N' if lat >= 0 else b'S',
|
||||
piexif.GPSIFD.GPSLatitude: lat_deg,
|
||||
piexif.GPSIFD.GPSLongitudeRef: b'E' if lng >= 0 else b'W',
|
||||
piexif.GPSIFD.GPSLongitude: lng_deg,
|
||||
}
|
||||
|
||||
exif_bytes = piexif.dump(exif_dict)
|
||||
output = io.BytesIO()
|
||||
piexif.insert(exif_bytes, image_data, output)
|
||||
return base64.b64encode(output.getvalue()).decode('utf-8')
|
||||
|
||||
except Exception as e:
|
||||
_logger.error("Failed to geo-tag image: %s", str(e))
|
||||
return image_b64
|
||||
|
||||
@staticmethod
|
||||
def _decimal_to_dms(decimal):
|
||||
"""Convert decimal degrees to EXIF DMS format (degrees, minutes, seconds as rationals)."""
|
||||
degrees = int(decimal)
|
||||
minutes_float = (decimal - degrees) * 60
|
||||
minutes = int(minutes_float)
|
||||
seconds = int((minutes_float - minutes) * 60 * 10000)
|
||||
return (
|
||||
(degrees, 1),
|
||||
(minutes, 1),
|
||||
(seconds, 10000),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def prepare_wc_image(image_b64, filename, alt_text='', caption='', title='', description=''):
|
||||
"""Prepare image data for WooCommerce upload.
|
||||
|
||||
Returns dict ready for WC product images array, using src as base64 data URL.
|
||||
Note: WC REST API v3 accepts image URLs in 'src'. For base64, we need to upload
|
||||
via WordPress media endpoint first, then reference by URL.
|
||||
"""
|
||||
return {
|
||||
'name': title or filename,
|
||||
'alt': alt_text or '',
|
||||
'caption': caption or '',
|
||||
'description': description or '',
|
||||
# The actual upload will be handled by the wizard
|
||||
'_base64': image_b64,
|
||||
'_filename': filename,
|
||||
}
|
||||
Reference in New Issue
Block a user