File size: 5,130 Bytes
677c57e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 |
# from flask import Flask, jsonify
# import cloudinary
# import cloudinary.api
# import cloudinary.uploader
# import os
# import glob
# import requests
# from dotenv import load_dotenv
# from deepface.commons.logger import Logger
# logger = Logger(module="modules/cloudservice.py")
# load_dotenv()
# # Configure Cloudinary
# cloudinary.config(
# cloud_name=os.getenv('CLOUDINARY_CLOUD_NAME'),
# api_key=os.getenv('CLOUDINARY_API_KEY'),
# api_secret=os.getenv('CLOUDINARY_API_SECRET')
# )
# def fetch_cloudinary_images(folder_name):
# resources = []
# res = cloudinary.api.resources(type='upload', resource_type='image', prefix=f'mafqoud/images/{folder_name}')
# resources.extend(res.get('resources', []))
# return resources
# def download_image(url, local_path):
# response = requests.get(url, stream=True)
# if response.status_code == 200:
# with open(local_path, 'wb') as file:
# for chunk in response:
# file.write(chunk)
# def sync_folder(folder_name, local_dir):
# cloudinary_images = fetch_cloudinary_images(folder_name)
# cloudinary_urls = {img['secure_url']: img['public_id'] for img in cloudinary_images}
# # Download new images and track downloaded image paths
# downloaded_paths = []
# for img in cloudinary_images:
# url = img['secure_url']
# public_id = img['public_id']
# file_name = url.split('/')[-1] # Get the actual file name
# local_path = os.path.join(local_dir, file_name)
# print(local_path)
# if not os.path.exists(local_path):
# download_image(url, local_path)
# downloaded_paths.append(local_path)
# # Remove old images
# local_images = [os.path.join(local_dir, f) for f in os.listdir(local_dir) if os.path.isfile(os.path.join(local_dir, f))]
# for local_path in local_images:
# if local_path not in downloaded_paths and (local_path.endswith('.jpg') or local_path.endswith('.jpeg') or local_path.endswith('.png')):
# os.remove(local_path)
# def delete_pkl_files(directory):
# """Delete all .pkl files in the specified directory."""
# pkl_files = glob.glob(os.path.join(directory, '*.pkl'))
# for pkl_file in pkl_files:
# os.remove(pkl_file)
from flask import Flask, jsonify
import cloudinary
import cloudinary.api
import cloudinary.uploader
import os
import glob
import requests
from dotenv import load_dotenv
from deepface.commons.logger import Logger
from deepface import DeepFace # Assuming this is the correct import for training
logger = Logger(module="modules/cloudservice.py")
load_dotenv()
# Configure Cloudinary
cloudinary.config(
cloud_name=os.getenv('CLOUDINARY_CLOUD_NAME'),
api_key=os.getenv('CLOUDINARY_API_KEY'),
api_secret=os.getenv('CLOUDINARY_API_SECRET')
)
def fetch_cloudinary_images(folder_name):
resources = []
next_cursor = None
while True:
if next_cursor:
res = cloudinary.api.resources(
type='upload',
resource_type='image',
prefix=f'mafqoud/images/{folder_name}',
max_results=500,
next_cursor=next_cursor
)
else:
res = cloudinary.api.resources(
type='upload',
resource_type='image',
prefix=f'mafqoud/images/{folder_name}',
max_results=500
)
resources.extend(res.get('resources', []))
next_cursor = res.get('next_cursor')
if not next_cursor:
break
return resources
def download_image(url, local_path):
response = requests.get(url, stream=True)
if response.status_code == 200:
with open(local_path, 'wb') as file:
for chunk in response:
file.write(chunk)
def sync_folder(folder_name, local_dir):
cloudinary_images = fetch_cloudinary_images(folder_name)
cloudinary_urls = {img['secure_url']: img['public_id'] for img in cloudinary_images}
# Download new images and track downloaded image paths
downloaded_paths = []
for img in cloudinary_images:
url = img['secure_url']
public_id = img['public_id']
file_name = url.split('/')[-1] # Get the actual file name
local_path = os.path.join(local_dir, file_name)
print(local_path)
if not os.path.exists(local_path):
download_image(url, local_path)
downloaded_paths.append(local_path)
# Remove old images
local_images = [os.path.join(local_dir, f) for f in os.listdir(local_dir) if os.path.isfile(os.path.join(local_dir, f))]
for local_path in local_images:
if local_path not in downloaded_paths and (local_path.endswith('.jpg') or local_path.endswith('.jpeg') or local_path.endswith('.png')):
os.remove(local_path)
def delete_pkl_files(directory):
"""Delete all .pkl files in the specified directory."""
pkl_files = glob.glob(os.path.join(directory, '*.pkl'))
for pkl_file in pkl_files:
os.remove(pkl_file)
|