import os import base64 import json #import csv import requests from dotenv import load_dotenv from database import Database class Describe: def __init__(self): load_dotenv() self.base_url = os.environ.get('BASE_URL') self.api_key = os.environ.get('API_KEY') self.image_path = os.environ.get('IMAGE_PATH') self.category_path = os.environ.get('CATEGORY_PATH') self.db = Database() def load_categories(self): # To open a directory and list its elements, use os.listdir: categories = os.listdir(self.category_path) img_categories = os.listdir(self.image_path) for category in categories: if category in img_categories: hay_fotos = True else: hay_fotos = False category_name = ' '.join(category.capitalize().split('_')) self.db.add_category(category_name, category, hay_fotos) for img_category in img_categories: if img_category not in categories: img_category_name = ' '.join(img_category.capitalize().split('_')) self.db.add_category( img_category_name, img_category, hay_tamanos=False) def load_images(self): categories = self.db.get_all("Category", type="pictures") for category in categories: categoria = category.name prompt = f"Describe en detalle esta figura de yeso {categoria} para un sitio web de comercio electrónico. Usa entre 400 y 800 caracteres." prompt += "Que la descripción se enfoque en las formas del producto y sus características. Evita hablar de tamaño en la descripción. Se conciso y directo." prompt += "Propon un nombre corto para el producto que lo identifique, además de la descripción." prompt += f"Se creativo, no uses las palabras escultura, figura, yeso, ni {categoria} en el nombre del producto." prompt += "Entrega tus resultados en español con el siguiente formato: " prompt += "{\"name\": \"Nombre del producto\", \"description\": \"Descripción del producto\"}" headers = { 'content-type': 'application/json', 'Authorization': f"Bearer {self.api_key}", } images = os.listdir(f"{self.image_path}{category.dir_name}") if len(images) == 0: print(f"No images found in {category.dir_name}") continue for image in images: image_path = f"{self.image_path}{category.dir_name}/{image}" try: with open(image_path, "rb") as image_file: if image_file.name != image_path: print(f"Image {image} not found in {image_path}") continue if self.db.image_exists(image_file.name): print(f"Image {image_file.name} already exists in database. Skipping.") continue image_data = image_file.read() base64_image = base64.b64encode(image_data).decode('utf-8') data = { "image": base64_image, "question": prompt, } try: response = requests.post( self.base_url, headers=headers, json=data) response.raise_for_status() response = response.json()['response'] try: response_json = json.loads(response) except json.decoder.JSONDecodeError as e: print(f"JSON decode error for image '{image_file.name}': {e}\n") with open("json_decode_errors.log", "a", encoding="utf-8") as log_file: import datetime log_file.write("************************************************\n") log_file.write(f"Timestamp: {datetime.datetime.now().isoformat()}\n") log_file.write(f"JSON decode error for image '{image_file.name}': {e}\n") log_file.write(f"Raw response: {response}\n") log_file.write("************************************************\n") continue response = response_json self.db.add_image( response['name'], image_file.name, response['description'], category.id) print(f"Image {image_file.name} added to category {category.name}") last_record = self.db.last_record() print(f"Last record: {last_record}") except requests.exceptions.RequestException as e: if e.response is not None: try: print(f"API Response Error body: {e.response.text}") except Exception as ex: print(f"Could not read API error response: {ex}") print(f"Error making request: {e}") except (FileNotFoundError, PermissionError, IOError) as e: print(f"Error opening image file: {e}") print(f"Images added to category {category.name}: {self.db.count_images(category.id)}") print(f"Images in category {category.name}: {len(images)}") def count_images_in_category(self, category_id=None): """ Count images by category in both folder and database. If category_id is provided, only count for that category. Otherwise, count for all categories and show totals. """ categories = self.db.get_all("Category", type="pictures") total_folder_images = 0 total_db_images = 0 category_counts = [] print("\n" + "="*60) print("IMAGE COUNT BY CATEGORY") print("="*60) print(f"{'Category':<30} {'Folder':<10} {'Database':<10}") print("-"*60) for category in categories: # Count images in folder try: folder_path = f"{self.image_path}{category.dir_name}" if os.path.exists(folder_path): images_in_folder = len([f for f in os.listdir(folder_path) if os.path.isfile(os.path.join(folder_path, f))]) else: images_in_folder = 0 except (OSError, PermissionError) as e: print(f"Error accessing folder for {category.name}: {e}") images_in_folder = 0 # Count images in database images_in_db = self.db.count_images(category.id) # Add to totals total_folder_images += images_in_folder total_db_images += images_in_db # Store category counts category_counts.append({ 'name': category.name, 'folder_count': images_in_folder, 'db_count': images_in_db }) # Print category count print(f"{category.name:<30} {images_in_folder:<10} {images_in_db:<10}") # Print totals # Write output to a file instead of printing to stdout output_filename = "image_category_counts.txt" with open(output_filename, "w", encoding="utf-8") as f: f.write("\n" + "="*60 + "\n") f.write("IMAGE COUNT BY CATEGORY\n") f.write("="*60 + "\n") f.write(f"{'Category':<30} {'Folder':<10} {'Database':<10}\n") f.write("-"*60 + "\n") for c in category_counts: f.write(f"{c['name']:<30} {c['folder_count']:<10} {c['db_count']:<10}\n") f.write("-"*60 + "\n") f.write(f"{'TOTAL':<30} {total_folder_images:<10} {total_db_images:<10}\n") f.write("="*60 + "\n\n") return { 'total_folder_images': total_folder_images, 'total_db_images': total_db_images, 'categories': category_counts } if __name__ == "__main__": describe = Describe() #describe.load_categories() # categories = describe.db.get_all("Category") # Create CSV file from categories #csv_filename = "categories.csv" #with open(csv_filename, 'w', newline='', encoding='utf-8') as csvfile: # fieldnames = ['id', 'name', 'dir_name', 'hay_fotos', 'hay_tamanos'] # writer = csv.DictWriter(csvfile, fieldnames=fieldnames) # # writer.writeheader() # for category in categories: # writer.writerow({ # 'id': category.id, # 'name': category.name, # 'dir_name': category.dir_name, # 'hay_fotos': category.hay_fotos, # 'hay_tamanos': category.hay_tamanos # }) #print(f"CSV file '{csv_filename}' created successfully with {len(categories)} categories.") #describe.load_images() describe.count_images_in_category()