# import os # import mysql.connector # from dotenv import load_dotenv # # ---------- Load Environment ---------- # load_dotenv() # DB_HOST = os.getenv("DB_HOST", "localhost") # DB_PORT = int(os.getenv("DB_PORT", 3306)) # DB_USER = os.getenv("DB_USER", "root") # DB_PASSWORD = os.getenv("DB_PASSWORD", "") # DB_NAME = os.getenv("DB_NAME", "tattoo_project") # BASE_URL = os.getenv("CLOUD_FRONT_URL", "https://d2p2o2gkzhif4m.cloudfront.net") # # ---------- Helpers ---------- # def build_category_urls(slug: str) -> tuple[str, str]: # icon_url = f"https://{BASE_URL}/categories/{slug}/icon_{slug}.png" # thumbnail_url = f"https://{BASE_URL}/categories/{slug}/{slug}_thumbnail.png" # return icon_url, thumbnail_url # def build_style_url(category_slug: str, style_slug: str) -> str: # return f"https://{BASE_URL}/categories/{category_slug}/style/{style_slug}.png" # def slug_to_name(slug: str) -> str: # return slug.replace("_", " ").title() # # ---------- Update Categories ---------- # def update_category_urls(conn): # cursor = conn.cursor(dictionary=True) # cursor.execute("SELECT category_id, slug FROM categories") # categories = cursor.fetchall() # for cat in categories: # slug = cat["slug"] # category_id = cat["category_id"] # icon_url, thumbnail_url = build_category_urls(slug) # cursor.execute(""" # UPDATE categories # SET icon_url = %s, # thumbnail_url = %s # WHERE category_id = %s # """, (icon_url, thumbnail_url, category_id)) # print(f"āœ… Updated category {slug}: {icon_url}, {thumbnail_url}") # conn.commit() # cursor.close() # print("šŸŽ‰ All categories updated successfully!") # # ---------- Update Tattoo Styles from Text File ---------- # def update_styles_from_file(conn, file_path: str): # cursor = conn.cursor(dictionary=True) # with open(file_path, "r", encoding="utf-8") as f: # lines = [line.strip() for line in f if line.strip()] # current_category = None # category_id = None # sort_order = 1 # for line in lines: # if line[0].isdigit() and "." in line: # Category header like "1. traditional" # current_category = line.split(".", 1)[1].strip() # sort_order = 1 # # Get category_id from DB # cursor.execute("SELECT category_id FROM categories WHERE slug = %s", (current_category,)) # row = cursor.fetchone() # if not row: # print(f"āš ļø Category '{current_category}' not found in DB, skipping...") # current_category = None # continue # category_id = row["category_id"] # # Optional: clear existing styles for this category to avoid mismatch # cursor.execute("DELETE FROM tattoo_styles WHERE category_id = %s", (category_id,)) # print(f"šŸ—‘ļø Cleared existing styles for category '{current_category}'") # print(f"\nšŸ“‚ Processing category: {current_category} (ID: {category_id})") # elif current_category: # Style slug # style_slug = line # style_name = slug_to_name(style_slug) # thumb_url = build_style_url(current_category, style_slug) # cursor.execute(""" # INSERT INTO tattoo_styles (category_id, name, slug, thumbnail_url, sort_order, is_active) # VALUES (%s, %s, %s, %s, %s, 1) # ON DUPLICATE KEY UPDATE # name = VALUES(name), # thumbnail_url = VALUES(thumbnail_url), # sort_order = VALUES(sort_order), # updated_at = CURRENT_TIMESTAMP # """, (category_id, style_name, style_slug, thumb_url, sort_order)) # print(f" āœ… Added/Updated style: {style_name} ({style_slug}) [Order {sort_order}]") # sort_order += 1 # conn.commit() # cursor.close() # print("\nšŸŽ‰ All tattoo styles updated successfully!") # # ---------- Main ---------- # def main(): # conn = mysql.connector.connect( # host=DB_HOST, # port=DB_PORT, # user=DB_USER, # password=DB_PASSWORD, # database=DB_NAME, # ) # # 1. Update categories # update_category_urls(conn) # # 2. Update tattoo styles from text file # update_styles_from_file(conn, "category_slug_s3.txt") # conn.close() # if __name__ == "__main__": # main() #========WORKING FOR S3============================ # import os # import boto3 # from io import BytesIO # from botocore.exceptions import NoCredentialsError # from dotenv import load_dotenv # from app.core.config import settings # # ---------- CONFIG ---------- # LOCAL_BASE_DIR = "Tattoo_categories" # BUCKET_NAME = settings.CLOUDINARY_BUCKET_NAME # BASE_URL = settings.CLOUD_FRONT_URL # # ---------- AWS CLIENT ---------- # s3_client = boto3.client( # "s3", # aws_access_key_id=settings.CLOUDINARY_API_KEY, # aws_secret_access_key=settings.CLOUDINARY_API_SECRET, # ) # def upload_file(local_path, bucket, s3_key): # """ # Uploads a single file to S3 using upload_fileobj # """ # try: # with open(local_path, "rb") as f: # image_bytes = BytesIO(f.read()) # s3_client.upload_fileobj( # image_bytes, # bucket, # s3_key, # ExtraArgs={'ContentType': 'image/png'} # adjust if other formats # ) # print(f"āœ… Uploaded {local_path} ā†’ s3://{bucket}/{s3_key}") # except Exception as e: # print(f"āŒ Failed to upload {local_path}: {e}") # def upload_directory(local_dir, bucket, s3_prefix): # """ # Uploads category folders to S3. # Each category contains thumbnail.png, icon.png, and style images (inside subfolders). # """ # for category_name in os.listdir(local_dir): # category_path = os.path.join(local_dir, category_name) # if not os.path.isdir(category_path): # continue # only process folders # # Upload icon.png and thumbnail.png directly under category # for special_file in ["icon.png", "thumbnail.png"]: # special_path = os.path.join(category_path, special_file) # if os.path.exists(special_path): # s3_key = f"{s3_prefix}/{category_name}/{special_file}" # # print('=====s3_key=========',s3_key) # upload_file(special_path, bucket, s3_key) # # Upload style images inside their own subfolder # for file in os.listdir(category_path): # if file in ["icon.png", "thumbnail.png"]: # continue # already uploaded # if file.lower().endswith((".png", ".jpg", ".jpeg", ".webp")): # style_name, _ = os.path.splitext(file) # remove extension # s3_key = f"{s3_prefix}/{category_name}/style/{file}" # local_path = os.path.join(category_path, file) # # print('===local_path==========',local_path) # # print('==============',s3_key) # upload_file(local_path, bucket, s3_key) # if __name__ == "__main__": # upload_directory(LOCAL_BASE_DIR, BUCKET_NAME, "categories") # print("šŸŽ‰ All uploads finished!") #======================================================================================================= # import os # import boto3 # import mysql.connector # from io import BytesIO # from dotenv import load_dotenv # from datetime import datetime # load_dotenv() # # ---------- CONFIG ---------- # LOCAL_BASE_DIR = "Tattoo" # BUCKET_NAME = os.getenv("CLOUDINARY_BUCKET_NAME") # BASE_URL = os.getenv("CLOUD_FRONT_URL") # # ---------- AWS CLIENT ---------- # s3_client = boto3.client( # "s3", # aws_access_key_id=os.getenv("CLOUDINARY_API_KEY"), # aws_secret_access_key=os.getenv("CLOUDINARY_API_SECRET"), # ) # # ---------- MYSQL CONFIG ---------- # DB_HOST = os.getenv("DB_HOST", "localhost") # DB_PORT = int(os.getenv("DB_PORT", 3306)) # DB_USER = os.getenv("DB_USER", "root") # DB_PASSWORD = os.getenv("DB_PASSWORD", "password") # DB_NAME = os.getenv("DB_NAME", "zyBLyukn9LGSC8b9") # now = datetime.now().strftime('%Y-%m-%d %H:%M:%S') # def get_db_connection(): # return mysql.connector.connect( # host=DB_HOST, # port=DB_PORT, # user=DB_USER, # password=DB_PASSWORD, # database=DB_NAME # ) # def update_category_urls(category_name, icon_url, thumbnail_url): # """ # Update categories table with icon and thumbnail URLs where slug = category_name # """ # try: # print('=====icon_url======', icon_url) # print('=====thumbnail_url=====', thumbnail_url) # conn = get_db_connection() # cursor = conn.cursor() # query = """ # UPDATE categories # SET icon_url = %s, thumbnail_url = %s # WHERE slug = %s # """ # cursor.execute(query, (icon_url, thumbnail_url, category_name)) # conn.commit() # print(f"šŸ“ Updated DB for category '{category_name}'") # cursor.close() # conn.close() # except Exception as e: # print(f"āŒ DB update failed for {category_name}: {e}") # def update_tattoo_style(category_name, file, s3_prefix): # """ # Insert or update tattoo_styles thumbnail_url using category_id + slug # """ # try: # conn = get_db_connection() # cursor = conn.cursor() # # Get category_id # cursor.execute("SELECT category_id FROM categories WHERE slug = %s", (category_name,)) # row = cursor.fetchone() # if not row: # print(f"āŒ No category found for slug={category_name}") # return # category_id = row[0] # # Extract slug from filename # style_slug, _ = os.path.splitext(file) # # Construct thumbnail URL # thumbnail_url = f"https://{BASE_URL}/{s3_prefix}/{category_name}/style/{file}" # # Check if row exists # cursor.execute( # "SELECT style_id FROM tattoo_styles WHERE category_id = %s AND slug = %s", # (category_id, style_slug) # ) # exists = cursor.fetchone() # if exists: # # UPDATE # cursor.execute( # """ # UPDATE tattoo_styles # SET thumbnail_url = %s # WHERE category_id = %s AND slug = %s # """, # (thumbnail_url, category_id, style_slug) # ) # print(f"āœ… Updated tattoo_style {style_slug}") # else: # # INSERT # cursor.execute( # """ # INSERT INTO tattoo_styles (category_id, name, slug, thumbnail_url, is_active, created_at, updated_at) # VALUES (%s, %s, %s, %s, 1, %s, %s) # """, # (category_id, style_slug.capitalize(), style_slug, thumbnail_url, now, now) # ) # print(f"šŸ†• Inserted tattoo_style {style_slug}") # conn.commit() # cursor.close() # conn.close() # except Exception as e: # print(f"āŒ Failed to upsert tattoo_style {file}: {e}") # def upload_file(local_path, bucket, s3_key): # """ # Uploads a single file to S3 using upload_fileobj # """ # try: # with open(local_path, "rb") as f: # image_bytes = BytesIO(f.read()) # s3_client.upload_fileobj( # image_bytes, # bucket, # s3_key, # ExtraArgs={'ContentType': 'image/png'} # ) # print(f"āœ… Uploaded {local_path} ā†’ s3://{bucket}/{s3_key}") # except Exception as e: # print(f"āŒ Failed to upload {local_path}: {e}") # def upload_directory(local_dir, bucket, s3_prefix): # """ # Uploads category folders to S3. # Updates DB with icon_url and thumbnail_url for each category. # Also updates tattoo_styles for style images. # """ # for category_name in os.listdir(local_dir): # category_path = os.path.join(local_dir, category_name) # if not os.path.isdir(category_path): # continue # only process folders # icon_url = None # thumbnail_url = None # # Upload icon.png and thumbnail.png directly under category # for special_file in ["icon.png", "thumbnail.png"]: # special_path = os.path.join(category_path, special_file) # if os.path.exists(special_path): # s3_key = f"{s3_prefix}/{category_name}/{special_file}" # upload_file(special_path, bucket, s3_key) # file_url = f"https://{BASE_URL}/{s3_key}" # if special_file == "icon.png": # icon_url = file_url # elif special_file == "thumbnail.png": # thumbnail_url = file_url # # Update DB if both URLs are found # # if icon_url and thumbnail_url: # if icon_url: # update_category_urls(category_name, icon_url, thumbnail_url) # # Upload style images inside their own subfolder # for file in os.listdir(category_path): # if file in ["icon.png", "thumbnail.png"]: # continue # if file.lower().endswith((".png", ".jpg", ".jpeg", ".webp")): # s3_key = f"{s3_prefix}/{category_name}/style/{file}" # local_path = os.path.join(category_path, file) # # Upload to S3 # upload_file(local_path, bucket, s3_key) # # Update tattoo_styles DB # update_tattoo_style(category_name, file, s3_prefix) # if __name__ == "__main__": # upload_directory(LOCAL_BASE_DIR, BUCKET_NAME, "categories") # print("šŸŽ‰ All uploads finished!") import os import boto3 import mysql.connector from io import BytesIO from dotenv import load_dotenv from datetime import datetime import time load_dotenv() # ---------- CONFIG ---------- LOCAL_BASE_DIR = "Tattoo_new" BUCKET_NAME = os.getenv("CLOUDINARY_BUCKET_NAME") BASE_URL = os.getenv("CLOUD_FRONT_URL") # ---------- AWS CLIENT ---------- s3_client = boto3.client( "s3", aws_access_key_id=os.getenv("CLOUDINARY_API_KEY"), aws_secret_access_key=os.getenv("CLOUDINARY_API_SECRET"), ) # ---------- MYSQL CONFIG ---------- DB_HOST = os.getenv("DB_HOST", "localhost") DB_PORT = int(os.getenv("DB_PORT", 3306)) DB_USER = os.getenv("DB_USER", "root") DB_PASSWORD = os.getenv("DB_PASSWORD", "password") DB_NAME = os.getenv("DB_NAME", "zyBLyukn9LGSC8b9") now = datetime.now().strftime('%Y-%m-%d %H:%M:%S') def get_db_connection(): return mysql.connector.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, database=DB_NAME ) def format_category_name(folder_name): """ Convert folder name to proper category name Example: animal_nature -> Animal Nature """ return ' '.join(word.capitalize() for word in folder_name.split('_')) def insert_or_get_category(slug): """ Insert category if it doesn't exist, or return existing category_id Uses folder name as slug and formatted name as category name """ try: conn = get_db_connection() cursor = conn.cursor() # Check if category exists cursor.execute("SELECT category_id FROM categories WHERE slug = %s", (slug,)) row = cursor.fetchone() if row: print(f"šŸ“‚ Category '{slug}' already exists (ID: {row[0]})") cursor.close() conn.close() return row[0] # Category doesn't exist, insert it name = format_category_name(slug) created_at = int(time.time()) updated_at = created_at insert_query = """ INSERT INTO categories (name, slug, is_active, sort_order, created_at, updated_at) VALUES (%s, %s, 1, 0, %s, %s) """ cursor.execute(insert_query, (name, slug, created_at, updated_at)) conn.commit() category_id = cursor.lastrowid print(f"šŸ†• Created new category: '{name}' (slug: {slug}, ID: {category_id})") cursor.close() conn.close() return category_id except Exception as e: print(f"āŒ Failed to insert/get category '{slug}': {e}") return None def update_category_urls(category_name, icon_url, thumbnail_url): """ Update categories table with icon and thumbnail URLs where slug = category_name """ try: print('=====icon_url======', icon_url) print('=====thumbnail_url=====', thumbnail_url) conn = get_db_connection() cursor = conn.cursor() query = """ UPDATE categories SET icon_url = %s, thumbnail_url = %s, updated_at = %s WHERE slug = %s """ updated_at = int(time.time()) cursor.execute(query, (icon_url, thumbnail_url, updated_at, category_name)) conn.commit() print(f"šŸ“ Updated DB for category '{category_name}'") cursor.close() conn.close() except Exception as e: print(f"āŒ DB update failed for {category_name}: {e}") def update_tattoo_style(category_name, file, s3_prefix): """ Insert or update tattoo_styles thumbnail_url using category_id + slug """ try: conn = get_db_connection() cursor = conn.cursor() # Get category_id cursor.execute("SELECT category_id FROM categories WHERE slug = %s", (category_name,)) row = cursor.fetchone() if not row: print(f"āŒ No category found for slug={category_name}") return category_id = row[0] # Extract slug from filename style_slug, _ = os.path.splitext(file) # Construct thumbnail URL thumbnail_url = f"https://{BASE_URL}/{s3_prefix}/{category_name}/style/{file}" # Check if row exists cursor.execute( "SELECT style_id FROM tattoo_styles WHERE category_id = %s AND slug = %s", (category_id, style_slug) ) exists = cursor.fetchone() # Use datetime format for tattoo_styles timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') if exists: # UPDATE cursor.execute( """ UPDATE tattoo_styles SET thumbnail_url = %s, updated_at = %s WHERE category_id = %s AND slug = %s """, (thumbnail_url, timestamp, category_id, style_slug) ) print(f"āœ… Updated tattoo_style {style_slug}") else: # INSERT cursor.execute( """ INSERT INTO tattoo_styles (category_id, name, slug, thumbnail_url, is_active, created_at, updated_at) VALUES (%s, %s, %s, %s, 1, %s, %s) """, (category_id, style_slug.capitalize(), style_slug, thumbnail_url, timestamp, timestamp) ) print(f"šŸ†• Inserted tattoo_style {style_slug}") conn.commit() cursor.close() conn.close() except Exception as e: print(f"āŒ Failed to upsert tattoo_style {file}: {e}") def upload_file(local_path, bucket, s3_key): """ Uploads a single file to S3 using upload_fileobj """ try: with open(local_path, "rb") as f: image_bytes = BytesIO(f.read()) s3_client.upload_fileobj( image_bytes, bucket, s3_key, ExtraArgs={'ContentType': 'image/png'} ) print(f"āœ… Uploaded {local_path} ā†’ s3://{bucket}/{s3_key}") except Exception as e: print(f"āŒ Failed to upload {local_path}: {e}") def upload_directory(local_dir, bucket, s3_prefix): """ Uploads category folders to S3. First ensures category exists in DB, then uploads files. Updates DB with icon_url and thumbnail_url for each category. Also updates tattoo_styles for style images. """ for category_name in os.listdir(local_dir): category_path = os.path.join(local_dir, category_name) if not os.path.isdir(category_path): continue # only process folders print(f"\n{'='*60}") print(f"Processing category: {category_name}") print(f"{'='*60}") # First, ensure category exists in database category_id = insert_or_get_category(category_name) if not category_id: print(f"āš ļø Skipping {category_name} - could not create/get category") continue icon_url = None thumbnail_url = None # Upload icon.png and thumbnail.png directly under category for special_file in ["icon.png", "thumbnail.png"]: special_path = os.path.join(category_path, special_file) if os.path.exists(special_path): s3_key = f"{s3_prefix}/{category_name}/{special_file}" upload_file(special_path, bucket, s3_key) file_url = f"https://{BASE_URL}/{s3_key}" if special_file == "icon.png": icon_url = file_url elif special_file == "thumbnail.png": thumbnail_url = file_url # Update DB if URLs are found if icon_url: update_category_urls(category_name, icon_url, thumbnail_url) # Upload style images inside their own subfolder for file in os.listdir(category_path): if file in ["icon.png", "thumbnail.png"]: continue if file.lower().endswith((".png", ".jpg", ".jpeg", ".webp")): s3_key = f"{s3_prefix}/{category_name}/style/{file}" local_path = os.path.join(category_path, file) # Upload to S3 upload_file(local_path, bucket, s3_key) # Update tattoo_styles DB update_tattoo_style(category_name, file, s3_prefix) if __name__ == "__main__": upload_directory(LOCAL_BASE_DIR, BUCKET_NAME, "categories") print("\n" + "="*60) print("šŸŽ‰ All uploads finished!") print("="*60)