# durable_functions_orchestrator.py
import azure.functions as func
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    # Step 1: Extract data from source
    raw_data = yield context.call_activity(
        'extract_data',
        {'source': 'sales_api', 'date': context.current_utc_date}
    )
    
    # Step 2: Validate data quality
    validation_result = yield context.call_activity(
        'validate_data',
        {'data': raw_data, 'rules': 'sales_validation_rules'}
    )
    
    if not validation_result['is_valid']:
        # Step 2a: Handle validation failure
        yield context.call_activity(
            'send_alert',
            {'message': f"Data validation failed: {validation_result['errors']}"}
        )
        return {'status': 'failed', 'errors': validation_result['errors']}
    
    # Step 3: Transform data (parallel processing)
    transformed_chunks = yield context.task_all([
        context.call_activity('transform_chunk', {'chunk': chunk})
        for chunk in raw_data['chunks']
    ])
    
    # Step 4: Load to data lake
    load_result = yield context.call_activity(
        'load_to_lake',
        {'data': transformed_chunks, 'zone': 'curated'}
    )
    
    # Step 5: Notify downstream consumers
    yield context.call_activity(
        'notify_consumers',
        {'dataset': 'sales', 'partition': load_result['partition']}
    )
    
    return {'status': 'completed', 'records_processed': load_result['count']}

# Activity functions
def extract_data(input_data):
    import requests
    response = requests.get(f"https://api.example.com/{input_data['source']}")
    return {'records': response.json(), 'chunk_size': 10000}

def validate_data(input_data):
    errors = []
    for record in input_data['data']['records']:
        if not record.get('amount') or record['amount'] < 0:
            errors.append(f"Invalid amount for record {record['id']}")
    return {'is_valid': len(errors) == 0, 'errors': errors}

def transform_chunk(input_data):
    # Apply transformations
    transformed = []
    for record in input_data['chunk']:
        transformed.append({
            'id': record['id'],
            'amount_usd': record['amount'] * record['exchange_rate'],
            'processed_date': context.current_utc_date
        })
    return transformed

# blob_trigger_ingestion.py
import azure.functions as func
import json
import logging
from azure.storage.filedatalake import DataLakeServiceClient
from azure.identity import DefaultAzureCredential

app = func.FunctionApp()

@app.blob_trigger(
    arg_name="myblob",
    path="raw/{folder}/{name}",
    connection="AzureWebJobsStorage"
)
def blob_trigger_ingestion(myblob: func.InputStream):
    logging.info(f"Processing blob: {myblob.name}, Size: {myblob.length}")
    
    # Parse the blob path
    path_parts = myblob.name.split('/')
    date_partition = path_parts[1]  # e.g., "2024/01/15"
    
    # Read blob content
    data = json.loads(myblob.read().decode('utf-8'))
    
    # Process and write to curated zone
    credential = DefaultAzureCredential()
    datalake = DataLakeServiceClient(
        account_url="https://stdatalake001.dfs.core.windows.net",
        credential=credential
    )
    
    # Write processed data
    file_client = datalake.get_file_client(
        "curated",
        f"sales/{date_partition}/processed.json"
    )
    
    processed_data = process_records(data)
    file_client.upload_data(
        json.dumps(processed_data),
        overwrite=True
    )
    
    logging.info(f"Processed {len(processed_data)} records")

def process_records(data):
    return [
        {
            'id': r['id'],
            'amount': float(r['amount']),
            'category': r.get('category', 'unknown'),
            'processed': True
        }
        for r in data['records']
        if r.get('amount') and float(r['amount']) > 0
    ]

# timer_trigger_scheduled.py
import azure.functions as func
import logging
import datetime
import json
from azure.identity import DefaultAzureCredential
from azure.synapse.artifacts import ArtifactsClient

app = func.FunctionApp()

@app.timer_trigger(
    arg_name="myTimer",
    schedule="0 0 2 * * *",  # Daily at 2 AM
    run_on_startup=False,
    use_monitor=False
)
def scheduled_etl(myTimer: func.TimerRequest):
    utc_timestamp = datetime.datetime.utcnow().isoformat()
    
    if myTimer.past_due:
        logging.info("The timer is past due!")
    
    logging.info("Starting daily ETL pipeline at %s", utc_timestamp)
    
    # Trigger ADF pipeline
    credential = DefaultAzureCredential()
    artifacts_client = ArtifactsClient(
        credential=credential,
        endpoint="https://syn-workspace.dev.azuresynapse.net"
    )
    
    # Start pipeline run
    pipeline_run = artifacts_client.pipeline.create_pipeline_run(
        pipeline_name="daily_sales_etl",
        parameters={
            "date": datetime.date.today().isoformat(),
            "source": "sales_api",
            "target": "curated/sales"
        }
    )
    
    logging.info(f"Pipeline run started: {pipeline_run.run_id}")
    
    # Wait for completion (simplified - in production, use webhook or polling)
    import time
    while True:
        run_status = artifacts_client.pipeline_run.get_pipeline_run(
            pipeline_run.run_id
        )
        if run_status.status in ['Succeeded', 'Failed', 'Cancelled']:
            break
        time.sleep(30)
    
    logging.info(f"Pipeline completed with status: {run_status.status}")

# http_trigger_api.py
import azure.functions as func
import json
import logging
from azure.identity import DefaultAzureCredential
from azure.cosmos import CosmosClient

app = func.FunctionApp()

@app.route(route="data/{dataset}", auth_level=func.AuthLevel.FUNCTION)
def data_api(req: func.HttpRequest) -> func.HttpResponse:
    dataset = req.params.get('dataset')
    start_date = req.params.get('start_date')
    end_date = req.params.get('end_date')
    
    if not dataset:
        return func.HttpResponse(
            "Please provide a dataset name",
            status_code=400
        )
    
    # Query Cosmos DB
    credential = DefaultAzureCredential()
    client = CosmosClient(
        "https://cosmos-prod.documents.azure.com:443/",
        credential
    )
    
    database = client.get_database_client("analytics")
    container = database.get_container_client(dataset)
    
    query = f"""
        SELECT * FROM c 
        WHERE c.date >= '{start_date}' 
        AND c.date <= '{end_date}'
    """
    
    items = list(container.query_items(
        query=query,
        enable_cross_partition_query=True
    ))
    
    return func.HttpResponse(
        json.dumps({"records": items, "count": len(items)}),
        status_code=200,
        mimetype="application/json"
    )