"""Generate blog images (staged locally) and upload to Shopify at publish.""" import logging from blogs.models import BlogArticle, BlogArticleImage, BlogGenerationLog, BlogSettings from blogs.services.blog_image_storage import BlogImageStorage, is_shopify_cdn_url from blogs.services.openai_image_client import OpenAIImageClient from blogs.services.shopify_file_service import ShopifyFileAccessError, ShopifyFileService logger = logging.getLogger(__name__) class BlogImageService: """Orchestrates image generation, local staging, and Shopify CDN upload.""" @classmethod def generate_for_article(cls, article_id: int) -> BlogArticle: article = BlogArticle.objects.get(pk=article_id) settings_obj = BlogSettings.load() if not settings_obj.include_images: return article pending = list( article.images.filter(generation_status="pending").order_by("position") ) if not pending: return article BlogGenerationLog.objects.create( article=article, step="image_generation", status="started", details={"count": len(pending)}, ) niche = (settings_obj.description or "")[:100] ok, fail = 0, 0 featured_preview = "" for img in pending: try: result = cls._generate_for_slot(img, article, niche) except Exception: logger.exception( "image failed article=%s pos=%s", article_id, img.position ) img.generation_status = "failed" img.save(update_fields=["generation_status"]) fail += 1 continue img.image_url = result["preview_url"] img.local_file = result["local_file"] img.prompt_used = result.get("prompt_used", "") img.generation_status = "completed" img.save( update_fields=[ "image_url", "local_file", "prompt_used", "generation_status", ] ) ok += 1 if img.position == "header": featured_preview = img.image_url if featured_preview: article.featured_image_url = featured_preview article.save(update_fields=["featured_image_url", "updated_at"]) cls._inject_inline_images_into_content(article) BlogGenerationLog.objects.create( article=article, step="image_generation", status="completed", details={"ok": ok, "failed": fail}, ) return article @classmethod def upload_staged_images_to_shopify(cls, article: BlogArticle) -> BlogArticle: """Upload staged images to Shopify Files; rewrite inline URLs in body only.""" images = list( article.images.filter(generation_status="completed").order_by("position") ) if not any(img.local_file for img in images): return article content = article.content or "" featured_url = article.featured_image_url uploaded = 0 use_attachment_fallback = False for img in images: if img.position == "header": if not img.local_file: if is_shopify_cdn_url(img.image_url): featured_url = img.image_url continue preview_url = img.image_url image_bytes = BlogImageStorage.read_bytes(img.local_file) filename = BlogImageStorage.resolve_path(img.local_file).name if use_attachment_fallback: featured_url = preview_url continue try: shopify_url = ShopifyFileService.upload_image( image_bytes, filename=filename, alt=img.alt_text or article.title, ) img.image_url = shopify_url local_file = img.local_file img.local_file = "" img.save(update_fields=["image_url", "local_file"]) BlogImageStorage.delete(local_file) featured_url = shopify_url uploaded += 1 except ShopifyFileAccessError: if not use_attachment_fallback: logger.warning( "Shopify file upload denied for article=%s (reinstall OAuth with " "write_files). Featured image will use REST attachment at publish.", article.pk, ) use_attachment_fallback = True featured_url = preview_url continue if not img.local_file: continue if is_shopify_cdn_url(img.image_url): continue preview_url = img.image_url image_bytes = BlogImageStorage.read_bytes(img.local_file) filename = BlogImageStorage.resolve_path(img.local_file).name if use_attachment_fallback: continue try: shopify_url = ShopifyFileService.upload_image( image_bytes, filename=filename, alt=img.alt_text or article.title, ) except ShopifyFileAccessError: if not use_attachment_fallback: logger.warning( "Shopify file upload denied for article=%s (reinstall OAuth with " "write_files). Inline images will use preview URLs.", article.pk, ) use_attachment_fallback = True continue if preview_url and preview_url in content: content = content.replace(preview_url, shopify_url) img.image_url = shopify_url local_file = img.local_file img.local_file = "" img.save(update_fields=["image_url", "local_file"]) BlogImageStorage.delete(local_file) uploaded += 1 if uploaded: article.content = content article.featured_image_url = featured_url article.save( update_fields=["content", "featured_image_url", "updated_at"] ) BlogGenerationLog.objects.create( article=article, step="shopify_image_upload", status="completed", details={"uploaded": uploaded}, ) return article @classmethod def _generate_for_slot( cls, img: BlogArticleImage, article: BlogArticle, niche: str, ) -> dict[str, str]: if img.position == "header": result = OpenAIImageClient.generate_header(article.title, niche) else: result = OpenAIImageClient.generate_inline(img.alt_text or article.title) local_file = BlogImageStorage.save( article.pk, img.position, result["image_bytes"] ) return { "local_file": local_file, "preview_url": BlogImageStorage.preview_url(local_file), "prompt_used": result.get("prompt_used", ""), } @staticmethod def _img_tag(image_url: str, alt: str) -> str: return ( f'{alt}' ) @classmethod def _inject_inline_images_into_content(cls, article: BlogArticle) -> None: """Replace inline image placeholders only (header is featured_image_url, not body).""" content = article.content or "" changed = False for img in article.images.filter( generation_status="completed" ).exclude(position="header").order_by("position"): if not img.image_url: continue placeholder = f"" tag = cls._img_tag(img.image_url, img.alt_text or "") if placeholder in content: content = content.replace(placeholder, tag) changed = True if changed: article.content = content article.save(update_fields=["content", "updated_at"])