Fabric Cli Powerbi

#!/usr/bin/env python3
"""
list_refresh_history.py - Show refresh history and failure details

This script displays the refresh history for a semantic model, including
duration, status, and error details.

Usage:
    python list_refresh_history.py <model> [--last N]
    python list_refresh_history.py Production.Workspace/Sales.SemanticModel --last 10

Exit codes:
    0 - History retrieved successfully
    1 - Failed to retrieve history
"""

import argparse
import json
import subprocess
import sys
from datetime import datetime, timedelta
from typing import Dict, Any, List, Optional


def run_fab_command(args: list[str], timeout: int = 60) -> tuple[int, str, str]:
    """Execute a fab CLI command and return exit code, stdout, stderr."""
    try:
        result = subprocess.run(
            ["fab"] + args,
            capture_output=True,
            text=True,
            timeout=timeout
        )
        return result.returncode, result.stdout, result.stderr
    except FileNotFoundError:
        return -1, "", "fab CLI not found in PATH"
    except subprocess.TimeoutExpired:
        return -2, "", f"Command timed out after {timeout} seconds"
    except Exception as e:
        return -3, "", str(e)


def ensure_semantic_model_suffix(path: str) -> str:
    """Ensure path has .SemanticModel suffix if it's just a model path."""
    # If it already ends with a type suffix, return as is
    known_suffixes = [".SemanticModel", ".Dataset", ".Workspace"]
    for suffix in known_suffixes:
        if path.endswith(suffix):
            return path
    
    # If path contains a workspace but no model type, add SemanticModel
    if "/" in path and not path.split("/")[-1].count("."):
        return f"{path}.SemanticModel"
    
    return path


def check_path_exists(path: str) -> bool:
    """Check if path exists."""
    exit_code, _, _ = run_fab_command(["exists", path])
    return exit_code == 0


def get_refresh_history(model_path: str, limit: int = 10) -> List[Dict[str, Any]]:
    """Get refresh history for a semantic model."""
    # Use the jobs list command to get refresh history
    cmd = ["jobs", "list", model_path, "-t", "Refresh", "-l", str(limit), "-f", "json"]
    
    exit_code, stdout, stderr = run_fab_command(cmd)
    
    if exit_code == 0:
        try:
            result = json.loads(stdout.strip())
            if isinstance(result, list):
                return result
        except json.JSONDecodeError:
            pass
    
    return []


def get_job_details(job_id: str, model_path: str) -> Dict[str, Any]:
    """Get detailed information about a specific job."""
    cmd = ["jobs", "get", model_path, "-j", job_id, "-f", "json"]
    
    exit_code, stdout, stderr = run_fab_command(cmd)
    
    if exit_code == 0:
        try:
            return json.loads(stdout.strip())
        except json.JSONDecodeError:
            pass
    
    return {}


def format_duration(start_time: str, end_time: str) -> str:
    """Format duration between two ISO timestamps."""
    try:
        start = datetime.fromisoformat(start_time.replace("Z", "+00:00"))
        end = datetime.fromisoformat(end_time.replace("Z", "+00:00"))
        duration = end - start
        
        total_seconds = int(duration.total_seconds())
        hours, remainder = divmod(total_seconds, 3600)
        minutes, seconds = divmod(remainder, 60)
        
        if hours > 0:
            return f"{hours}h {minutes}m {seconds}s"
        elif minutes > 0:
            return f"{minutes}m {seconds}s"
        else:
            return f"{seconds}s"
    except:
        return "N/A"


def format_timestamp(timestamp: str) -> str:
    """Format ISO timestamp for display."""
    try:
        dt = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
        return dt.strftime("%Y-%m-%d %H:%M:%S UTC")
    except:
        return timestamp


def analyze_failure_patterns(history: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Analyze refresh history for failure patterns."""
    analysis = {
        "total_refreshes": len(history),
        "successful": 0,
        "failed": 0,
        "cancelled": 0,
        "in_progress": 0,
        "success_rate": 0.0,
        "avg_duration_seconds": 0,
        "common_errors": {},
        "recent_failures": []
    }
    
    durations = []
    
    for refresh in history:
        status = refresh.get("status", "").lower()
        
        if status in ["completed", "succeeded", "success"]:
            analysis["successful"] += 1
            
            # Calculate duration for successful refreshes
            start = refresh.get("startTime", refresh.get("startDateTime"))
            end = refresh.get("endTime", refresh.get("endDateTime"))
            if start and end:
                try:
                    start_dt = datetime.fromisoformat(start.replace("Z", "+00:00"))
                    end_dt = datetime.fromisoformat(end.replace("Z", "+00:00"))
                    durations.append((end_dt - start_dt).total_seconds())
                except:
                    pass
                    
        elif status in ["failed", "error"]:
            analysis["failed"] += 1
            
            # Track error messages
            error_msg = refresh.get("error", refresh.get("errorMessage", "Unknown error"))
            if error_msg:
                analysis["common_errors"][error_msg] = analysis["common_errors"].get(error_msg, 0) + 1
            
            # Track recent failures
            if len(analysis["recent_failures"]) < 5:
                analysis["recent_failures"].append({
                    "time": refresh.get("startTime", refresh.get("startDateTime", "Unknown")),
                    "error": error_msg
                })
                
        elif status in ["cancelled", "canceled"]:
            analysis["cancelled"] += 1
            
        elif status in ["running", "in_progress", "inprogress"]:
            analysis["in_progress"] += 1
    
    # Calculate success rate
    completed = analysis["successful"] + analysis["failed"]
    if completed > 0:
        analysis["success_rate"] = round(analysis["successful"] / completed * 100, 1)
    
    # Calculate average duration
    if durations:
        analysis["avg_duration_seconds"] = round(sum(durations) / len(durations), 1)
    
    return analysis


def print_refresh_table(history: List[Dict[str, Any]]):
    """Print refresh history in a formatted table."""
    if not history:
        print("No refresh history found.")
        return
    
    # Table header
    print("\n{:<6} {:<22} {:<12} {:<12} {}".format(
        "#", "Started", "Duration", "Status", "Details"
    ))
    print("-" * 80)
    
    for i, refresh in enumerate(history, 1):
        # Extract fields
        start_time = refresh.get("startTime", refresh.get("startDateTime", "N/A"))
        end_time = refresh.get("endTime", refresh.get("endDateTime", ""))
        status = refresh.get("status", "Unknown")
        error = refresh.get("error", refresh.get("errorMessage", ""))
        
        # Format values
        start_formatted = format_timestamp(start_time)[:22] if start_time != "N/A" else "N/A"
        duration = format_duration(start_time, end_time) if end_time else "Running"
        
        # Status emoji
        status_display = status
        if status.lower() in ["completed", "succeeded", "success"]:
            status_display = "✓ Success"
        elif status.lower() in ["failed", "error"]:
            status_display = "✗ Failed"
        elif status.lower() in ["running", "in_progress"]:
            status_display = "⟳ Running"
        elif status.lower() in ["cancelled", "canceled"]:
            status_display = "⊘ Cancelled"
        
        # Details (truncate if too long)
        details = error[:30] + "..." if len(error) > 30 else error
        
        print("{:<6} {:<22} {:<12} {:<12} {}".format(
            i, start_formatted, duration, status_display, details
        ))


def list_refresh_history(
    model_path: str,
    limit: int = 10,
    show_details: bool = False
) -> Dict[str, Any]:
    """List refresh history for a semantic model."""
    
    model_path = ensure_semantic_model_suffix(model_path)
    
    report = {
        "model": model_path,
        "limit": limit,
        "history": [],
        "analysis": {}
    }
    
    # Verify model exists
    print(f"\nVerifying model: {model_path}")
    if not check_path_exists(model_path):
        print(f"Error: Model '{model_path}' does not exist or is not accessible")
        return report
    print("  Model found")
    
    # Get refresh history
    print(f"\nFetching refresh history (last {limit})...")
    history = get_refresh_history(model_path, limit)
    report["history"] = history
    
    if not history:
        print("  No refresh history found")
        return report
    
    print(f"  Found {len(history)} refresh record(s)")
    
    # Get additional details if requested
    if show_details:
        print("\nFetching detailed information...")
        for i, refresh in enumerate(history):
            job_id = refresh.get("id", refresh.get("jobId"))
            if job_id:
                details = get_job_details(job_id, model_path)
                if details:
                    history[i].update(details)
    
    # Print table
    print_refresh_table(history)
    
    # Analyze patterns
    print("\nAnalyzing refresh patterns...")
    analysis = analyze_failure_patterns(history)
    report["analysis"] = analysis
    
    return report


def main():
    parser = argparse.ArgumentParser(
        description="Show refresh history and failure details for a semantic model",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
    python list_refresh_history.py Production.Workspace/Sales.SemanticModel
    python list_refresh_history.py Production.Workspace/Sales.SemanticModel --last 20
    python list_refresh_history.py "My Workspace/Revenue Model" --last 5 --details

Output includes:
    - Refresh start time
    - Duration
    - Status (Success, Failed, Running, Cancelled)
    - Error details for failed refreshes
    - Success rate analysis
    - Common error patterns
        """
    )
    parser.add_argument(
        "model",
        help="Semantic model path (Workspace/Model.SemanticModel)"
    )
    parser.add_argument(
        "--last", "-n",
        type=int,
        default=10,
        help="Number of recent refreshes to show (default: 10)"
    )
    parser.add_argument(
        "--details", "-d",
        action="store_true",
        help="Fetch detailed information for each refresh"
    )
    parser.add_argument(
        "--json", "-j",
        action="store_true",
        help="Output results in JSON format"
    )
    
    args = parser.parse_args()
    
    try:
        report = list_refresh_history(
            model_path=args.model,
            limit=args.last,
            show_details=args.details
        )
        
        # Print analysis summary
        if report["analysis"]:
            analysis = report["analysis"]
            print("\n" + "=" * 50)
            print("REFRESH ANALYSIS")
            print("=" * 50)
            print(f"Total Refreshes: {analysis['total_refreshes']}")
            print(f"  Successful: {analysis['successful']}")
            print(f"  Failed: {analysis['failed']}")
            print(f"  Cancelled: {analysis['cancelled']}")
            print(f"  In Progress: {analysis['in_progress']}")
            print(f"\nSuccess Rate: {analysis['success_rate']}%")
            
            if analysis['avg_duration_seconds'] > 0:
                avg_mins = analysis['avg_duration_seconds'] / 60
                print(f"Avg Duration: {avg_mins:.1f} minutes")
            
            if analysis['common_errors']:
                print("\nCommon Errors:")
                for error, count in sorted(analysis['common_errors'].items(), key=lambda x: -x[1])[:3]:
                    print(f"  [{count}x] {error[:60]}...")
            
            print("=" * 50)
        
        if args.json:
            print("\n" + json.dumps(report, indent=2, default=str))
        
        # Exit code: 0 if we got history, 1 if not
        sys.exit(0 if report['history'] else 1)
        
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
rebind_report.py - Rebind report to different semantic model

This script changes the semantic model connection for a Power BI report in
Microsoft Fabric.

Usage:
    python rebind_report.py <report> --model <new-model>
    python rebind_report.py Production.Workspace/SalesReport.Report --model Production.Workspace/NewSales.SemanticModel

Exit codes:
    0 - Report successfully rebound to new model
    1 - Rebind failed or error occurred
"""

import argparse
import json
import subprocess
import sys
import time
from typing import Dict, Any, Optional, Tuple


def run_fab_command(args: list[str], timeout: int = 60) -> tuple[int, str, str]:
    """Execute a fab CLI command and return exit code, stdout, stderr."""
    try:
        result = subprocess.run(
            ["fab"] + args,
            capture_output=True,
            text=True,
            timeout=timeout
        )
        return result.returncode, result.stdout, result.stderr
    except FileNotFoundError:
        return -1, "", "fab CLI not found in PATH"
    except subprocess.TimeoutExpired:
        return -2, "", f"Command timed out after {timeout} seconds"
    except Exception as e:
        return -3, "", str(e)


def parse_json_output(output: str) -> Any:
    """Parse JSON output from fab CLI."""
    try:
        return json.loads(output.strip())
    except json.JSONDecodeError:
        return None


def parse_item_path(path: str, item_type: str) -> Tuple[Optional[str], Optional[str]]:
    """Parse workspace and item name from path."""
    parts = path.split("/")
    if len(parts) != 2:
        return None, None
    
    workspace = parts[0]
    item = parts[1]
    
    # Ensure workspace suffix
    if not workspace.endswith(".Workspace"):
        workspace = f"{workspace}.Workspace"
    
    # Ensure item suffix
    suffix = f".{item_type}"
    if not item.endswith(suffix):
        item = f"{item}{suffix}"
    
    return workspace, item


def check_path_exists(path: str) -> bool:
    """Check if path exists."""
    exit_code, _, _ = run_fab_command(["exists", path])
    return exit_code == 0


def get_item_details(path: str) -> Dict[str, Any]:
    """Get item details."""
    exit_code, stdout, stderr = run_fab_command(["get", path, "-f", "json"])
    
    if exit_code == 0:
        return parse_json_output(stdout) or {"error": "Failed to parse item details"}
    return {"error": stderr.strip() or "Failed to get item details"}


def get_item_id(path: str) -> Optional[str]:
    """Get item ID from path."""
    exit_code, stdout, stderr = run_fab_command(["get", path, "-q", "id", "-f", "json"])
    
    if exit_code == 0:
        result = parse_json_output(stdout)
        if isinstance(result, str):
            return result.strip('"')
        return result
    return None


def get_report_datasource(report_path: str) -> Dict[str, Any]:
    """Get the current datasource configuration for a report."""
    exit_code, stdout, stderr = run_fab_command(["get", report_path, "-f", "json"])
    
    if exit_code == 0:
        details = parse_json_output(stdout) or {}
        return {
            "current_model": details.get("semanticModelId") or details.get("datasetId"),
            "workspace_id": details.get("workspaceId"),
            "details": details
        }
    return {"error": stderr}


def rebind_report(workspace_id: str, report_id: str, new_model_id: str, 
                  new_model_workspace_id: Optional[str] = None) -> Dict[str, Any]:
    """Rebind report to a new semantic model using the API."""
    
    result = {
        "status": "unknown",
        "message": "",
        "report_id": report_id,
        "new_model_id": new_model_id
    }
    
    # Build the API request body
    request_body = {
        "datasetId": new_model_id
    }
    
    # If model is in different workspace, include workspace ID
    if new_model_workspace_id and new_model_workspace_id != workspace_id:
        request_body["datasetWorkspaceId"] = new_model_workspace_id
    
    # Use the API command to rebind
    # POST /groups/{workspaceId}/reports/{reportId}/Rebind
    api_path = f"/groups/{workspace_id}/reports/{report_id}/Rebind"
    
    exit_code, stdout, stderr = run_fab_command([
        "api", "post", api_path,
        "-b", json.dumps(request_body),
        "--api-version", "v1.0",
        "-f", "json"
    ], timeout=60)
    
    if exit_code == 0:
        result["status"] = "success"
        result["message"] = "Report successfully rebound to new semantic model"
    else:
        # Check for specific error conditions
        if "already bound" in stderr.lower() or "same dataset" in stderr.lower():
            result["status"] = "success"
            result["message"] = "Report is already bound to the specified model"
        else:
            result["status"] = "failed"
            result["message"] = stderr.strip() or "Failed to rebind report"
    
    return result


def verify_rebind(report_path: str, expected_model_id: str) -> bool:
    """Verify that rebind was successful."""
    time.sleep(2)  # Give it a moment to propagate
    
    datasource = get_report_datasource(report_path)
    current_model = datasource.get("current_model")
    
    return current_model == expected_model_id


def print_result(result: Dict[str, Any], output_format: str, output_file: Optional[str] = None):
    """Print rebind result."""
    if output_format == "json":
        output = json.dumps(result, indent=2)
    else:
        # Text format
        lines = []
        lines.append("=" * 60)
        lines.append("REPORT REBIND RESULT")
        lines.append("=" * 60)
        lines.append(f"Report: {result.get('report_path', 'Unknown')}")
        lines.append(f"Target Model: {result.get('model_path', 'Unknown')}")
        lines.append("")
        
        status = result.get("status", "unknown")
        if status == "success":
            lines.append("✓ REBIND SUCCESSFUL")
            lines.append(f"  {result.get('message', '')}")
        else:
            lines.append("✗ REBIND FAILED")
            lines.append(f"  {result.get('message', 'Unknown error')}")
        
        if result.get("verified"):
            lines.append("")
            lines.append("✓ Verification: Report is now connected to the new model")
        elif result.get("verified") is False:
            lines.append("")
            lines.append("⚠ Verification: Could not confirm rebind (may need time to propagate)")
        
        if result.get("previous_model"):
            lines.append("")
            lines.append(f"Previous Model ID: {result['previous_model']}")
        
        lines.append("")
        lines.append("=" * 60)
        output = "\n".join(lines)
    
    if output_file:
        with open(output_file, "w", encoding="utf-8") as f:
            f.write(output)
        print(f"Result written to: {output_file}", file=sys.stderr)
    else:
        print(output)


def main():
    parser = argparse.ArgumentParser(
        description="Rebind report to different semantic model",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
    python rebind_report.py Production.Workspace/SalesReport.Report --model Production.Workspace/NewSales.SemanticModel
    python rebind_report.py MyWorkspace/Report --model MyWorkspace/Model -f json
    python rebind_report.py Dev.Workspace/TestReport.Report --model Prod.Workspace/ProdModel.SemanticModel --verify
        """
    )
    
    parser.add_argument(
        "report",
        help="Report path (e.g., 'MyWorkspace.Workspace/MyReport.Report')"
    )
    
    parser.add_argument(
        "--model",
        required=True,
        help="New semantic model path (e.g., 'MyWorkspace.Workspace/MyModel.SemanticModel')"
    )
    
    parser.add_argument(
        "-f", "--format",
        choices=["text", "json"],
        default="text",
        help="Output format (default: text)"
    )
    
    parser.add_argument(
        "-o", "--output",
        help="Output file path (default: stdout)"
    )
    
    parser.add_argument(
        "--verify",
        action="store_true",
        help="Verify rebind was successful after operation"
    )
    
    parser.add_argument(
        "-v", "--verbose",
        action="store_true",
        help="Enable verbose output"
    )
    
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Show what would be done without making changes"
    )
    
    args = parser.parse_args()
    
    # Parse report path
    report_workspace, report_name = parse_item_path(args.report, "Report")
    if not report_workspace or not report_name:
        print(f"Error: Invalid report path: {args.report}", file=sys.stderr)
        print("Expected format: Workspace.Workspace/Report.Report", file=sys.stderr)
        return 1
    
    report_path = f"{report_workspace}/{report_name}"
    
    # Parse model path
    model_workspace, model_name = parse_item_path(args.model, "SemanticModel")
    if not model_workspace or not model_name:
        print(f"Error: Invalid model path: {args.model}", file=sys.stderr)
        print("Expected format: Workspace.Workspace/Model.SemanticModel", file=sys.stderr)
        return 1
    
    model_path = f"{model_workspace}/{model_name}"
    
    if args.verbose:
        print(f"Report: {report_path}", file=sys.stderr)
        print(f"Target Model: {model_path}", file=sys.stderr)
    
    # Check report exists
    if not check_path_exists(report_path):
        print(f"Error: Report does not exist: {report_path}", file=sys.stderr)
        return 1
    
    # Check model exists
    if not check_path_exists(model_path):
        print(f"Error: Semantic model does not exist: {model_path}", file=sys.stderr)
        return 1
    
    # Get current datasource info
    if args.verbose:
        print("Getting current report configuration...", file=sys.stderr)
    
    current_datasource = get_report_datasource(report_path)
    if "error" in current_datasource:
        print(f"Warning: Could not get current datasource: {current_datasource['error']}", file=sys.stderr)
    
    # Get IDs
    if args.verbose:
        print("Getting item IDs...", file=sys.stderr)
    
    # Get report workspace ID and report ID
    report_workspace_id = get_item_id(report_workspace)
    report_id = get_item_id(report_path)
    
    if not report_workspace_id or not report_id:
        print("Error: Could not get report workspace or report ID", file=sys.stderr)
        return 1
    
    # Get model workspace ID and model ID
    model_workspace_id = get_item_id(model_workspace)
    model_id = get_item_id(model_path)
    
    if not model_workspace_id or not model_id:
        print("Error: Could not get model workspace or model ID", file=sys.stderr)
        return 1
    
    if args.verbose:
        print(f"Report Workspace ID: {report_workspace_id}", file=sys.stderr)
        print(f"Report ID: {report_id}", file=sys.stderr)
        print(f"Model Workspace ID: {model_workspace_id}", file=sys.stderr)
        print(f"Model ID: {model_id}", file=sys.stderr)
    
    result = {
        "report_path": report_path,
        "model_path": model_path,
        "report_id": report_id,
        "model_id": model_id,
        "previous_model": current_datasource.get("current_model"),
        "status": "unknown",
        "message": ""
    }
    
    # Dry run mode
    if args.dry_run:
        result["status"] = "dry_run"
        result["message"] = "Would rebind report to specified model (dry run, no changes made)"
        print_result(result, args.format, args.output)
        return 0
    
    # Perform rebind
    if args.verbose:
        print("Rebinding report to new model...", file=sys.stderr)
    
    rebind_result = rebind_report(
        report_workspace_id, 
        report_id, 
        model_id,
        model_workspace_id if model_workspace_id != report_workspace_id else None
    )
    
    result["status"] = rebind_result["status"]
    result["message"] = rebind_result["message"]
    
    # Verify if requested
    if args.verify and rebind_result["status"] == "success":
        if args.verbose:
            print("Verifying rebind...", file=sys.stderr)
        
        verified = verify_rebind(report_path, model_id)
        result["verified"] = verified
    
    # Output result
    print_result(result, args.format, args.output)
    
    # Return exit code
    return 0 if result["status"] == "success" else 1


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
refresh_model.py - Trigger and monitor semantic model refresh

This script triggers a semantic model refresh in Microsoft Fabric and optionally
waits for completion with configurable timeout.

Usage:
    python refresh_model.py <model> [--wait] [--timeout 300]
    python refresh_model.py Production.Workspace/Sales.SemanticModel --wait --timeout 600

Exit codes:
    0 - Refresh completed successfully (or triggered successfully without --wait)
    1 - Refresh failed or error occurred
"""

import argparse
import json
import re
import subprocess
import sys
import time
from typing import Dict, Any, Optional, Tuple


def run_fab_command(args: list[str], timeout: int = 60) -> tuple[int, str, str]:
    """Execute a fab CLI command and return exit code, stdout, stderr."""
    try:
        result = subprocess.run(
            ["fab"] + args,
            capture_output=True,
            text=True,
            timeout=timeout
        )
        return result.returncode, result.stdout, result.stderr
    except FileNotFoundError:
        return -1, "", "fab CLI not found in PATH"
    except subprocess.TimeoutExpired:
        return -2, "", f"Command timed out after {timeout} seconds"
    except Exception as e:
        return -3, "", str(e)


def parse_json_output(output: str) -> Any:
    """Parse JSON output from fab CLI."""
    try:
        return json.loads(output.strip())
    except json.JSONDecodeError:
        return None


def parse_model_path(model_path: str) -> Tuple[Optional[str], Optional[str]]:
    """Parse workspace and model name from path."""
    # Handle format: Workspace.Workspace/Model.SemanticModel or Workspace/Model.SemanticModel
    parts = model_path.split("/")
    if len(parts) != 2:
        return None, None
    
    workspace = parts[0]
    model = parts[1]
    
    # Ensure suffixes
    if not workspace.endswith(".Workspace"):
        workspace = f"{workspace}.Workspace"
    if not model.endswith(".SemanticModel"):
        model = f"{model}.SemanticModel"
    
    return workspace, model


def get_ids_from_path(workspace: str, model: str) -> Tuple[Optional[str], Optional[str]]:
    """Get workspace ID and model ID from names."""
    # Get workspace ID
    exit_code, stdout, stderr = run_fab_command(["get", workspace, "-q", "id", "-f", "json"])
    if exit_code != 0:
        print(f"Error: Failed to get workspace ID: {stderr}", file=sys.stderr)
        return None, None
    
    workspace_id = parse_json_output(stdout)
    if not workspace_id:
        print(f"Error: Could not parse workspace ID", file=sys.stderr)
        return None, None
    
    # Get model ID
    model_path = f"{workspace}/{model}"
    exit_code, stdout, stderr = run_fab_command(["get", model_path, "-q", "id", "-f", "json"])
    if exit_code != 0:
        print(f"Error: Failed to get model ID: {stderr}", file=sys.stderr)
        return None, None
    
    model_id = parse_json_output(stdout)
    if not model_id:
        print(f"Error: Could not parse model ID", file=sys.stderr)
        return None, None
    
    return workspace_id.strip('"'), model_id.strip('"')


def trigger_refresh(workspace_id: str, model_id: str, retry_count: int = 1) -> Dict[str, Any]:
    """Trigger semantic model refresh via Power BI API."""
    result = {
        "status": "unknown",
        "refresh_id": None,
        "polling_endpoint": None,
        "message": ""
    }
    
    # Build request body
    request_body = json.dumps({"retryCount": str(retry_count)})
    
    # Trigger refresh using fab api command
    api_path = f"groups/{workspace_id}/datasets/{model_id}/refreshes"
    exit_code, stdout, stderr = run_fab_command([
        "api", "-A", "powerbi", "-X", "post", api_path,
        "--show_headers", "-i", request_body
    ])
    
    if exit_code != 0:
        result["status"] = "failed"
        result["message"] = stderr.strip() or "Failed to trigger refresh"
        return result
    
    # Parse response
    response = parse_json_output(stdout)
    if not response:
        result["status"] = "failed"
        result["message"] = "Failed to parse API response"
        return result
    
    status_code = response.get("status_code")
    if status_code != 202:
        result["status"] = "failed"
        result["message"] = f"Unexpected status code: {status_code}"
        return result
    
    # Extract polling endpoint from Location header
    headers = response.get("headers", {})
    location = headers.get("Location", "")
    
    if location:
        # Extract the API path from the full URL
        match = re.search(r'https?://[^/]+/v1\.0/myorg/(.+)', location)
        if match:
            result["polling_endpoint"] = match.group(1).strip()
    
    # Fallback to RequestId header
    if not result["polling_endpoint"]:
        request_id = headers.get("RequestId")
        if request_id:
            result["polling_endpoint"] = f"groups/{workspace_id}/datasets/{model_id}/refreshes/{request_id}"
            result["refresh_id"] = request_id
    
    result["status"] = "triggered"
    result["message"] = "Refresh triggered successfully"
    
    return result


def poll_refresh_status(polling_endpoint: str, timeout: int = 300, poll_interval: int = 10) -> Dict[str, Any]:
    """Poll refresh status until completion or timeout."""
    result = {
        "status": "unknown",
        "final_status": None,
        "duration_seconds": 0,
        "message": ""
    }
    
    start_time = time.time()
    
    print(f"\nPolling refresh status (timeout: {timeout}s)...")
    
    while True:
        elapsed = time.time() - start_time
        
        if elapsed > timeout:
            result["status"] = "timeout"
            result["duration_seconds"] = int(elapsed)
            result["message"] = f"Refresh timed out after {timeout} seconds"
            return result
        
        # Get refresh status
        exit_code, stdout, stderr = run_fab_command([
            "api", "-A", "powerbi", polling_endpoint, "--show_headers"
        ])
        
        if exit_code != 0:
            result["status"] = "error"
            result["message"] = stderr.strip() or "Failed to get refresh status"
            return result
        
        response = parse_json_output(stdout)
        if not response:
            result["status"] = "error"
            result["message"] = "Failed to parse status response"
            return result
        
        status_code = response.get("status_code")
        if status_code not in [200, 202]:
            result["status"] = "error"
            result["message"] = f"Unexpected status code: {status_code}"
            return result
        
        # Parse the status text
        status_text = response.get("text", {})
        if isinstance(status_text, str):
            status_text = parse_json_output(status_text) or {}
        
        refresh_status = status_text.get("extendedStatus") or status_text.get("status", "Unknown")
        
        print(f"  Status: {refresh_status} (elapsed: {int(elapsed)}s)")
        
        # Check for completion states
        if refresh_status == "Completed":
            result["status"] = "completed"
            result["final_status"] = refresh_status
            result["duration_seconds"] = int(time.time() - start_time)
            result["message"] = "Refresh completed successfully"
            return result
        
        if refresh_status in ["Failed", "Cancelled", "Disabled", "TimedOut"]:
            result["status"] = "failed"
            result["final_status"] = refresh_status
            result["duration_seconds"] = int(time.time() - start_time)
            result["message"] = f"Refresh ended with status: {refresh_status}"
            
            # Try to get error details
            if "error" in status_text:
                result["error_details"] = status_text["error"]
            
            return result
        
        # Still in progress, wait and poll again
        time.sleep(poll_interval)


def refresh_model(
    model_path: str,
    wait: bool = False,
    timeout: int = 300,
    retry_count: int = 1
) -> Dict[str, Any]:
    """Main function to refresh a semantic model."""
    
    result = {
        "model_path": model_path,
        "wait": wait,
        "timeout": timeout,
        "trigger_result": None,
        "poll_result": None,
        "overall_status": "unknown"
    }
    
    # Parse model path
    workspace, model = parse_model_path(model_path)
    if not workspace or not model:
        result["overall_status"] = "error"
        result["message"] = f"Invalid model path: {model_path}. Expected format: Workspace/Model.SemanticModel"
        return result
    
    print(f"\nModel: {workspace}/{model}")
    
    # Get IDs
    print("Resolving workspace and model IDs...")
    workspace_id, model_id = get_ids_from_path(workspace, model)
    if not workspace_id or not model_id:
        result["overall_status"] = "error"
        result["message"] = "Failed to resolve workspace or model ID"
        return result
    
    result["workspace_id"] = workspace_id
    result["model_id"] = model_id
    
    # Trigger refresh
    print("Triggering refresh...")
    trigger_result = trigger_refresh(workspace_id, model_id, retry_count)
    result["trigger_result"] = trigger_result
    
    if trigger_result["status"] != "triggered":
        result["overall_status"] = "failed"
        result["message"] = trigger_result["message"]
        return result
    
    print(f"✓ {trigger_result['message']}")
    
    if not wait:
        result["overall_status"] = "triggered"
        result["message"] = "Refresh triggered. Use --wait to monitor completion."
        return result
    
    # Poll for completion
    if not trigger_result.get("polling_endpoint"):
        result["overall_status"] = "warning"
        result["message"] = "Refresh triggered but no polling endpoint available"
        return result
    
    poll_result = poll_refresh_status(
        trigger_result["polling_endpoint"],
        timeout=timeout
    )
    result["poll_result"] = poll_result
    
    if poll_result["status"] == "completed":
        result["overall_status"] = "completed"
        result["message"] = f"Refresh completed in {poll_result['duration_seconds']} seconds"
    else:
        result["overall_status"] = "failed"
        result["message"] = poll_result["message"]
    
    return result


def main():
    parser = argparse.ArgumentParser(
        description="Trigger and monitor semantic model refresh",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
    python refresh_model.py Production.Workspace/Sales.SemanticModel
    python refresh_model.py Production/Sales.SemanticModel --wait
    python refresh_model.py Dev.Workspace/Model.SemanticModel --wait --timeout 600
    python refresh_model.py "My Workspace/My Model.SemanticModel" --wait --json

Note:
    Without --wait, the script triggers the refresh and exits immediately.
    With --wait, the script polls until completion or timeout.
        """
    )
    parser.add_argument(
        "model",
        help="Semantic model path (format: Workspace/Model.SemanticModel)"
    )
    parser.add_argument(
        "--wait", "-w",
        action="store_true",
        help="Wait for refresh to complete"
    )
    parser.add_argument(
        "--timeout", "-t",
        type=int,
        default=300,
        help="Timeout in seconds when waiting (default: 300)"
    )
    parser.add_argument(
        "--retry-count", "-r",
        type=int,
        default=1,
        help="Number of retries for the refresh operation (default: 1)"
    )
    parser.add_argument(
        "--json", "-j",
        action="store_true",
        help="Output results in JSON format"
    )
    
    args = parser.parse_args()
    
    try:
        result = refresh_model(
            model_path=args.model,
            wait=args.wait,
            timeout=args.timeout,
            retry_count=args.retry_count
        )
        
        # Print summary
        print("\n" + "=" * 50)
        print("REFRESH RESULT")
        print("=" * 50)
        print(f"Model: {result['model_path']}")
        print(f"Status: {result['overall_status'].upper()}")
        print(f"Message: {result.get('message', 'N/A')}")
        
        if result.get("poll_result"):
            print(f"Duration: {result['poll_result'].get('duration_seconds', 'N/A')} seconds")
        
        print("=" * 50)
        
        if args.json:
            print("\n" + json.dumps(result, indent=2))
        
        # Exit code based on status
        success_statuses = ["completed", "triggered"]
        sys.exit(0 if result["overall_status"] in success_statuses else 1)
        
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()