Upload 5 files

src/helpers.py

@@ -0,0 +1,8 @@
+import base64, os
+
+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"))

src/meta_data.py

@@ -0,0 +1,165 @@
+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()

src/mongo_logger.py

@@ -0,0 +1,36 @@
+import os, uuid
+from pymongo import MongoClient
+from datetime import datetime, timezone
+
+client = MongoClient(os.getenv("MONGO_URI"))
+collection = client.adgenesis.records
+
+def create_log(category, file_name, prompt):
+    log_id = str(uuid.uuid4())
+    collection.insert_one({
+        "_id": log_id,
+        "category": category,
+        "prompt": prompt,
+        "file_name": file_name,
+        "status": "in_progress",
+        "urls": [],
+        "message": "Processing started",
+        "created_at": datetime.now(timezone.utc),
+        "updated_at": datetime.now(timezone.utc),
+        "lob": "test"
+    })
+    return log_id
+
+def update_log_status(log_id, status, urls=None, message=None):
+    update = {
+        "$set": {
+            "status": status,
+            "updated_at": datetime.now(timezone.utc)
+        }
+    }
+    if urls:
+        update["$set"]["urls"] = urls
+    if message:
+        update["$set"]["message"] = message
+
+    collection.update_one({"_id": log_id}, update)

src/prompt_service.py

@@ -0,0 +1,55 @@
+import openai, os, json, re
+
+openai.api_key = os.getenv("OPENAI_API_KEY")
+sys_prompt = os.getenv("SYS_PROMPT")
+
+def get_prompts(image_base64, category, user_prompt, sentiment, negative_prompt):
+    try:
+        if negative_prompt:
+            message = [
+                {
+                    "role": "system",
+                    "content": f"""{sys_prompt}
+                            Return only a JSON with 2 variations of ad prompts for image generation, based on the input image.
+                            Respond in this JSON format:\n{{\"variations\": [\"Prompt 1\", \"Prompt 2\", ..., \"Prompt 10\"]}}"""
+                },
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": f"""Generate prompt variations for this ad image for {category} category having {sentiment} sentiment in JSON format only based on the following instruction:
+                                                    {user_prompt}
+                                                    Don't consider following things and treat them as negative prompt:
+                                                    {negative_prompt}"""},
+                        {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}}
+                    ]
+                }
+            ]
+        else:
+            message = [
+                {
+                    "role": "system",
+                    "content": f"""{sys_prompt}
+                                        Return only a JSON with 2 variations of ad prompts for image generation, based on the input image.
+                                        Respond in this JSON format:\n{{\"variations\": [\"Prompt 1\", \"Prompt 2\", ..., \"Prompt 10\"]}}"""
+                },
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text",
+                         "text": f"""Generate prompt variations for this ad image for {category} category having {sentiment} sentiment in JSON format only based on the following instruction:
+                                                                {user_prompt}"""},
+                        {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}}
+                    ]
+                }
+            ]
+
+        response = openai.chat.completions.create(
+            model="gpt-4o",
+            messages=message,
+            temperature=0.7
+        )
+        content = response.choices[0].message.content.strip()
+        content = re.sub(r"^```json\s*|\s*```$", "", content)
+        return json.loads(content)["variations"]
+    except Exception:
+        return []

src/r2_uploader.py

@@ -0,0 +1,13 @@
+import boto3, os
+from uuid import uuid4
+
+def upload_image_to_r2(image_bytes):
+    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")
+    )
+    file_key = f"{uuid4().hex}.png"
+    s3.put_object(Bucket=os.getenv("R2_BUCKET_NAME"), Key=file_key, Body=image_bytes, ContentType="image/png")
+    return f"{os.getenv('R2_ENDPOINT')}/{os.getenv('R2_BUCKET_NAME')}/{file_key}"