# Install required packages
# pip install azure-storage-file-datalake azure-identity

from azure.storage.filedatalake import DataLakeServiceClient
from azure.identity import DefaultAzureCredential

# Initialize client with managed identity (recommended for production)
credential = DefaultAzureCredential()
service_client = DataLakeServiceClient(
    account_url="https://yourstorageaccount.dfs.core.windows.net",
    credential=credential
)

# Alternative: Connection string (for development)
# service_client = DataLakeServiceClient.from_connection_string(
#     "DefaultEndpointsProtocol=https;AccountName=youraccount;AccountKey=yourkey;EndpointSuffix=core.windows.net"
# )

# Get file system (container) client
file_system_client = service_client.get_file_system_client("datalake")

# Create directory structure
directory_client = file_system_client.get_directory_client("data/raw/2024/01")
directory_client.create_directory()

# Upload a file
file_client = file_system_client.get_file_client("data/raw/2024/01/sales_data.csv")

# Upload from local file
with open("local_sales_data.csv", "rb") as data:
    file_client.upload_data(data, overwrite=True)

# Upload from string/bytes
csv_content = "date,product,sales\n2024-01-01,Widget A,100\n2024-01-02,Widget B,150"
file_client.upload_data(csv_content.encode(), overwrite=True)

print(f"File uploaded: {file_client.url}")

# Download file content
file_client = file_system_client.get_file_client("data/raw/2024/01/sales_data.csv")
download = file_client.download_file()
content = download.readall().decode('utf-8')

print("File content:")
print(content)

# Stream large files efficiently
def process_large_file(file_path):
    file_client = file_system_client.get_file_client(file_path)
    
    # Download in chunks to manage memory
    download = file_client.download_file()
    
    chunk_size = 1024 * 1024  # 1MB chunks
    processed_lines = 0
    
    while True:
        chunk = download.read(chunk_size)
        if not chunk:
            break
            
        # Process chunk (example: count lines)
        processed_lines += chunk.decode('utf-8').count('\n')
    
    return processed_lines

# Usage
line_count = process_large_file("data/raw/2024/01/large_dataset.csv")
print(f"Processed {line_count} lines")

# List files and directories
def list_directory_contents(directory_path):
    directory_client = file_system_client.get_directory_client(directory_path)
    
    paths = directory_client.get_paths()
    
    files = []
    directories = []
    
    for path in paths:
        if path.is_directory:
            directories.append({
                'name': path.name,
                'last_modified': path.last_modified,
                'type': 'directory'
            })
        else:
            files.append({
                'name': path.name,
                'size': path.content_length,
                'last_modified': path.last_modified,
                'type': 'file'
            })
    
    return {'files': files, 'directories': directories}

# List contents
contents = list_directory_contents("data/raw/2024")
print(f"Found {len(contents['files'])} files and {len(contents['directories'])} directories")

# Get file metadata
file_client = file_system_client.get_file_client("data/raw/2024/01/sales_data.csv")
properties = file_client.get_file_properties()

print(f"File size: {properties.size} bytes")
print(f"Last modified: {properties.last_modified}")
print(f"Content type: {properties.content_settings.content_type}")

import pandas as pd
from io import StringIO

# Read CSV directly from ADLS into pandas
def read_csv_from_adls(file_path):
    file_client = file_system_client.get_file_client(file_path)
    download = file_client.download_file()
    content = download.readall().decode('utf-8')
    
    # Create DataFrame from string content
    df = pd.read_csv(StringIO(content))
    return df

# Usage
df = read_csv_from_adls("data/raw/2024/01/sales_data.csv")
print(df.head())

# Write processed DataFrame back to ADLS
def write_dataframe_to_adls(df, file_path, file_format='csv'):
    if file_format == 'csv':
        content = df.to_csv(index=False)
    elif file_format == 'parquet':
        # For parquet, we need to use BytesIO
        from io import BytesIO
        buffer = BytesIO()
        df.to_parquet(buffer, index=False)
        content = buffer.getvalue()
    
    file_client = file_system_client.get_file_client(file_path)
    
    if file_format == 'csv':
        file_client.upload_data(content.encode(), overwrite=True)
    else:
        file_client.upload_data(content, overwrite=True)

# Process and save data
processed_df = df.groupby('product')['sales'].sum().reset_index()
write_dataframe_to_adls(processed_df, "data/processed/2024/01/sales_summary.csv")

import os
from datetime import datetime, timedelta

def process_daily_files(start_date, end_date, source_path, target_path):
    """
    Process files for a date range
    """
    current_date = start_date
    processed_files = []
    
    while current_date <= end_date:
        date_str = current_date.strftime("%Y/%m/%d")
        source_file = f"{source_path}/{date_str}/sales_data.csv"
        target_file = f"{target_path}/{date_str}/processed_sales.csv"
        
        try:
            # Check if source file exists
            file_client = file_system_client.get_file_client(source_file)
            file_client.get_file_properties()  # This will raise if file doesn't exist
            
            # Process the file
            df = read_csv_from_adls(source_file)
            
            # Example processing: add calculated columns
            df['total_revenue'] = df['sales'] * df.get('price', 10)  # Assume price if not present
            df['processing_date'] = datetime.now().isoformat()
            
            # Ensure target directory exists
            target_dir = os.path.dirname(target_file)
            directory_client = file_system_client.get_directory_client(target_dir)
            directory_client.create_directory()
            
            # Save processed file
            write_dataframe_to_adls(df, target_file)
            processed_files.append(target_file)
            
            print(f"Processed {source_file} -> {target_file}")
            
        except Exception as e:
            print(f"Error processing {source_file}: {str(e)}")
        
        current_date += timedelta(days=1)
    
    return processed_files

# Usage
from datetime import date
start = date(2024, 1, 1)
end = date(2024, 1, 7)

processed = process_daily_files(
    start, end, 
    "data/raw", 
    "data/processed"
)

print(f"Processed {len(processed)} files")

# Set ACL (Access Control List) permissions
def set_directory_permissions(directory_path, permissions):
    """
    Set POSIX-style permissions on a directory
    permissions format: "user::rwx,group::r--,other::---"
    """
    directory_client = file_system_client.get_directory_client(directory_path)
    
    # Set ACL
    directory_client.set_access_control(permissions=permissions)
    
    print(f"Permissions set for {directory_path}")

# Example: Give read/write to owner, read to group, no access to others
set_directory_permissions("data/sensitive", "user::rwx,group::r--,other::---")

# Get current permissions
def get_directory_permissions(directory_path):
    directory_client = file_system_client.get_directory_client(directory_path)
    acl = directory_client.get_access_control()
    
    return {
        'permissions': acl['permissions'],
        'owner': acl.get('owner'),
        'group': acl.get('group')
    }

permissions = get_directory_permissions("data/sensitive")
print(f"Current permissions: {permissions}")

# Production-ready authentication using Managed Identity
from azure.identity import ManagedIdentityCredential, ChainedTokenCredential, AzureCliCredential

def get_authenticated_client(storage_account_name):
    """
    Get authenticated client using managed identity with fallback to Azure CLI
    """
    # Try managed identity first (works in Azure services)
    managed_identity = ManagedIdentityCredential()
    
    # Fallback to Azure CLI (works in local development)
    cli_credential = AzureCliCredential()
    
    # Chain credentials - try managed identity first, then CLI
    credential = ChainedTokenCredential(managed_identity, cli_credential)
    
    account_url = f"https://{storage_account_name}.dfs.core.windows.net"
    
    return DataLakeServiceClient(
        account_url=account_url,
        credential=credential
    )

# Usage in production
service_client = get_authenticated_client("yourstorageaccount")

import concurrent.futures
from threading import Lock

# Thread-safe file processing
class ADLSProcessor:
    def __init__(self, service_client):
        self.service_client = service_client
        self.file_system_client = service_client.get_file_system_client("datalake")
        self.results_lock = Lock()
        self.results = []
    
    def process_file(self, file_path):
        """Process a single file"""
        try:
            # Read file
            file_client = self.file_system_client.get_file_client(file_path)
            download = file_client.download_file()
            content = download.readall().decode('utf-8')
            
            # Simple processing: count lines
            line_count = content.count('\n')
            
            # Thread-safe result storage
            with self.results_lock:
                self.results.append({
                    'file': file_path,
                    'lines': line_count,
                    'size': len(content)
                })
            
            return f"Processed {file_path}: {line_count} lines"
            
        except Exception as e:
            return f"Error processing {file_path}: {str(e)}"
    
    def process_files_parallel(self, file_paths, max_workers=5):
        """Process multiple files in parallel"""
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            # Submit all tasks
            future_to_file = {
                executor.submit(self.process_file, file_path): file_path 
                for file_path in file_paths
            }
            
            # Collect results
            for future in concurrent.futures.as_completed(future_to_file):
                result = future.result()
                print(result)
        
        return self.results

# Usage
processor = ADLSProcessor(service_client)
file_list = [
    "data/raw/2024/01/01/sales.csv",
    "data/raw/2024/01/02/sales.csv",
    "data/raw/2024/01/03/sales.csv"
]

results = processor.process_files_parallel(file_list, max_workers=3)
print(f"Processed {len(results)} files")

# Optimized upload for large files
def upload_large_file_optimized(local_file_path, remote_file_path, chunk_size=4*1024*1024):
    """
    Upload large files in chunks with progress tracking
    """
    file_client = file_system_client.get_file_client(remote_file_path)
    
    # Get file size
    file_size = os.path.getsize(local_file_path)
    
    print(f"Uploading {local_file_path} ({file_size:,} bytes) to {remote_file_path}")
    
    with open(local_file_path, 'rb') as file:
        # Create the file
        file_client.create_file()
        
        uploaded_bytes = 0
        chunk_number = 0
        
        while True:
            chunk = file.read(chunk_size)
            if not chunk:
                break
            
            # Append chunk
            file_client.append_data(chunk, offset=uploaded_bytes)
            uploaded_bytes += len(chunk)
            chunk_number += 1
            
            # Progress update
            progress = (uploaded_bytes / file_size) * 100
            print(f"Progress: {progress:.1f}% ({uploaded_bytes:,}/{file_size:,} bytes)")
        
        # Flush to finalize
        file_client.flush_data(uploaded_bytes)
    
    print(f"Upload completed: {remote_file_path}")

# Usage
# upload_large_file_optimized("large_dataset.csv", "data/raw/2024/01/large_dataset.csv")

import logging
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('adls_operations.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

def logged_file_operation(operation_name, file_path, operation_func):
    """
    Wrapper function to log file operations
    """
    start_time = datetime.now()
    logger.info(f"Starting {operation_name} for {file_path}")
    
    try:
        result = operation_func()
        duration = (datetime.now() - start_time).total_seconds()
        logger.info(f"Completed {operation_name} for {file_path} in {duration:.2f}s")
        return result
    
    except Exception as e:
        duration = (datetime.now() - start_time).total_seconds()
        logger.error(f"Failed {operation_name} for {file_path} after {duration:.2f}s: {str(e)}")
        raise

# Usage example
def upload_with_logging(local_path, remote_path):
    def upload_operation():
        with open(local_path, 'rb') as data:
            file_client = file_system_client.get_file_client(remote_path)
            return file_client.upload_data(data, overwrite=True)
    
    return logged_file_operation("upload", remote_path, upload_operation)

# Set appropriate access tiers for cost optimization
def optimize_file_access_tiers():
    """
    Move old files to cooler storage tiers to reduce costs
    """
    from datetime import datetime, timedelta
    
    # Define tier thresholds
    hot_threshold = timedelta(days=30)    # Keep recent files in hot tier
    cool_threshold = timedelta(days=90)   # Move to cool after 30 days
    archive_threshold = timedelta(days=365) # Archive after 1 year
    
    now = datetime.now()
    
    # List all files in the data directory
    directory_client = file_system_client.get_directory_client("data")
    paths = directory_client.get_paths(recursive=True)
    
    tier_changes = {'hot': 0, 'cool': 0, 'archive': 0}
    
    for path in paths:
        if not path.is_directory:
            file_age = now - path.last_modified.replace(tzinfo=None)
            
            # Determine appropriate tier
            if file_age > archive_threshold:
                target_tier = 'Archive'
            elif file_age > cool_threshold:
                target_tier = 'Cool'
            else:
                target_tier = 'Hot'
            
            # Note: Actual tier change would require Blob Storage client
            # This is a demonstration of the logic
            tier_changes[target_tier.lower()] += 1
            
            print(f"File: {path.name}, Age: {file_age.days} days, Recommended tier: {target_tier}")
    
    return tier_changes

# Run optimization analysis
tier_recommendations = optimize_file_access_tiers()
print(f"Tier recommendations: {tier_recommendations}")

# Use managed identity in production
credential = ManagedIdentityCredential()

# Implement least privilege access
# Set specific permissions for different user groups
# Use Azure AD groups for access management

# Use appropriate chunk sizes for large files
OPTIMAL_CHUNK_SIZE = 4 * 1024 * 1024  # 4MB

# Implement parallel processing for multiple files
# Use connection pooling for high-throughput scenarios
# Cache frequently accessed metadata

# Implement lifecycle policies
# Use appropriate access tiers
# Monitor and optimize data transfer costs
# Implement data compression where appropriate

from azure.core.exceptions import ResourceNotFoundError, ServiceRequestError

def robust_file_operation(file_path, operation):
    max_retries = 3
    retry_count = 0
    
    while retry_count < max_retries:
        try:
            return operation()
        
        except ResourceNotFoundError:
            logger.error(f"File not found: {file_path}")
            raise
        
        except ServiceRequestError as e:
            retry_count += 1
            if retry_count >= max_retries:
                logger.error(f"Service error after {max_retries} retries: {str(e)}")
                raise
            
            logger.warning(f"Service error, retrying ({retry_count}/{max_retries}): {str(e)}")
            time.sleep(2 ** retry_count)  # Exponential backoff