github-actions[bot]
Automated deployment from GitHub Actions: a9d5a91de63ad7224bafe6039fb2b7479388fbf4
0a8c81b | import logging | |
| logger = logging.getLogger(__name__) | |
| import os | |
| import asyncio | |
| from openai import OpenAI | |
| from pydantic import BaseModel | |
| from src.models.product_idea import ProductIdea, ProductStatus | |
| from src.models.marketing_content import MarketingContent | |
| from src.models.database import get_session | |
| from sqlmodel import select | |
| from mcp.client.stdio import stdio_client, StdioServerParameters | |
| from mcp.client.session import ClientSession | |
| import urllib.request | |
| import urllib.error | |
| import urllib.parse | |
| import json | |
| import base64 | |
| import re | |
| def verify_and_fallback_image(image_url: str, title: str) -> str: | |
| """Verifies if the image URL is accessible, else falls back to free AI generation.""" | |
| if not image_url: | |
| return "" | |
| try: | |
| req = urllib.request.Request(image_url, method="HEAD") | |
| # Pretend to be a browser to avoid some basic 403 blocks | |
| req.add_header('User-Agent', 'Mozilla/5.0') | |
| with urllib.request.urlopen(req, timeout=5) as res: | |
| if res.status == 200: | |
| # If image exists and is valid, return original | |
| return image_url | |
| except Exception: | |
| pass | |
| # Image is dead, blocked, or timeout. Fallback to free AI gen | |
| # Remove special characters to prevent Pollinations 500 errors | |
| clean_title = re.sub(r'[^a-zA-Z0-9\s]', '', title).strip() | |
| safe_title = urllib.parse.quote(f"A high quality studio commercial product photo of {clean_title}, white background, crisp lighting") | |
| return f"https://image.pollinations.ai/prompt/{safe_title}?width=1024&height=1024&nologo=True" | |
| async def ensure_smart_collection(collection_name: str, token: str, domain: str): | |
| if not collection_name: | |
| return | |
| logger.info(f"📂 [Store Curator] Checking if Smart Collection '{collection_name}' exists...") | |
| url = f"https://{domain}/admin/api/2024-04/smart_collections.json" | |
| # 1. Fetch existing | |
| req = urllib.request.Request(url, headers={"X-Shopify-Access-Token": token}) | |
| try: | |
| with urllib.request.urlopen(req) as response: | |
| data = json.loads(response.read().decode()) | |
| existing = [c for c in data.get("smart_collections", []) if c.get("title") == collection_name] | |
| if existing: | |
| logger.info(f"✅ [Store Curator] Collection '{collection_name}' already exists.") | |
| return existing[0]["id"] | |
| except Exception as e: | |
| logger.error(f"Failed to fetch collections: {e}") | |
| # 2. Create new if not exists | |
| logger.warning(f"✨ [Store Curator] Creating new Smart Collection: '{collection_name}'") | |
| payload = { | |
| "smart_collection": { | |
| "title": collection_name, | |
| "rules": [ | |
| { | |
| "column": "tag", | |
| "relation": "equals", | |
| "condition": collection_name | |
| } | |
| ] | |
| } | |
| } | |
| req_post = urllib.request.Request( | |
| url, | |
| data=json.dumps(payload).encode("utf-8"), | |
| headers={"X-Shopify-Access-Token": token, "Content-Type": "application/json"}, | |
| method="POST" | |
| ) | |
| try: | |
| with urllib.request.urlopen(req_post) as response: | |
| res = json.loads(response.read().decode()) | |
| logger.info(f"🎉 [Store Curator] Successfully created Collection '{collection_name}' (ID: {res['smart_collection']['id']})") | |
| return res['smart_collection']['id'] | |
| except urllib.error.HTTPError as e: | |
| logger.error(f"❌ Failed to create collection: {e.read().decode()}") | |
| async def push_to_shopify_mcp(content: MarketingContent): | |
| logger.info("🔌 [MCP] Booting Community Shopify MCP Server (geli2001/shopify-mcp)...") | |
| token = str(os.environ.get("SHOPIFY_ADMIN_ACCESS_TOKEN", "")) | |
| domain = str(os.environ.get("SHOPIFY_STORE_DOMAIN", "")) | |
| # 🌟 Store Curator Logic: Auto-create Smart Collection | |
| if content.collection_name: | |
| await ensure_smart_collection(content.collection_name, token, domain) | |
| server_params = StdioServerParameters( | |
| command="npx", | |
| args=[ | |
| "-y", "shopify-mcp", | |
| "--accessToken", token, | |
| "--domain", domain | |
| ] | |
| ) | |
| try: | |
| async def run_mcp(): | |
| async with stdio_client(server_params) as (read, write): | |
| async with ClientSession(read, write) as session: | |
| await session.initialize() | |
| logger.warning(f"🛠️ [MCP] Connected! Invoking Agent Tool 'create-product' for: {content.title}") | |
| # Invoke the MCP tool natively | |
| # ⚠️ We currently FORCE an exception here to gracefully degrade to the direct REST API | |
| # because the community shopify-mcp create-product tool does not easily support `tags` (for Smart Collections) and `images`. | |
| raise Exception("Bypassing MCP to use Direct REST API for full Tags & Image support") | |
| result = await asyncio.wait_for(run_mcp(), timeout=15.0) | |
| return result | |
| except Exception as e: | |
| logger.warning(f"⚠️ Bypassing MCP: {str(e)}") | |
| logger.info("🔄 [MCP FALLBACK] Agent gracefully degrading to direct Shopify Admin API...") | |
| import urllib.request | |
| import json | |
| domain = os.environ.get("SHOPIFY_STORE_DOMAIN") | |
| token = os.environ.get("SHOPIFY_ADMIN_ACCESS_TOKEN") | |
| url = f"https://{domain}/admin/api/2024-04/products.json" | |
| payload = { | |
| "product": { | |
| "title": content.title, | |
| "body_html": content.description_html, | |
| "product_type": content.product_type, | |
| "tags": content.collection_name, # The tag linking it to the Smart Collection | |
| "variants": [{"price": str(content.recommended_price_usd), "sku": content.product_idea_id}], | |
| "status": "active", | |
| "published": True, | |
| "published_scope": "web" | |
| } | |
| } | |
| if content.primary_image_url: | |
| valid_image_url = verify_and_fallback_image(content.primary_image_url, content.title) | |
| if valid_image_url: | |
| if "pollinations.ai" in valid_image_url: | |
| # Download and base64 encode to avoid Shopify 422 fetch errors | |
| try: | |
| req_img = urllib.request.Request(valid_image_url, headers={'User-Agent': 'Mozilla/5.0'}) | |
| with urllib.request.urlopen(req_img, timeout=20) as res_img: | |
| b64_data = base64.b64encode(res_img.read()).decode('utf-8') | |
| payload["product"]["images"] = [{"attachment": b64_data}] | |
| except Exception as e: | |
| logger.error(f"Failed to fetch AI image: {e}") | |
| else: | |
| payload["product"]["images"] = [{"src": valid_image_url}] | |
| req = urllib.request.Request( | |
| url, | |
| data=json.dumps(payload).encode("utf-8"), | |
| headers={"X-Shopify-Access-Token": token, "Content-Type": "application/json"}, | |
| method="POST" | |
| ) | |
| with urllib.request.urlopen(req) as response: | |
| res = json.loads(response.read().decode()) | |
| logger.info(f"✅ [DIRECT API FALLBACK] Successfully Created Product ID: {res['product']['id']}") | |
| return res | |
| async def generate_marketing_content(idea: ProductIdea) -> dict: | |
| # Mock fallback for MVP testing without API Key | |
| if not os.environ.get("OPENAI_API_KEY"): | |
| logger.warning("⚠️ OPENAI_API_KEY not found. Using Mock Content Agent output.") | |
| mock_data = MarketingContent( | |
| product_idea_id=idea.id, | |
| title="Ultra-Premium 1688 Imported Gadget", | |
| description_html="<p>Experience the ultimate quality with this sourced gadget.</p><p>Built with premium materials.</p><p>Order today for fast shipping!</p>", | |
| seo_keywords=["premium gadget", "imported tech", "quality tech", "dropshipping", "1688 import"], | |
| recommended_price_usd=idea.recommended_retail_price_usd, | |
| product_type="Gadget", | |
| collection_name="Smart Home & Electronics", | |
| primary_image_url=idea.primary_image_url or "https://cbu01.alicdn.com/img/ibank/O1CN01fN1pC6228sM5kF2s3_!!2214643033501-0-cib.jpg", | |
| language="en" | |
| ) | |
| shopify_res = await push_to_shopify_mcp(mock_data) | |
| result = mock_data.model_dump() | |
| if shopify_res and "product" in shopify_res: | |
| result["shopify_product_id"] = str(shopify_res["product"].get("id", "")) | |
| handle = shopify_res["product"].get("handle", "") | |
| domain = os.environ.get("SHOPIFY_STORE_DOMAIN", "vlta-dropshop.myshopify.com") | |
| result["shopify_product_url"] = f"https://{domain}/products/{handle}" if handle else "" | |
| return result | |
| # Initialize OpenAI (DeepSeek) client | |
| client = OpenAI( | |
| api_key=os.environ.get("OPENAI_API_KEY"), | |
| base_url=os.environ.get("OPENAI_BASE_URL", "https://api.qnaigc.com/v1") | |
| ) | |
| prompt = f""" | |
| You are an expert e-commerce copywriter. | |
| I have a product sourced from the following URL: {idea.source_url} | |
| Pricing information: | |
| - Target Retail Price: ${idea.recommended_retail_price_usd} | |
| - Estimated Cost: ¥{idea.est_total_cost_cny} | |
| Generate a highly converting product listing for a Shopify store. | |
| You MUST determine the best broad storefront category for this item (e.g. "Pet Supplies", "Kitchen & Dining") and set it as `collection_name`. | |
| You MUST also determine a specific product type (e.g. "Dog Bed") and set it as `product_type`. | |
| You MUST respond with valid JSON matching this schema: | |
| {MarketingContent.model_json_schema()} | |
| """ | |
| response = client.chat.completions.create( | |
| model=os.environ.get("DEEPSEEK_MODEL", "deepseek/deepseek-v4-pro-202606"), | |
| messages=[{"role": "user", "content": prompt}], | |
| response_format={"type": "json_object"}, | |
| temperature=0.7, | |
| ) | |
| try: | |
| content_json = response.choices[0].message.content | |
| content = MarketingContent.model_validate_json(content_json) | |
| # 🎨 Inject Image from Idea or use a high-quality fallback | |
| if idea.primary_image_url: | |
| content.primary_image_url = idea.primary_image_url | |
| else: | |
| # High-quality e-commerce placeholder | |
| content.primary_image_url = "https://images.unsplash.com/photo-1523275335684-37898b6baf30?q=80&w=1000&auto=format&fit=crop" | |
| # 🔌 AGENT-NATIVE MCP PUSH | |
| shopify_res = await push_to_shopify_mcp(content) | |
| result = content.model_dump() | |
| if shopify_res and "product" in shopify_res: | |
| result["shopify_product_id"] = str(shopify_res["product"].get("id", "")) | |
| handle = shopify_res["product"].get("handle", "") | |
| domain = os.environ.get("SHOPIFY_STORE_DOMAIN", "vlta-dropshop.myshopify.com") | |
| result["shopify_product_url"] = f"https://{domain}/products/{handle}" if handle else "" | |
| return result | |
| except Exception as e: | |
| logger.error(f"❌ Failed to parse AI response: {e}") | |
| raise e | |
| def run_content_agent(product_id: str): | |
| session = next(get_session()) | |
| idea = session.exec(select(ProductIdea).where(ProductIdea.id == product_id)).first() | |
| if not idea: | |
| logger.error(f"❌ ProductIdea {product_id} not found in DB.") | |
| return | |
| logger.info(f"🚀 Content Agent processing Product: {product_id}") | |
| # Run async function | |
| loop = asyncio.get_event_loop() | |
| result = loop.run_until_complete(generate_marketing_content(idea)) | |
| idea.status = ProductStatus.PUBLISHED | |
| session.add(idea) | |
| session.commit() | |
| logger.info(f"✅ Product {product_id} marked as PUBLISHED in DB.") | |
| if __name__ == "__main__": | |
| import argparse | |
| from dotenv import load_dotenv | |
| load_dotenv("../../.env") | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("product_id") | |
| args = parser.parse_args() | |
| run_content_agent(args.product_id) | |