github-actions[bot]
Automated deployment from GitHub Actions: a9d5a91de63ad7224bafe6039fb2b7479388fbf4
0a8c81b | import os | |
| import sys | |
| import json | |
| import logging | |
| import uuid | |
| from sqlalchemy.orm import Session | |
| # Add apps/engine to sys.path so we can import src | |
| sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) | |
| from src.models.database import engine | |
| from src.models.product_idea import ProductIdea, ProductStatus | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| def main(): | |
| if len(sys.argv) < 2: | |
| print("Usage: python antigravity_insert.py '<JSON_ARRAY>'") | |
| sys.exit(1) | |
| json_str = sys.argv[1] | |
| try: | |
| ideas = json.loads(json_str) | |
| except json.JSONDecodeError as e: | |
| logger.error(f"Failed to parse JSON: {e}") | |
| sys.exit(1) | |
| if not isinstance(ideas, list): | |
| logger.error("Input must be a JSON array of objects.") | |
| sys.exit(1) | |
| inserted_count = 0 | |
| with Session(engine) as session: | |
| for idea_data in ideas: | |
| try: | |
| # Ensure UUID if missing | |
| idea_id = idea_data.get("id") | |
| if not idea_id: | |
| idea_id = str(uuid.uuid4()) | |
| source_url = idea_data.get("source_url") | |
| if source_url: | |
| import urllib.request | |
| try: | |
| req = urllib.request.Request(source_url, method="HEAD") | |
| req.add_header('User-Agent', 'Mozilla/5.0') | |
| with urllib.request.urlopen(req, timeout=5) as res: | |
| if res.status == 404: | |
| raise ValueError(f"URL returned 404 Not Found") | |
| except Exception as e: | |
| logger.error(f"❌ Sourcing Link Validation Failed for '{source_url}': {e}") | |
| raise ValueError(f"Invalid sourcing link: {e}") | |
| idea = ProductIdea( | |
| id=idea_id, | |
| shopify_title=idea_data.get("title"), | |
| est_total_cost_cny=idea_data.get("price_cny"), | |
| recommended_retail_price_usd=idea_data.get("retail_price_usd"), | |
| zone_1_shipping_usd=5.0, # default shipping cost | |
| source_url=source_url, | |
| primary_image_url=idea_data.get("primary_image_url"), | |
| status=ProductStatus.SCRAPED | |
| ) | |
| session.add(idea) | |
| inserted_count += 1 | |
| except Exception as e: | |
| logger.error(f"Failed to process idea {idea_data.get('title')}: {e}") | |
| # Propagate exception to fail the script if validation fails | |
| raise e | |
| session.commit() | |
| logger.info(f"Successfully inserted {inserted_count} products into Turso DB.") | |
| if __name__ == "__main__": | |
| main() | |