Best Practice for Remote Real Estate Photographers Delivering Virtual Tours Efficiently

#!/usr/bin/env python3
import os
import subprocess
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path
from datetime import datetime

class VirtualTourProcessor:
    def __init__(self, base_path, output_dir):
        self.base_path = Path(base_path)
        self.output_dir = Path(output_dir)
        self.quality = 85
        self.max_workers = 4

    def optimize_image(self, image_path):
        """Optimize single image using ImageMagick"""
        output_name = f"opt_{image_path.name}"
        output_path = self.output_dir / output_name

        cmd = [
            'convert', str(image_path),
            '-quality', str(self.quality),
            '-resize', '1920x1080>',
            '-auto-orient',
            str(output_path)
        ]

        subprocess.run(cmd, check=True, capture_output=True)
        return output_path

    def process_property(self, property_dir):
        """Process single property directory"""
        property_name = property_dir.name
        property_output = self.output_dir / property_name
        property_output.mkdir(parents=True, exist_ok=True)

        images = list(property_dir.glob('*.jpg')) + list(property_dir.glob('*.png'))

        with ProcessPoolExecutor(max_workers=self.max_workers) as executor:
            futures = {executor.submit(self.optimize_image, img): img for img in images}

            results = []
            for future in as_completed(futures):
                img = futures[future]
                try:
                    result = future.result()
                    results.append(result)
                    print(f"Processed: {img.name}")
                except Exception as e:
                    print(f"Error processing {img.name}: {e}")

        return property_name, len(results)

    def batch_process(self):
        """Process all property directories"""
        property_dirs = [d for d in self.base_path.iterdir() if d.is_dir()]
        results = []

        for prop_dir in property_dirs:
            name, count = self.process_property(prop_dir)
            results.append(f"{name}: {count} images")

        return results

# Usage
processor = VirtualTourProcessor(
    base_path='/path/to/raw_properties',
    output_dir='/path/to/processed_properties'
)
results = processor.batch_process()
for r in results:
    print(r)

import boto3
from datetime import datetime, timedelta

class TourDeliveryService:
    def __init__(self, bucket_name, aws_region='us-east-1'):
        self.s3 = boto3.client('s3', region_name=aws_region)
        self.bucket = bucket_name
        self.url_expiry = 7 * 24 * 60 * 60  # 7 days

    def upload_tour(self, local_path, property_id):
        """Upload tour files to S3 with proper structure"""
        s3_prefix = f"tours/{property_id}"

        for root, dirs, files in os.walk(local_path):
            for file in files:
                local_file = os.path.join(root, file)
                relative_path = os.path.relpath(local_file, local_path)
                s3_key = f"{s3_prefix}/{relative_path}"

                self.s3.upload_file(
                    local_file,
                    self.bucket,
                    s3_key,
                    ExtraArgs={
                        'ContentType': self.get_content_type(file),
                        'CacheControl': 'max-age=31536000'
                    }
                )

        return s3_prefix

    def generate_delivery_link(self, s3_prefix, client_email):
        """Generate time-limited delivery link"""
        url = self.s3.generate_presigned_url(
            'list_objects_v2',
            Params={
                'Bucket': self.bucket,
                'Prefix': s3_prefix
            },
            ExpiresIn=self.url_expiry
        )

        # Store delivery record
        delivery_record = {
            'property_id': s3_prefix.split('/')[-1],
            'client_email': client_email,
            'generated_at': datetime.utcnow().isoformat(),
            'expires_at': (datetime.utcnow() + timedelta(seconds=self.url_expiry)).isoformat()
        }

        return url, delivery_record

    @staticmethod
    def get_content_type(filename):
        ext = filename.split('.')[-1].lower()
        types = {
            'json': 'application/json',
            'jpg': 'image/jpeg',
            'jpeg': 'image/jpeg',
            'png': 'image/png',
            'html': 'text/html'
        }
        return types.get(ext, 'application/octet-stream')

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from string import Template

DELIVERY_EMAIL_TEMPLATE = Template("""
Subject: Your Virtual Tour for $property_address is Ready

Hi $client_name,

Your virtual tour for $property_address is now ready for review.

View your tour: $tour_url

This link will be active for 7 days. If you need an extension,
reply to this email and we'll generate a new link.

Questions about the tour? Reply here or call $photographer_phone.

Best,
$photographer_name
""")

def send_delivery_notification(client_email: str, delivery_info: dict, smtp_config: dict):
    msg = MIMEMultipart()
    msg["From"] = smtp_config["from_address"]
    msg["To"] = client_email
    msg["Subject"] = f"Your Virtual Tour for {delivery_info['property_address']} is Ready"

    body = DELIVERY_EMAIL_TEMPLATE.substitute(**delivery_info)
    msg.attach(MIMEText(body, "plain"))

    with smtplib.SMTP_SSL(smtp_config["host"], 465) as server:
        server.login(smtp_config["username"], smtp_config["password"])
        server.sendmail(smtp_config["from_address"], client_email, msg.as_string())

from redis import Redis
from rq import Queue
from rq.job import Job

redis_conn = Redis()
tour_queue = Queue("tour_processing", connection=redis_conn)

def enqueue_property(property_dir: str, output_dir: str, client_email: str) -> str:
    """Add a property to the processing queue. Returns job ID."""
    job = tour_queue.enqueue(
        process_and_deliver_property,
        property_dir,
        output_dir,
        client_email,
        job_timeout=1800,  # 30 minutes max per property
        result_ttl=86400   # Keep result for 24 hours
    )
    return job.id

def get_job_status(job_id: str) -> dict:
    """Check the status of a queued property."""
    job = Job.fetch(job_id, connection=redis_conn)
    return {
        "id": job.id,
        "status": job.get_status(),
        "enqueued_at": job.enqueued_at.isoformat() if job.enqueued_at else None,
        "started_at": job.started_at.isoformat() if job.started_at else None,
        "ended_at": job.ended_at.isoformat() if job.ended_at else None,
    }