Upload 3 files

helpers_function/helper_email.py

@@ -0,0 +1,75 @@
+import logging
+import smtplib
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+
+from core.settings import cnf
+from database.connections import get_mongo_client
+
+logger = logging.getLogger(__name__)
+
+
+def _send_email(to_email: str, subject: str, body: str) -> bool:
+    """Low-level Gmail SMTP email sender."""
+    try:
+        SMTP_SERVER = "smtp.gmail.com"
+        SMTP_PORT = 587
+        EMAIL_USER = cnf.EMAIL_USER
+        EMAIL_PASS = cnf.EMAIL_PASS
+
+        if not all([to_email, subject, body, EMAIL_USER, EMAIL_PASS]):
+            logger.error("Missing required email parameters")
+            return False
+
+        msg = MIMEMultipart()
+        msg["From"] = EMAIL_USER
+        msg["To"] = to_email
+        msg["Subject"] = subject
+        msg.attach(MIMEText(body, "plain"))
+
+        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
+            server.starttls()
+            server.login(EMAIL_USER, EMAIL_PASS)
+            server.sendmail(EMAIL_USER, to_email, msg.as_string())
+
+        logger.info(f"Email sent to {to_email}")
+        return True
+    except Exception as e:
+        logger.error(f"Failed to send email: {e}")
+        return False
+
+
+def all_tasks_completed_notification(user_email: str, product_name: str) -> bool:
+    """
+    Send success email if the user has completed image, script, and video tasks.
+    """
+    db = get_mongo_client()
+    if db is None:
+        logger.error("MongoDB not available for notifications")
+        return False
+
+    has_image = db["results"].count_documents({"created_by": user_email, "type": {"$in": ["generation", "variation"]}}) > 0
+    has_script = db["script_generator"].count_documents({"created_by": user_email}) > 0
+    has_video = db["video_analyses"].count_documents({"created_by": user_email}) > 0
+
+
+    if has_image and has_script and has_video:
+        body = f"""Hey {user_email},
+
+🎉 Congratulations!
+
+You have successfully:
+- Generated Images
+- Generated Scripts
+- Analyzed Videos
+
+Your work for **{product_name}** has been completed successfully.  
+You can now review all assets in the AI Library.
+
+Best regards,  
+The Creative AdGenesis Team
+"""
+        return _send_email(user_email, "All Tasks Completed Successfully ", body)
+
+    logger.info(f"Tasks not fully completed for {user_email}")
+    return False

helpers_function/helper_meta_data.py

@@ -0,0 +1,166 @@
+import random
+import piexif
+from PIL import Image
+from io import BytesIO
+
+def generate_metadata(model):
+    # Define different values based on iPhone model
+    exposure_times = {
+        "iPhone 11": "1/60",
+        "iPhone 11 Pro": "1/70",
+        "iPhone 12": "1/100",
+        "iPhone 12 Pro": "1/110",
+        "iPhone 13": "1/120",
+        "iPhone 13 Pro": "1/130",
+        "iPhone 13 Pro Max": "1/140",
+        "iPhone 14": "1/200",
+        "iPhone 14 Pro": "1/220",
+        "iPhone 14 Pro Max": "1/240",
+        "iPhone 15": "1/300",
+        "iPhone 15 Plus": "1/320",
+        "iPhone 15 Pro": "1/400",
+        "iPhone 15 Pro Max": "1/500",
+        "iPhone 16": "1/600",
+        "iPhone 16 Pro": "1/700",
+        "iPhone 16 Pro Max": "1/1000"
+    }
+    f_numbers = {
+        "iPhone 11": "f/1.8",
+        "iPhone 11 Pro": "f/1.8",
+        "iPhone 12": "f/1.6",
+        "iPhone 12 Pro": "f/1.6",
+        "iPhone 13": "f/1.5",
+        "iPhone 13 Pro": "f/1.5",
+        "iPhone 13 Pro Max": "f/1.5",
+        "iPhone 14": "f/1.4",
+        "iPhone 14 Pro": "f/1.4",
+        "iPhone 14 Pro Max": "f/1.4",
+        "iPhone 15": "f/1.4",
+        "iPhone 15 Plus": "f/1.4",
+        "iPhone 15 Pro": "f/1.4",
+        "iPhone 15 Pro Max": "f/1.3",
+        "iPhone 16": "f/1.3",
+        "iPhone 16 Pro": "f/1.3",
+        "iPhone 16 Pro Max": "f/1.3"
+    }
+    focal_lengths = {
+        "iPhone 11": "3.99 mm",
+        "iPhone 11 Pro": "4.0 mm",
+        "iPhone 12": "4.2 mm",
+        "iPhone 12 Pro": "4.3 mm",
+        "iPhone 13": "5.0 mm",
+        "iPhone 13 Pro": "5.1 mm",
+        "iPhone 13 Pro Max": "5.2 mm",
+        "iPhone 14": "6.0 mm",
+        "iPhone 14 Pro": "6.1 mm",
+        "iPhone 14 Pro Max": "6.2 mm",
+        "iPhone 15": "6.0 mm",
+        "iPhone 15 Plus": "6.1 mm",
+        "iPhone 15 Pro": "6.1 mm",
+        "iPhone 15 Pro Max": "6.2 mm",
+        "iPhone 16": "6.5 mm",
+        "iPhone 16 Pro": "6.6 mm",
+        "iPhone 16 Pro Max": "6.7 mm"
+    }
+
+    metadata = {
+        "Make": "Apple",
+        "Model": model,
+        "Software": "iOS {}".format(random.choice([14, 15, 16, 17, 18, 19])),
+        "Orientation": random.choice(["Horizontal (normal)", "Rotate 90 CW", "Rotate 180", "Rotate 90 CCW"]),
+        "DateTime": "{}:{}:{} {:02}:{:02}:{:02}".format(random.randint(2022, 2024), random.randint(1, 12), random.randint(1, 28), random.randint(0, 23), random.randint(0, 59), random.randint(0, 59)),
+        "ExposureTime": exposure_times[model],
+        "FNumber": f_numbers[model],
+        "ISOSpeedRatings": random.choice([100, 200, 400, 800, 1600]),
+        "FocalLength": focal_lengths[model],
+        "Flash": random.choice(["Flash fired", "Flash did not fire, compulsory mode", "Flash fired, auto mode"]),
+        "WhiteBalance": random.choice(["Auto", "Manual"]),
+        "MeteringMode": random.choice(["Pattern", "CenterWeightedAverage", "Spot"]),
+        "SceneCaptureType": random.choice(["Standard", "Landscape", "Portrait", "NightScene"]),
+        "GPSLatitude": "{:.4f} N".format(random.uniform(0.0, 90.0)),
+        "GPSLongitude": "{:.4f} W".format(random.uniform(0.0, 180.0)),
+        "Altitude": "{:.1f} m".format(random.uniform(0, 100)),
+        "LensMake": "Apple",
+        "LensModel": "{} back triple camera {} f/1.5".format(model, focal_lengths[model]),
+        "ColorSpace": random.choice(["sRGB", "Adobe RGB"]),
+        "PixelXDimension": random.choice([3024, 4032, 2160]),
+        "PixelYDimension": random.choice([3024, 4032, 2160]),
+        "ExposureBiasValue": random.choice(["0 EV", "+1 EV", "-1 EV"]),
+        "BrightnessValue": "{:.1f}".format(random.uniform(-5, 10)),
+        "ExposureMode": random.choice(["Auto Exposure", "Manual Exposure"])
+    }
+    return metadata
+
+def dms_coordinates(value):
+    """Convert decimal degrees to EXIF GPS (DMS format)."""
+    degrees = int(value)
+    minutes_float = (value - degrees) * 60
+    minutes = int(minutes_float)
+    seconds = round((minutes_float - minutes) * 60 * 10000)
+    return [(degrees, 1), (minutes, 1), (seconds, 10000)]
+
+def meta_data_helper_function(image_bytes,model=None):
+    """
+    Takes raw image bytes, adds realistic EXIF metadata, and returns new bytes.
+    """
+    if model is None:
+        model = random.choice([
+            "iPhone 11", "iPhone 11 Pro", "iPhone 12", "iPhone 12 Pro",
+            "iPhone 13", "iPhone 13 Pro", "iPhone 13 Pro Max", "iPhone 14",
+            "iPhone 14 Pro", "iPhone 14 Pro Max", "iPhone 15", "iPhone 15 Plus",
+            "iPhone 15 Pro", "iPhone 15 Pro Max", "iPhone 16", "iPhone 16 Pro",
+            "iPhone 16 Pro Max"
+        ])
+
+    # Load image from bytes
+    img = Image.open(BytesIO(image_bytes))
+    img = img.convert("RGB")  # Ensure compatibility
+
+    # Generate metadata
+    metadata = generate_metadata(model)
+
+    # Build EXIF data
+    exif_dict = {"0th": {}, "Exif": {}, "GPS": {}}
+
+    exif_dict["0th"][piexif.ImageIFD.Make] = metadata["Make"]
+    exif_dict["0th"][piexif.ImageIFD.Model] = metadata["Model"]
+    exif_dict["0th"][piexif.ImageIFD.Software] = metadata["Software"]
+    exif_dict["0th"][piexif.ImageIFD.DateTime] = metadata["DateTime"]
+
+    # Exposure time (rational)
+    num, den = map(int, metadata["ExposureTime"].split("/"))
+    exif_dict["Exif"][piexif.ExifIFD.ExposureTime] = (num, den)
+
+    # FNumber
+    f_number = float(metadata["FNumber"].replace("f/", ""))
+    exif_dict["Exif"][piexif.ExifIFD.FNumber] = (int(f_number * 10), 10)
+
+    # ISO
+    exif_dict["Exif"][piexif.ExifIFD.ISOSpeedRatings] = metadata["ISOSpeedRatings"]
+
+    # Focal Length
+    focal_length = float(metadata["FocalLength"].split()[0])
+    exif_dict["Exif"][piexif.ExifIFD.FocalLength] = (int(focal_length * 10), 10)
+
+    # GPS
+    lat = float(metadata["GPSLatitude"].split()[0])
+    lon = float(metadata["GPSLongitude"].split()[0])
+
+    exif_dict["GPS"][piexif.GPSIFD.GPSLatitudeRef] = b'N' if lat >= 0 else b'S'
+    exif_dict["GPS"][piexif.GPSIFD.GPSLatitude] = dms_coordinates(abs(lat))
+
+    exif_dict["GPS"][piexif.GPSIFD.GPSLongitudeRef] = b'E' if lon >= 0 else b'W'
+    exif_dict["GPS"][piexif.GPSIFD.GPSLongitude] = dms_coordinates(abs(lon))
+
+    altitude_value = float(metadata["Altitude"].replace(" m", ""))
+    exif_dict["GPS"][piexif.GPSIFD.GPSAltitude] = (int(altitude_value), 1)
+
+
+    # Dump to bytes
+    exif_bytes = piexif.dump(exif_dict)
+
+    # Save to in-memory buffer
+    output_io = BytesIO()
+    img.save(output_io, format="JPEG", exif=exif_bytes)
+    return output_io.getvalue()
+

helpers_function/helpers.py

@@ -0,0 +1,61 @@
+import boto3
+from uuid import uuid4
+import os
+
+import cv2
+import base64
+import logging
+
+logger = logging.getLogger(__name__)
+
+def get_video_thumbnail_base64(video_path: str, time_sec: int = 1) -> str:
+    try:
+        cap = cv2.VideoCapture(video_path)
+        cap.set(cv2.CAP_PROP_POS_MSEC, time_sec * 1000)
+        success, frame = cap.read()
+        cap.release()
+        if not success:
+            return ""
+        _, buffer = cv2.imencode(".jpg", frame)
+        return base64.b64encode(buffer).decode("utf-8")
+    except Exception:
+        logger.exception("Thumbnail extraction failed")
+        return ""
+
+
+log = logging.getLogger(__name__)
+
+def upload_image_to_r2(image_bytes, folder_name="search_arb", app_type="bulk_generation"):
+    try:
+        s3 = boto3.client(
+            "s3",
+            endpoint_url=os.getenv('R2_ENDPOINT'),
+            aws_access_key_id=os.getenv('R2_ACCESS_KEY'),
+            aws_secret_access_key=os.getenv('R2_SECRET_KEY'),
+            region_name="auto"
+        )
+        filename = f"{uuid4().hex}.png"
+        file_key = f"hug_face/{app_type.strip('/')}/{folder_name.strip('/')}/{filename}" if folder_name else f"hug_face/{app_type.strip('/')}/{filename}"
+        s3.put_object(Bucket=os.getenv('R2_BUCKET_NAME'), Key=file_key, Body=image_bytes, ContentType="image/png")
+        return f"{os.getenv('NEW_BASE').rstrip('/')}/{file_key}"
+    except Exception as e:
+        log.error(e)
+        return None
+
+
+def encode_image_to_base64(image_path: str) -> str:
+    with open(image_path, "rb") as f:
+        return base64.b64encode(f.read()).decode("utf-8")
+
+def is_valid_image(file_name: str) -> bool:
+    return file_name.lower().endswith((".png", ".jpg", ".jpeg", ".bmp", ".gif", ".webp"))
+
+def _mean_effectiveness(metrics):
+    if not metrics: return 0.0
+    scores = []
+    for m in metrics:
+        s = str(m.get("effectiveness_score", "0/10")).split("/")[0]
+        try: scores.append(int(s))
+        except Exception: pass
+    return round(sum(scores) / len(scores), 2) if scores else 0.0
+