Azure Devops

keyring>=24.0.0

#!/usr/bin/env python3
"""
Shared HTTP client for Azure DevOps REST API.
All domain scripts import api_request from this module.
Supports both OAuth (Bearer) and PAT (Basic) auth transparently.
"""

import json
import urllib.request
import urllib.error
import urllib.parse
from typing import Optional, Union

from auth import get_organization, get_auth_header


def _get_base_url(organization: str, area: Optional[str] = None) -> str:
    """Build base URL based on API area."""
    org = urllib.parse.quote(organization, safe="")
    if area == "vssps":
        return f"https://vssps.dev.azure.com/{org}"
    elif area == "almsearch":
        return f"https://almsearch.dev.azure.com/{org}"
    elif area == "advsec":
        return f"https://advsec.dev.azure.com/{org}"
    else:
        return f"https://dev.azure.com/{org}"


def encode_path(path: str) -> str:
    """URL-encode each segment of an API path while preserving / and $ delimiters."""
    parts = path.split("/")
    encoded = []
    for part in parts:
        if part.startswith("$"):
            # Work item type prefix like $User Story -> $User%20Story
            encoded.append("$" + urllib.parse.quote(part[1:], safe=""))
        else:
            encoded.append(urllib.parse.quote(part, safe="@"))
    return "/".join(encoded)


def api_request(
    method: str,
    path: str,
    data: Optional[Union[dict, list]] = None,
    params: Optional[dict] = None,
    api_version: str = "7.1",
    content_type: str = "application/json",
    area: Optional[str] = None,
    raw_body: Optional[bytes] = None,
    extra_headers: Optional[dict] = None,
) -> dict:
    """
    Make an authenticated request to the Azure DevOps REST API.

    Args:
        method: HTTP method (GET, POST, PATCH, PUT, DELETE)
        path: API path - will be URL-encoded automatically
        data: JSON body (dict or list for JSON Patch)
        params: Query string parameters
        api_version: API version (default 7.1)
        content_type: Content-Type header
        area: API area for different base URLs (vssps, almsearch, advsec)
        raw_body: Raw bytes body (overrides data if set)
        extra_headers: Additional headers to include (e.g., If-Match)

    Returns:
        Parsed JSON response dict, or {"error": "..."} on failure.
        For non-JSON responses, returns {"text": "...", "_continuation_token": "..."}.
    """
    organization = get_organization()
    auth_header = get_auth_header()

    if not organization or not auth_header:
        return {"error": "Not authenticated. Run: python auth.py login"}

    base_url = _get_base_url(organization, area)
    url = f"{base_url}/{encode_path(path)}"

    # Add api-version to params
    if params is None:
        params = {}
    params["api-version"] = api_version

    url += "?" + urllib.parse.urlencode(params)

    headers = {
        "Authorization": auth_header,
        "Content-Type": content_type,
        "Accept": "application/json",
    }
    if extra_headers:
        headers.update(extra_headers)

    if raw_body is not None:
        body = raw_body
    elif data is not None:
        body = json.dumps(data).encode("utf-8")
    else:
        body = None

    try:
        req = urllib.request.Request(url, data=body, headers=headers, method=method)
        with urllib.request.urlopen(req, timeout=30) as response:
            resp_data = response.read().decode("utf-8")
            ct = response.headers.get("Content-Type", "")
            continuation = response.headers.get("x-ms-continuationtoken", "")

            if not resp_data:
                result = {"status": "success"}
            elif "application/json" in ct:
                result = json.loads(resp_data)
            else:
                # Non-JSON response (e.g., build log text)
                result = {"text": resp_data}

            if continuation:
                result["_continuation_token"] = continuation
            return result
    except urllib.error.HTTPError as e:
        error_body = e.read().decode("utf-8") if e.fp else str(e)
        try:
            error_json = json.loads(error_body)
            msg = error_json.get("message", error_body)
        except (json.JSONDecodeError, AttributeError):
            msg = error_body[:500] if len(error_body) > 500 else error_body
        return {"error": f"HTTP {e.code}: {msg}"}
    except urllib.error.URLError as e:
        return {"error": f"Request failed: {e.reason}"}
    except json.JSONDecodeError:
        return {"error": "Invalid JSON response"}


def api_request_paged(
    path: str,
    params: Optional[dict] = None,
    api_version: str = "7.1",
    area: Optional[str] = None,
    top: Optional[int] = None,
) -> dict:
    """
    Make a paginated GET request, following continuation tokens.
    Checks both response body and x-ms-continuationtoken header.
    """
    if params is None:
        params = {}
    if top is not None:
        params["$top"] = str(top)

    all_items = []
    continuation_token = None

    while True:
        if continuation_token:
            params["continuationToken"] = continuation_token

        result = api_request("GET", path, params=params, api_version=api_version, area=area)

        if "error" in result:
            return result

        items = result.get("value", [])
        all_items.extend(items)

        # Check continuation token from both body and header
        continuation_token = (
            result.get("continuationToken")
            or result.get("_continuation_token")
        )
        if not continuation_token:
            break

        if top and len(all_items) >= top:
            all_items = all_items[:top]
            break

    return {"value": all_items, "count": len(all_items)}

#!/usr/bin/env python3
"""
Azure DevOps work item attachment operations: upload, attach, download, list, remove.
"""

import argparse
import json
import os
import sys
import urllib.error
import urllib.request

from api_client import api_request
from auth import get_auth_header


def upload(project, file_path, filename=None):
    """Upload an attachment and return the attachment URL."""
    if not os.path.isfile(file_path):
        return {"error": f"File not found: {file_path}"}
    with open(file_path, "rb") as f:
        file_bytes = f.read()
    filename = filename or os.path.basename(file_path)
    return api_request(
        "POST",
        f"{project}/_apis/wit/attachments",
        params={"fileName": filename},
        raw_body=file_bytes,
        content_type="application/octet-stream",
    )


def attach(project, work_item_id, url, comment=None):
    """Attach an already-uploaded file to a work item."""
    patch = [
        {
            "op": "add",
            "path": "/relations/-",
            "value": {
                "rel": "AttachedFile",
                "url": url,
                "attributes": {"comment": comment or ""},
            },
        }
    ]
    return api_request(
        "PATCH",
        f"{project}/_apis/wit/workitems/{work_item_id}",
        data=patch,
        content_type="application/json-patch+json",
    )


def upload_and_attach(project, work_item_id, file_path, comment=None):
    """Upload a file and attach it to a work item in one step."""
    result = upload(project, file_path)
    if "error" in result:
        return result
    attachment_url = result.get("url")
    if not attachment_url:
        return {"error": "Upload succeeded but no URL returned"}
    return attach(project, work_item_id, attachment_url, comment)


def list_attachments(project, work_item_id):
    """List attachments on a work item."""
    result = api_request(
        "GET",
        f"{project}/_apis/wit/workitems/{work_item_id}",
        params={"$expand": "relations"},
    )
    if "error" in result:
        return result
    relations = result.get("relations") or []
    attachments = []
    for i, r in enumerate(relations):
        if r.get("rel") == "AttachedFile":
            r["_relationIndex"] = i
            attachments.append(r)
    return {"attachments": attachments}


_ALLOWED_HOSTS = (".dev.azure.com", ".visualstudio.com")


def download(project, url, output_path):
    """Download an attachment to a local file."""
    # Validate URL to prevent credential leakage to non-Azure hosts
    from urllib.parse import urlparse
    parsed = urlparse(url)
    if not any(parsed.hostname and parsed.hostname.endswith(h) for h in _ALLOWED_HOSTS):
        return {"error": f"URL host must be *.dev.azure.com or *.visualstudio.com, got: {parsed.hostname}"}

    auth_header = get_auth_header()
    if not auth_header:
        return {"error": "Not authenticated. Run: python auth.py login"}

    separator = "&" if "?" in url else "?"
    download_url = f"{url}{separator}api-version=7.1"

    req = urllib.request.Request(download_url, method="GET")
    req.add_header("Authorization", auth_header)

    try:
        with urllib.request.urlopen(req, timeout=30) as response:
            with open(output_path, "wb") as f:
                while True:
                    chunk = response.read(8192)
                    if not chunk:
                        break
                    f.write(chunk)
        size = os.path.getsize(output_path)
        return {"status": "downloaded", "path": output_path, "size": size}
    except urllib.error.HTTPError as e:
        return {"error": f"HTTP {e.code}: {e.read().decode('utf-8', errors='replace')[:500]}"}
    except urllib.error.URLError as e:
        return {"error": f"Request failed: {e.reason}"}


def remove(project, work_item_id, index):
    """Remove an attachment from a work item by relation index."""
    patch = [{"op": "remove", "path": f"/relations/{index}"}]
    return api_request(
        "PATCH",
        f"{project}/_apis/wit/workitems/{work_item_id}",
        data=patch,
        content_type="application/json-patch+json",
    )


def main():
    parser = argparse.ArgumentParser(description="Azure DevOps work item attachment operations")
    subparsers = parser.add_subparsers(dest="command", required=True)

    # upload
    p = subparsers.add_parser("upload", help="Upload an attachment")
    p.add_argument("--project", required=True)
    p.add_argument("--file", required=True, dest="file_path")
    p.add_argument("--filename", help="Override the filename")

    # attach
    p = subparsers.add_parser("attach", help="Attach an uploaded file to a work item")
    p.add_argument("--project", required=True)
    p.add_argument("--id", required=True, type=int, dest="work_item_id")
    p.add_argument("--url", required=True)
    p.add_argument("--comment")

    # upload-and-attach
    p = subparsers.add_parser("upload-and-attach", help="Upload and attach a file to a work item")
    p.add_argument("--project", required=True)
    p.add_argument("--id", required=True, type=int, dest="work_item_id")
    p.add_argument("--file", required=True, dest="file_path")
    p.add_argument("--comment")

    # list
    p = subparsers.add_parser("list", help="List attachments on a work item")
    p.add_argument("--project", required=True)
    p.add_argument("--id", required=True, type=int, dest="work_item_id")

    # download
    p = subparsers.add_parser("download", help="Download an attachment")
    p.add_argument("--project", required=True)
    p.add_argument("--url", required=True)
    p.add_argument("--output", required=True, dest="output_path")

    # remove
    p = subparsers.add_parser("remove", help="Remove an attachment from a work item")
    p.add_argument("--project", required=True)
    p.add_argument("--id", required=True, type=int, dest="work_item_id")
    p.add_argument("--index", required=True, type=int)

    args = parser.parse_args()
    result = _dispatch(args)

    print(json.dumps(result, indent=2))
    if isinstance(result, dict) and "error" in result:
        sys.exit(1)


def _dispatch(args):
    if args.command == "upload":
        return upload(args.project, args.file_path, args.filename)
    elif args.command == "attach":
        return attach(args.project, args.work_item_id, args.url, args.comment)
    elif args.command == "upload-and-attach":
        return upload_and_attach(args.project, args.work_item_id, args.file_path, args.comment)
    elif args.command == "list":
        return list_attachments(args.project, args.work_item_id)
    elif args.command == "download":
        return download(args.project, args.url, args.output_path)
    elif args.command == "remove":
        return remove(args.project, args.work_item_id, args.index)
    return {"error": f"Unknown command: {args.command}"}


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Authentication management for Azure DevOps API.
Supports two methods:
  - OAuth (device code flow via Azure AD) - recommended
  - PAT (Personal Access Token) - fallback

Credentials stored in system keyring. Tokens are split across separate
keyring entries to stay under Windows Credential Manager's 2560-byte limit.
"""

import argparse
import base64
import json
import keyring
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

KEYCHAIN_SERVICE = "azure-devops-skill"
KEYCHAIN_ACCOUNT = "default"
# Separate keyring entries for large tokens (Windows Credential Manager has a
# 2560-byte limit per entry, so we split tokens into their own entries).
KEYCHAIN_ACCOUNT_ACCESS_TOKEN = f"{KEYCHAIN_ACCOUNT}_access_token"
KEYCHAIN_ACCOUNT_REFRESH_TOKEN = f"{KEYCHAIN_ACCOUNT}_refresh_token"

# Azure AD OAuth configuration
# Uses the Visual Studio Code client ID - shows "Visual Studio Code" prompt
AZURE_AD_CLIENT_ID = "aebc6443-996d-45c2-90f0-388ff96faa56"
AZURE_DEVOPS_RESOURCE = "499b84ac-1321-427f-aa17-267ca6975798"
AZURE_AD_SCOPE = f"{AZURE_DEVOPS_RESOURCE}/.default offline_access"
AZURE_AD_AUTHORITY = "https://login.microsoftonline.com/organizations"
DEVICE_CODE_ENDPOINT = f"{AZURE_AD_AUTHORITY}/oauth2/v2.0/devicecode"
TOKEN_ENDPOINT = f"{AZURE_AD_AUTHORITY}/oauth2/v2.0/token"

TOKEN_EXPIRY_BUFFER_S = 300  # 5 minutes


# ─── Storage ────────────────────────────────────────────────────────────────

def get_stored_config():
    """Get stored config from keyring.

    Reassembles the full config from the main entry (metadata) and
    separate token entries (access_token, refresh_token).
    """
    try:
        data_str = keyring.get_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT)
        if not data_str:
            return None
        config = json.loads(data_str)

        # Load tokens from their separate entries
        access_token = keyring.get_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT_ACCESS_TOKEN)
        if access_token:
            config["access_token"] = access_token

        refresh_token = keyring.get_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT_REFRESH_TOKEN)
        if refresh_token:
            config["refresh_token"] = refresh_token

        return config
    except json.JSONDecodeError as e:
        print(f"Failed to parse credentials from keyring: {e}", file=sys.stderr)
        return None
    except Exception as e:
        print(f"Failed to load credentials from keyring: {e}", file=sys.stderr)
        return None


def save_config(config):
    """Save config to keyring.

    Splits tokens into separate keyring entries so each stays under
    Windows Credential Manager's 2560-byte limit.
    """
    try:
        # Build metadata dict without tokens (avoids mutating caller's dict)
        access_token = config.get("access_token")
        refresh_token = config.get("refresh_token")
        metadata = {k: v for k, v in config.items() if k not in ("access_token", "refresh_token")}

        # Save metadata (org, auth_type, expires_at, pat) in main entry
        keyring.set_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT, json.dumps(metadata))

        # Save tokens in their own entries
        if access_token:
            keyring.set_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT_ACCESS_TOKEN, access_token)
        if refresh_token:
            keyring.set_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT_REFRESH_TOKEN, refresh_token)

        return True
    except Exception as e:
        print(f"Failed to store credentials in keyring: {e}", file=sys.stderr)
        return False


def clear_config():
    """Clear stored credentials from keyring, including split token entries."""
    cleared = False
    for account in (KEYCHAIN_ACCOUNT, KEYCHAIN_ACCOUNT_ACCESS_TOKEN, KEYCHAIN_ACCOUNT_REFRESH_TOKEN):
        try:
            keyring.delete_password(KEYCHAIN_SERVICE, account)
            cleared = True
        except Exception:
            pass
    return cleared


# ─── OAuth Device Code Flow ─────────────────────────────────────────────────

def _http_post_form(url, data):
    """POST form-encoded data, return parsed JSON."""
    body = urllib.parse.urlencode(data).encode("utf-8")
    req = urllib.request.Request(url, data=body, method="POST")
    req.add_header("Content-Type", "application/x-www-form-urlencoded")
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.loads(resp.read().decode("utf-8"))


def start_device_code_flow():
    """Start OAuth device code flow. Returns device code response."""
    try:
        return _http_post_form(DEVICE_CODE_ENDPOINT, {
            "client_id": AZURE_AD_CLIENT_ID,
            "scope": AZURE_AD_SCOPE,
        })
    except (urllib.error.HTTPError, urllib.error.URLError) as e:
        error_body = ""
        if hasattr(e, "read"):
            error_body = e.read().decode("utf-8", errors="replace")
        print(f"Failed to start device code flow: {e}\n{error_body}", file=sys.stderr)
        return None


def poll_for_token(device_code, interval=5, expires_in=900):
    """Poll for token after user authorizes the device code."""
    deadline = time.time() + expires_in
    while time.time() < deadline:
        time.sleep(interval)
        try:
            result = _http_post_form(TOKEN_ENDPOINT, {
                "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
                "client_id": AZURE_AD_CLIENT_ID,
                "device_code": device_code,
            })
            # Success - got tokens
            return result
        except urllib.error.HTTPError as e:
            body = e.read().decode("utf-8", errors="replace")
            try:
                err = json.loads(body)
            except json.JSONDecodeError:
                print(f"Unexpected error: {body}", file=sys.stderr)
                return None
            error_code = err.get("error", "")
            if error_code == "authorization_pending":
                continue
            elif error_code == "slow_down":
                interval += 5
                continue
            elif error_code == "expired_token":
                print("Device code expired. Please try again.", file=sys.stderr)
                return None
            else:
                print(f"Auth error: {err.get('error_description', error_code)}", file=sys.stderr)
                return None
        except urllib.error.URLError as e:
            print(f"Network error: {e.reason}", file=sys.stderr)
            return None
    print("Timed out waiting for authorization.", file=sys.stderr)
    return None


def refresh_access_token(refresh_token):
    """Refresh an OAuth access token."""
    try:
        return _http_post_form(TOKEN_ENDPOINT, {
            "grant_type": "refresh_token",
            "client_id": AZURE_AD_CLIENT_ID,
            "refresh_token": refresh_token,
            "scope": AZURE_AD_SCOPE,
        })
    except (urllib.error.HTTPError, urllib.error.URLError) as e:
        print(f"Token refresh failed: {e}", file=sys.stderr)
        return None


def perform_oauth_login(org):
    """Full OAuth device code login flow."""
    dc = start_device_code_flow()
    if not dc:
        return None

    print(f"\nTo sign in, open: {dc['verification_uri']}", file=sys.stderr, flush=True)
    print(f"Enter code: {dc['user_code']}\n", file=sys.stderr, flush=True)
    print("Waiting for authorization...", file=sys.stderr, flush=True)

    tokens = poll_for_token(
        dc["device_code"],
        interval=dc.get("interval", 5),
        expires_in=dc.get("expires_in", 900),
    )
    if not tokens:
        return None

    config = {
        "organization": org,
        "auth_type": "oauth",
        "access_token": tokens["access_token"],
        "refresh_token": tokens.get("refresh_token", ""),
        "expires_at": int(time.time()) + tokens.get("expires_in", 3600),
    }
    return config


# ─── PAT Auth ───────────────────────────────────────────────────────────────

def validate_pat(organization, pat):
    """Validate PAT by calling the projects API."""
    url = f"https://dev.azure.com/{organization}/_apis/projects?$top=1&api-version=7.1"
    creds = base64.b64encode(f":{pat}".encode()).decode()
    headers = {"Authorization": f"Basic {creds}"}
    try:
        req = urllib.request.Request(url, headers=headers)
        with urllib.request.urlopen(req, timeout=15) as resp:
            json.loads(resp.read().decode())
            return True
    except (urllib.error.HTTPError, urllib.error.URLError):
        return False


# ─── Public API (used by api_client.py) ─────────────────────────────────────

def get_organization():
    """Get stored organization name."""
    config = get_stored_config()
    return config.get("organization") if config else None


def get_auth_header():
    """Get the Authorization header value. Handles token refresh for OAuth."""
    config = get_stored_config()
    if not config:
        return None

    auth_type = config.get("auth_type", "pat")

    if auth_type == "oauth":
        # Check if token needs refresh
        expires_at = config.get("expires_at", 0)
        if time.time() > (expires_at - TOKEN_EXPIRY_BUFFER_S):
            refresh_token = config.get("refresh_token")
            if refresh_token:
                print("Refreshing access token...", file=sys.stderr)
                new_tokens = refresh_access_token(refresh_token)
                if new_tokens:
                    config["access_token"] = new_tokens["access_token"]
                    config["expires_at"] = int(time.time()) + new_tokens.get("expires_in", 3600)
                    if new_tokens.get("refresh_token"):
                        config["refresh_token"] = new_tokens["refresh_token"]
                    save_config(config)
                else:
                    print("Token refresh failed. Please re-login.", file=sys.stderr)
                    return None
            else:
                print("Token expired and no refresh token. Please re-login.", file=sys.stderr)
                return None
        return f"Bearer {config['access_token']}"

    elif auth_type == "pat":
        pat = config.get("pat", "")
        creds = base64.b64encode(f":{pat}".encode()).decode()
        return f"Basic {creds}"

    return None


# Keep backward compat for any code using get_pat directly
def get_pat():
    """Get stored PAT (for PAT auth mode only)."""
    config = get_stored_config()
    if config and config.get("auth_type", "pat") == "pat":
        return config.get("pat")
    return None


# ─── CLI ────────────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(description="Azure DevOps authentication")
    subparsers = parser.add_subparsers(dest="command")

    login_parser = subparsers.add_parser("login", help="Authenticate (OAuth device code flow)")
    login_parser.add_argument("--org", help="Organization name")
    login_parser.add_argument("--pat", help="Use PAT instead of OAuth")

    subparsers.add_parser("logout", help="Clear stored credentials")
    subparsers.add_parser("status", help="Check authentication status")
    subparsers.add_parser("token", help="Print current access token")

    args = parser.parse_args()

    if args.command == "login":
        org = args.org
        if not org:
            org = input("Organization name: ").strip()
        if not org:
            print("Organization name is required.", file=sys.stderr)
            sys.exit(1)

        if args.pat:
            # PAT auth
            print("Validating PAT...", file=sys.stderr)
            if not validate_pat(org, args.pat):
                print("Invalid credentials.", file=sys.stderr)
                sys.exit(1)
            config = {"organization": org, "auth_type": "pat", "pat": args.pat}
            if save_config(config):
                print(f"Login successful (PAT)! Organization: {org}")
            else:
                print("Failed to store credentials.", file=sys.stderr)
                sys.exit(1)
        else:
            # OAuth device code flow
            config = perform_oauth_login(org)
            if not config:
                print("OAuth login failed.", file=sys.stderr)
                sys.exit(1)
            if save_config(config):
                print(f"Login successful (OAuth)! Organization: {org}")
                print("Credentials stored in system keyring.")
            else:
                print("Failed to store credentials.", file=sys.stderr)
                sys.exit(1)

    elif args.command == "logout":
        if clear_config():
            print("Logged out successfully.")
        else:
            print("No credentials to clear.")

    elif args.command == "token":
        config = get_stored_config()
        if not config:
            print("Not authenticated. Run: python auth.py login", file=sys.stderr)
            sys.exit(1)
        auth_type = config.get("auth_type", "pat")
        if auth_type == "oauth":
            header = get_auth_header()
            if header:
                # Print just the token part
                print(header.split(" ", 1)[1])
            else:
                print("Token expired. Run: python auth.py login", file=sys.stderr)
                sys.exit(1)
        else:
            print(config.get("pat", ""))

    elif args.command == "status":
        config = get_stored_config()
        if not config:
            print("Status: Not authenticated")
            print("Run: python auth.py login")
            sys.exit(1)

        auth_type = config.get("auth_type", "pat")
        org = config.get("organization", "unknown")
        print(f"Status: Authenticated")
        print(f"Organization: {org}")
        print(f"Auth type: {auth_type}")
        print(f"Storage: System keyring")

        if auth_type == "oauth":
            expires_at = config.get("expires_at", 0)
            remaining = expires_at - time.time()
            if remaining > 0:
                print(f"Token expires in: {remaining / 60:.1f} minutes")
                print(f"Auto-refresh: Yes (has refresh token: {bool(config.get('refresh_token'))})")
            else:
                print(f"Token expired: Yes (will auto-refresh on next request)")
        elif auth_type == "pat":
            valid = validate_pat(org, config.get("pat", ""))
            print(f"PAT valid: {valid}")

    else:
        parser.print_help()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Azure DevOps core operations: projects, teams, identities.
"""

import argparse
import json
import sys

from api_client import api_request, api_request_paged


def list_projects(top=None):
    """List all projects in the organization."""
    return api_request_paged("_apis/projects", top=top)


def list_teams(project):
    """List teams in a project."""
    return api_request_paged(f"_apis/projects/{project}/teams")


def get_identity(search):
    """Search for a user identity."""
    return api_request(
        "GET",
        "_apis/identities",
        params={"searchFilter": "General", "filterValue": search},
        area="vssps",
    )


def main():
    parser = argparse.ArgumentParser(description="Azure DevOps core operations")
    subparsers = parser.add_subparsers(dest="command", required=True)

    lp = subparsers.add_parser("list-projects", help="List all projects")
    lp.add_argument("--top", type=int, help="Max results")

    lt = subparsers.add_parser("list-teams", help="List teams in a project")
    lt.add_argument("--project", required=True, help="Project name or ID")

    gi = subparsers.add_parser("get-identity", help="Search for a user identity")
    gi.add_argument("--search", required=True, help="Search string (name or email)")

    args = parser.parse_args()

    if args.command == "list-projects":
        result = list_projects(args.top)
    elif args.command == "list-teams":
        result = list_teams(args.project)
    elif args.command == "get-identity":
        result = get_identity(args.search)
    else:
        result = {"error": f"Unknown command: {args.command}"}

    print(json.dumps(result, indent=2))
    if isinstance(result, dict) and "error" in result:
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Azure DevOps environment and approval operations.
"""

import argparse
import json
import sys

from api_client import api_request


def list_environments(project, top=None):
    """List environments in a project."""
    params = {}
    if top:
        params["$top"] = str(top)
    return api_request("GET", f"{project}/_apis/pipelines/environments", params=params)


def get_environment(project, env_id):
    """Get environment details."""
    return api_request("GET", f"{project}/_apis/pipelines/environments/{env_id}")


def create_environment(project, name, description=None):
    """Create an environment."""
    data = {"name": name}
    if description:
        data["description"] = description
    return api_request("POST", f"{project}/_apis/pipelines/environments", data=data)


def delete_environment(project, env_id):
    """Delete an environment."""
    return api_request("DELETE", f"{project}/_apis/pipelines/environments/{env_id}")


def list_checks(project, env_id):
    """List approval/check configurations for an environment."""
    params = {"resourceType": "environment", "resourceId": str(env_id)}
    return api_request("GET", f"{project}/_apis/pipelines/checks/configurations", params=params, api_version="7.1-preview.1")


def list_approvals(project, state="pending"):
    """List pending approvals."""
    params = {}
    if state != "all":
        params["state"] = state
    return api_request("GET", f"{project}/_apis/pipelines/approvals", params=params, api_version="7.1-preview.1")


def approve_approval(project, approval_id, comment=None):
    """Approve a pending approval."""
    item = {"approvalId": approval_id, "status": "approved"}
    if comment:
        item["comment"] = comment
    return api_request("PATCH", f"{project}/_apis/pipelines/approvals", data=[item], api_version="7.1-preview.1")


def reject_approval(project, approval_id, comment=None):
    """Reject a pending approval."""
    item = {"approvalId": approval_id, "status": "rejected"}
    if comment:
        item["comment"] = comment
    return api_request("PATCH", f"{project}/_apis/pipelines/approvals", data=[item], api_version="7.1-preview.1")


def main():
    parser = argparse.ArgumentParser(description="Azure DevOps environment and approval operations")
    subparsers = parser.add_subparsers(dest="command", required=True)

    # list
    p = subparsers.add_parser("list", help="List environments")
    p.add_argument("--project", required=True)
    p.add_argument("--top", type=int)

    # get
    p = subparsers.add_parser("get", help="Get environment details")
    p.add_argument("--project", required=True)
    p.add_argument("--id", required=True, type=int, dest="env_id")

    # create
    p = subparsers.add_parser("create", help="Create an environment")
    p.add_argument("--project", required=True)
    p.add_argument("--name", required=True)
    p.add_argument("--description")

    # delete
    p = subparsers.add_parser("delete", help="Delete an environment")
    p.add_argument("--project", required=True)
    p.add_argument("--id", required=True, type=int, dest="env_id")

    # list-checks
    p = subparsers.add_parser("list-checks", help="List checks for an environment")
    p.add_argument("--project", required=True)
    p.add_argument("--id", required=True, type=int, dest="env_id")

    # list-approvals
    p = subparsers.add_parser("list-approvals", help="List pending approvals")
    p.add_argument("--project", required=True)
    p.add_argument("--state", choices=["pending", "approved", "rejected", "all"], default="pending")

    # approve
    p = subparsers.add_parser("approve", help="Approve a pending approval")
    p.add_argument("--project", required=True)
    p.add_argument("--approval-id", required=True)
    p.add_argument("--comment")

    # reject
    p = subparsers.add_parser("reject", help="Reject a pending approval")
    p.add_argument("--project", required=True)
    p.add_argument("--approval-id", required=True)
    p.add_argument("--comment")

    args = parser.parse_args()
    result = _dispatch(args)

    print(json.dumps(result, indent=2))
    if isinstance(result, dict) and "error" in result:
        sys.exit(1)


def _dispatch(args):
    if args.command == "list":
        return list_environments(args.project, args.top)
    elif args.command == "get":
        return get_environment(args.project, args.env_id)
    elif args.command == "create":
        return create_environment(args.project, args.name, args.description)
    elif args.command == "delete":
        return delete_environment(args.project, args.env_id)
    elif args.command == "list-checks":
        return list_checks(args.project, args.env_id)
    elif args.command == "list-approvals":
        return list_approvals(args.project, args.state)
    elif args.command == "approve":
        return approve_approval(args.project, args.approval_id, args.comment)
    elif args.command == "reject":
        return reject_approval(args.project, args.approval_id, args.comment)
    return {"error": f"Unknown command: {args.command}"}


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Azure DevOps build/pipeline operations.
"""

import argparse
import json
import sys

from api_client import api_request


def create_pipeline(project, name, repo_id, yaml_path, folder=None):
    """Create a new pipeline."""
    data = {
        "name": name,
        "configuration": {
            "type": "yaml",
            "path": yaml_path,
            "repository": {
                "id": repo_id,
                "type": "azureReposGit",
            },
        },
    }
    if folder:
        data["folder"] = folder
    return api_request("POST", f"{project}/_apis/pipelines", data=data)


def list_builds(project, top=None, status=None, definition_id=None, branch=None):
    """List builds."""
    params = {}
    if top:
        params["$top"] = str(top)
    if status:
        params["statusFilter"] = status
    if definition_id:
        params["definitions"] = str(definition_id)
    if branch:
        params["branchName"] = f"refs/heads/{branch}" if not branch.startswith("refs/") else branch
    return api_request("GET", f"{project}/_apis/build/builds", params=params)


def get_build(project, build_id):
    """Get build details."""
    return api_request("GET", f"{project}/_apis/build/builds/{build_id}")


def build_logs(project, build_id, log_id=None):
    """Get build logs."""
    path = f"{project}/_apis/build/builds/{build_id}/logs"
    if log_id:
        path += f"/{log_id}"
    return api_request("GET", path)


def build_changes(project, build_id):
    """Get changes associated with a build."""
    return api_request("GET", f"{project}/_apis/build/builds/{build_id}/changes")


def list_definitions(project, top=None, name=None):
    """List build/pipeline definitions."""
    params = {}
    if top:
        params["$top"] = str(top)
    if name:
        params["name"] = name
    return api_request("GET", f"{project}/_apis/build/definitions", params=params)


def run_pipeline(project, pipeline_id, branch=None, variables=None):
    """Run a pipeline."""
    data = {}
    if branch:
        ref = f"refs/heads/{branch}" if not branch.startswith("refs/") else branch
        data["resources"] = {"repositories": {"self": {"refName": ref}}}
    if variables:
        data["variables"] = {k: {"value": v} for k, v in variables.items()}
    return api_request("POST", f"{project}/_apis/pipelines/{pipeline_id}/runs", data=data)


def get_run(project, pipeline_id, run_id):
    """Get a pipeline run."""
    return api_request("GET", f"{project}/_apis/pipelines/{pipeline_id}/runs/{run_id}")


def list_runs(project, pipeline_id, top=None):
    """List pipeline runs."""
    params = {}
    if top:
        params["$top"] = str(top)
    return api_request("GET", f"{project}/_apis/pipelines/{pipeline_id}/runs", params=params)


def update_stage(project, build_id, stage_ref_name, state, force_retry=False):
    """Update a build stage (approve/retry)."""
    data = {"state": state, "forceRetryAllJobs": force_retry}
    return api_request(
        "PATCH",
        f"{project}/_apis/build/builds/{build_id}/stages/{stage_ref_name}",
        data=data,
    )


def get_artifacts(project, build_id):
    """Get build artifacts."""
    return api_request("GET", f"{project}/_apis/build/builds/{build_id}/artifacts")


def queue_build(project, definition_id, branch=None, parameters=None):
    """Queue a build."""
    data = {"definition": {"id": definition_id}}
    if branch:
        data["sourceBranch"] = f"refs/heads/{branch}" if not branch.startswith("refs/") else branch
    if parameters:
        data["parameters"] = json.dumps(parameters)
    return api_request("POST", f"{project}/_apis/build/builds", data=data)


def cancel_build(project, build_id):
    """Cancel a running build."""
    return api_request(
        "PATCH",
        f"{project}/_apis/build/builds/{build_id}",
        data={"status": "cancelling"},
    )


def build_status(project, definition_id=None, branch=None):
    """Get the latest build status."""
    params = {"$top": "1", "queryOrder": "finishTimeDescending"}
    if definition_id:
        params["definitions"] = str(definition_id)
    if branch:
        params["branchName"] = f"refs/heads/{branch}" if not branch.startswith("refs/") else branch
    return api_request("GET", f"{project}/_apis/build/builds", params=params)


def main():
    parser = argparse.ArgumentParser(description="Azure DevOps pipeline/build operations")
    subparsers = parser.add_subparsers(dest="command", required=True)

    # create
    p = subparsers.add_parser("create", help="Create a pipeline")
    p.add_argument("--project", required=True)
    p.add_argument("--name", required=True)
    p.add_argument("--repo-id", required=True)
    p.add_argument("--yaml-path", required=True, help="Path to YAML file in repo")
    p.add_argument("--folder")

    # list-builds
    p = subparsers.add_parser("list-builds", help="List builds")
    p.add_argument("--project", required=True)
    p.add_argument("--top", type=int)
    p.add_argument("--status", choices=["all", "cancelling", "completed", "inProgress", "none", "notStarted", "postponed"])
    p.add_argument("--definition-id", type=int)
    p.add_argument("--branch")

    # get-build
    p = subparsers.add_parser("get-build", help="Get build details")
    p.add_argument("--project", required=True)
    p.add_argument("--build-id", required=True, type=int)

    # build-logs
    p = subparsers.add_parser("build-logs", help="Get build logs")
    p.add_argument("--project", required=True)
    p.add_argument("--build-id", required=True, type=int)
    p.add_argument("--log-id", type=int, help="Specific log ID")

    # build-changes
    p = subparsers.add_parser("build-changes", help="Get build changes")
    p.add_argument("--project", required=True)
    p.add_argument("--build-id", required=True, type=int)

    # list-definitions
    p = subparsers.add_parser("list-definitions", help="List pipeline definitions")
    p.add_argument("--project", required=True)
    p.add_argument("--top", type=int)
    p.add_argument("--name", help="Filter by name")

    # run
    p = subparsers.add_parser("run", help="Run a pipeline")
    p.add_argument("--project", required=True)
    p.add_argument("--pipeline-id", required=True, type=int)
    p.add_argument("--branch")
    p.add_argument("--variable", action="append", nargs=2, metavar=("KEY", "VALUE"))

    # get-run
    p = subparsers.add_parser("get-run", help="Get pipeline run details")
    p.add_argument("--project", required=True)
    p.add_argument("--pipeline-id", required=True, type=int)
    p.add_argument("--run-id", required=True, type=int)

    # list-runs
    p = subparsers.add_parser("list-runs", help="List pipeline runs")
    p.add_argument("--project", required=True)
    p.add_argument("--pipeline-id", required=True, type=int)
    p.add_argument("--top", type=int)

    # update-stage
    p = subparsers.add_parser("update-stage", help="Update a build stage")
    p.add_argument("--project", required=True)
    p.add_argument("--build-id", required=True, type=int)
    p.add_argument("--stage", required=True, help="Stage reference name")
    p.add_argument("--state", required=True, choices=["cancel", "retry"])
    p.add_argument("--force-retry", action="store_true")

    # get-artifacts
    p = subparsers.add_parser("get-artifacts", help="Get build artifacts")
    p.add_argument("--project", required=True)
    p.add_argument("--build-id", required=True, type=int)

    # queue-build
    p = subparsers.add_parser("queue-build", help="Queue a build")
    p.add_argument("--project", required=True)
    p.add_argument("--definition-id", required=True, type=int)
    p.add_argument("--branch")
    p.add_argument("--parameter", action="append", nargs=2, metavar=("KEY", "VALUE"))

    # cancel-build
    p = subparsers.add_parser("cancel-build", help="Cancel a build")
    p.add_argument("--project", required=True)
    p.add_argument("--build-id", required=True, type=int)

    # build-status
    p = subparsers.add_parser("build-status", help="Get latest build status")
    p.add_argument("--project", required=True)
    p.add_argument("--definition-id", type=int)
    p.add_argument("--branch")

    args = parser.parse_args()
    result = _dispatch(args)

    print(json.dumps(result, indent=2))
    if isinstance(result, dict) and "error" in result:
        sys.exit(1)


def _dispatch(args):
    if args.command == "create":
        return create_pipeline(args.project, args.name, args.repo_id, args.yaml_path, args.folder)
    elif args.command == "list-builds":
        return list_builds(args.project, args.top, args.status, args.definition_id, args.branch)
    elif args.command == "get-build":
        return get_build(args.project, args.build_id)
    elif args.command == "build-logs":
        return build_logs(args.project, args.build_id, args.log_id)
    elif args.command == "build-changes":
        return build_changes(args.project, args.build_id)
    elif args.command == "list-definitions":
        return list_definitions(args.project, args.top, args.name)
    elif args.command == "run":
        variables = dict(args.variable) if args.variable else None
        return run_pipeline(args.project, args.pipeline_id, args.branch, variables)
    elif args.command == "get-run":
        return get_run(args.project, args.pipeline_id, args.run_id)
    elif args.command == "list-runs":
        return list_runs(args.project, args.pipeline_id, args.top)
    elif args.command == "update-stage":
        return update_stage(args.project, args.build_id, args.stage, args.state, args.force_retry)
    elif args.command == "get-artifacts":
        return get_artifacts(args.project, args.build_id)
    elif args.command == "queue-build":
        parameters = dict(args.parameter) if args.parameter else None
        return queue_build(args.project, args.definition_id, args.branch, parameters)
    elif args.command == "cancel-build":
        return cancel_build(args.project, args.build_id)
    elif args.command == "build-status":
        return build_status(args.project, args.definition_id, args.branch)
    return {"error": f"Unknown command: {args.command}"}


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Azure DevOps branch policy operations: list, create, update, delete policies.
"""

import argparse
import json
import sys

from api_client import api_request


def list_policies(project, branch=None, repo=None):
    """List all branch policies in a project."""
    params = {}
    if branch:
        ref_name = branch if branch.startswith("refs/") else f"refs/heads/{branch}"
        params["scope[0].refName"] = ref_name
    if repo:
        params["scope[0].repositoryId"] = repo
    return api_request("GET", f"{project}/_apis/policy/configurations", params=params)


def get_policy(project, policy_id):
    """Get a specific policy configuration."""
    return api_request("GET", f"{project}/_apis/policy/configurations/{policy_id}")


def list_types(project):
    """List available policy types."""
    return api_request("GET", f"{project}/_apis/policy/types")


def create_min_reviewers(project, repo_id, branch, min_reviewers,
                         creator_vote_counts=False, allow_downvotes=False, reset_on_push=False):
    """Create a minimum reviewers policy."""
    ref_name = branch if branch.startswith("refs/") else f"refs/heads/{branch}"
    data = {
        "isEnabled": True,
        "isBlocking": True,
        "type": {"id": "fa4e907d-c16b-4a4c-9dfa-4906e5d171dd"},
        "settings": {
            "minimumApproverCount": min_reviewers,
            "creatorVoteCounts": creator_vote_counts,
            "allowDownvotes": allow_downvotes,
            "resetOnSourcePush": reset_on_push,
            "scope": [{"refName": ref_name, "matchKind": "Exact", "repositoryId": repo_id}],
        },
    }
    return api_request("POST", f"{project}/_apis/policy/configurations", data=data)


def create_build_policy(project, repo_id, branch, build_definition_id,
                        display_name="Build Validation", valid_duration=720):
    """Create a build validation policy."""
    ref_name = branch if branch.startswith("refs/") else f"refs/heads/{branch}"
    data = {
        "isEnabled": True,
        "isBlocking": True,
        "type": {"id": "0609b952-1397-4640-95ec-e00a01b2c241"},
        "settings": {
            "buildDefinitionId": build_definition_id,
            "displayName": display_name,
            "validDuration": valid_duration,
            "scope": [{"refName": ref_name, "matchKind": "Exact", "repositoryId": repo_id}],
        },
    }
    return api_request("POST", f"{project}/_apis/policy/configurations", data=data)


def update_policy(project, policy_id, enabled=None, blocking=None):
    """Update a policy (enable/disable/change blocking)."""
    existing = get_policy(project, policy_id)
    if "error" in existing:
        return existing
    if enabled is not None:
        existing["isEnabled"] = enabled == "true"
    if blocking is not None:
        existing["isBlocking"] = blocking == "true"
    return api_request("PUT", f"{project}/_apis/policy/configurations/{policy_id}", data=existing)


def delete_policy(project, policy_id):
    """Delete a policy."""
    return api_request("DELETE", f"{project}/_apis/policy/configurations/{policy_id}")


def list_evaluations(project, pr_id):
    """List policy evaluations for a pull request."""
    # Resolve project GUID if a name was passed (artifactId requires GUID)
    proj_info = api_request("GET", f"_apis/projects/{project}")
    if "error" in proj_info:
        return proj_info
    project_id = proj_info.get("id", project)
    artifact_id = f"vstfs:///CodeReview/CodeReviewId/{project_id}/{pr_id}"
    return api_request("GET", f"{project}/_apis/policy/evaluations", params={"artifactId": artifact_id})


def main():
    parser = argparse.ArgumentParser(description="Azure DevOps branch policy operations")
    subparsers = parser.add_subparsers(dest="command", required=True)

    # list
    p = subparsers.add_parser("list", help="List branch policies")
    p.add_argument("--project", required=True)
    p.add_argument("--branch", help="Filter by branch name (e.g., main)")
    p.add_argument("--repo", help="Filter by repository ID or name")

    # get
    p = subparsers.add_parser("get", help="Get a policy configuration")
    p.add_argument("--project", required=True)
    p.add_argument("--id", required=True, type=int, dest="policy_id")

    # list-types
    p = subparsers.add_parser("list-types", help="List available policy types")
    p.add_argument("--project", required=True)

    # create-min-reviewers
    p = subparsers.add_parser("create-min-reviewers", help="Create minimum reviewers policy")
    p.add_argument("--project", required=True)
    p.add_argument("--repo-id", required=True, help="Repository ID")
    p.add_argument("--branch", required=True, help="Branch to protect (e.g., main)")
    p.add_argument("--min-reviewers", required=True, type=int, help="Minimum number of reviewers")
    p.add_argument("--creator-vote-counts", action="store_true", help="Whether creator's vote counts")
    p.add_argument("--allow-downvotes", action="store_true", help="Allow completion with rejections")
    p.add_argument("--reset-on-push", action="store_true", help="Reset votes on new push")

    # create-build-policy
    p = subparsers.add_parser("create-build-policy", help="Create build validation policy")
    p.add_argument("--project", required=True)
    p.add_argument("--repo-id", required=True, help="Repository ID")
    p.add_argument("--branch", required=True, help="Branch to protect (e.g., main)")
    p.add_argument("--build-definition-id", required=True, type=int, help="Build definition ID")
    p.add_argument("--display-name", default="Build Validation", help="Display name for the policy")
    p.add_argument("--valid-duration", type=int, default=720, help="Minutes build is valid (default: 720)")

    # update
    p = subparsers.add_parser("update", help="Update a policy")
    p.add_argument("--project", required=True)
    p.add_argument("--id", required=True, type=int, dest="policy_id")
    p.add_argument("--enabled", choices=["true", "false"], help="Enable or disable the policy")
    p.add_argument("--blocking", choices=["true", "false"], help="Set policy as blocking or optional")

    # delete
    p = subparsers.add_parser("delete", help="Delete a policy")
    p.add_argument("--project", required=True)
    p.add_argument("--id", required=True, type=int, dest="policy_id")

    # list-evaluations
    p = subparsers.add_parser("list-evaluations", help="List policy evaluations for a PR")
    p.add_argument("--project", required=True)
    p.add_argument("--pr-id", required=True, type=int)

    args = parser.parse_args()
    result = _dispatch(args)

    print(json.dumps(result, indent=2))
    if isinstance(result, dict) and "error" in result:
        sys.exit(1)


def _dispatch(args):
    if args.command == "list":
        return list_policies(args.project, args.branch, args.repo)
    elif args.command == "get":
        return get_policy(args.project, args.policy_id)
    elif args.command == "list-types":
        return list_types(args.project)
    elif args.command == "create-min-reviewers":
        return create_min_reviewers(args.project, args.repo_id, args.branch, args.min_reviewers,
                                    args.creator_vote_counts, args.allow_downvotes, args.reset_on_push)
    elif args.command == "create-build-policy":
        return create_build_policy(args.project, args.repo_id, args.branch, args.build_definition_id,
                                   args.display_name, args.valid_duration)
    elif args.command == "update":
        return update_policy(args.project, args.policy_id, args.enabled, args.blocking)
    elif args.command == "delete":
        return delete_policy(args.project, args.policy_id)
    elif args.command == "list-evaluations":
        return list_evaluations(args.project, args.pr_id)
    return {"error": f"Unknown command: {args.command}"}


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Azure DevOps Git repository operations: repos, branches, PRs.
"""

import argparse
import json
import sys

from api_client import api_request


def list_repos(project):
    """List all repositories in a project."""
    return api_request("GET", f"{project}/_apis/git/repositories")


def get_repo(project, repo):
    """Get a repository by name or ID."""
    return api_request("GET", f"{project}/_apis/git/repositories/{repo}")


def list_branches(project, repo):
    """List branches in a repository."""
    return api_request("GET", f"{project}/_apis/git/repositories/{repo}/refs", params={"filter": "heads/"})


def create_branch(project, repo, name, source_branch="main"):
    """Create a new branch from a source branch."""
    # First get the source branch ref
    refs = api_request("GET", f"{project}/_apis/git/repositories/{repo}/refs",
                       params={"filter": f"heads/{source_branch}"})
    if "error" in refs:
        return refs
    values = refs.get("value", [])
    if not values:
        return {"error": f"Source branch '{source_branch}' not found"}
    source_object_id = values[0].get("objectId")

    return api_request("POST", f"{project}/_apis/git/repositories/{repo}/refs", data=[
        {
            "name": f"refs/heads/{name}",
            "oldObjectId": "0000000000000000000000000000000000000000",
            "newObjectId": source_object_id,
        }
    ])


def search_commits(project, repo, item_path=None, author=None, top=None):
    """Search commits in a repository."""
    params = {}
    if item_path:
        params["searchCriteria.itemPath"] = item_path
    if author:
        params["searchCriteria.author"] = author
    if top:
        params["$top"] = str(top)
    return api_request("GET", f"{project}/_apis/git/repositories/{repo}/commits", params=params)


def list_prs(project, repo=None, status="active", top=None, creator=None, reviewer=None):
    """List pull requests."""
    params = {"searchCriteria.status": status}
    if creator:
        params["searchCriteria.creatorId"] = creator
    if reviewer:
        params["searchCriteria.reviewerId"] = reviewer
    if top:
        params["$top"] = str(top)
    path = f"{project}/_apis/git/repositories/{repo}/pullrequests" if repo else f"{project}/_apis/git/pullrequests"
    return api_request("GET", path, params=params)


def get_pr(project, repo, pr_id):
    """Get a pull request by ID."""
    return api_request("GET", f"{project}/_apis/git/repositories/{repo}/pullrequests/{pr_id}")


def create_pr(project, repo, source_branch, target_branch, title, description=None, draft=False, reviewers=None):
    """Create a pull request."""
    data = {
        "sourceRefName": f"refs/heads/{source_branch}" if not source_branch.startswith("refs/") else source_branch,
        "targetRefName": f"refs/heads/{target_branch}" if not target_branch.startswith("refs/") else target_branch,
        "title": title,
        "isDraft": draft,
    }
    if description:
        data["description"] = description
    if reviewers:
        data["reviewers"] = [{"id": r} for r in reviewers]
    return api_request("POST", f"{project}/_apis/git/repositories/{repo}/pullrequests", data=data)


def update_pr(project, repo, pr_id, fields):
    """Update a pull request."""
    return api_request("PATCH", f"{project}/_apis/git/repositories/{repo}/pullrequests/{pr_id}", data=fields)


def list_reviewers(project, repo, pr_id):
    """List reviewers on a pull request."""
    return api_request("GET", f"{project}/_apis/git/repositories/{repo}/pullrequests/{pr_id}/reviewers")


def add_reviewer(project, repo, pr_id, reviewer_id, vote=0):
    """Add a reviewer to a pull request."""
    return api_request(
        "PUT",
        f"{project}/_apis/git/repositories/{repo}/pullrequests/{pr_id}/reviewers/{reviewer_id}",
        data={"vote": vote},
    )


def list_threads(project, repo, pr_id):
    """List comment threads on a pull request."""
    return api_request("GET", f"{project}/_apis/git/repositories/{repo}/pullrequests/{pr_id}/threads")


def create_thread(project, repo, pr_id, content, file_path=None, line=None, status="active"):
    """Create a comment thread on a pull request."""
    data = {
        "comments": [{"content": content, "commentType": "text"}],
        "status": status,
    }
    if file_path:
        data["threadContext"] = {
            "filePath": file_path,
            "rightFileStart": {"line": line or 1, "offset": 1},
            "rightFileEnd": {"line": line or 1, "offset": 1},
        }
    return api_request("POST", f"{project}/_apis/git/repositories/{repo}/pullrequests/{pr_id}/threads", data=data)


def add_thread_comment(project, repo, pr_id, thread_id, content):
    """Add a comment to an existing thread."""
    return api_request(
        "POST",
        f"{project}/_apis/git/repositories/{repo}/pullrequests/{pr_id}/threads/{thread_id}/comments",
        data={"content": content, "commentType": "text"},
    )


def complete_pr(project, repo, pr_id, merge_strategy="squash", delete_source=True):
    """Complete (merge) a pull request."""
    pr = get_pr(project, repo, pr_id)
    if "error" in pr:
        return pr
    last_merge = pr.get("lastMergeSourceCommit", {}).get("commitId", "")
    data = {
        "status": "completed",
        "lastMergeSourceCommit": {"commitId": last_merge},
        "completionOptions": {
            "mergeStrategy": merge_strategy,
            "deleteSourceBranch": delete_source,
        },
    }
    return api_request("PATCH", f"{project}/_apis/git/repositories/{repo}/pullrequests/{pr_id}", data=data)


def abandon_pr(project, repo, pr_id):
    """Abandon a pull request."""
    return api_request(
        "PATCH",
        f"{project}/_apis/git/repositories/{repo}/pullrequests/{pr_id}",
        data={"status": "abandoned"},
    )


def get_diff(project, repo, base_version, target_version, base_version_type="branch", target_version_type="branch"):
    """Get diff between two versions."""
    params = {
        "baseVersion": base_version,
        "baseVersionType": base_version_type,
        "targetVersion": target_version,
        "targetVersionType": target_version_type,
    }
    return api_request("GET", f"{project}/_apis/git/repositories/{repo}/diffs/commits", params=params)


def list_files(project, repo, path="/", branch=None):
    """List files/folders in a repository path."""
    params = {"scopePath": path, "recursionLevel": "OneLevel"}
    if branch:
        params["versionDescriptor.version"] = branch
        params["versionDescriptor.versionType"] = "branch"
    return api_request("GET", f"{project}/_apis/git/repositories/{repo}/items", params=params)


def main():
    parser = argparse.ArgumentParser(description="Azure DevOps Git operations")
    subparsers = parser.add_subparsers(dest="command", required=True)

    # list
    p = subparsers.add_parser("list", help="List repositories")
    p.add_argument("--project", required=True)

    # get
    p = subparsers.add_parser("get", help="Get a repository")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)

    # list-branches
    p = subparsers.add_parser("list-branches", help="List branches")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)

    # create-branch
    p = subparsers.add_parser("create-branch", help="Create a branch")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--name", required=True, help="New branch name")
    p.add_argument("--source", default="main", help="Source branch (default: main)")

    # search-commits
    p = subparsers.add_parser("search-commits", help="Search commits")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--path", help="Item path filter", dest="item_path")
    p.add_argument("--author", help="Author filter")
    p.add_argument("--top", type=int)

    # list-prs
    p = subparsers.add_parser("list-prs", help="List pull requests")
    p.add_argument("--project", required=True)
    p.add_argument("--repo")
    p.add_argument("--status", default="active", choices=["active", "completed", "abandoned", "all"])
    p.add_argument("--top", type=int)
    p.add_argument("--creator")
    p.add_argument("--reviewer")

    # get-pr
    p = subparsers.add_parser("get-pr", help="Get a pull request")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--pr-id", required=True, type=int)

    # create-pr
    p = subparsers.add_parser("create-pr", help="Create a pull request")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--source", required=True, help="Source branch")
    p.add_argument("--target", required=True, help="Target branch")
    p.add_argument("--title", required=True)
    p.add_argument("--description")
    p.add_argument("--draft", action="store_true")
    p.add_argument("--reviewer", action="append", dest="reviewers", help="Reviewer ID (repeatable)")

    # update-pr
    p = subparsers.add_parser("update-pr", help="Update a pull request")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--pr-id", required=True, type=int)
    p.add_argument("--title")
    p.add_argument("--description")
    p.add_argument("--status", choices=["active", "completed", "abandoned"])

    # list-reviewers
    p = subparsers.add_parser("list-reviewers", help="List PR reviewers")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--pr-id", required=True, type=int)

    # add-reviewer
    p = subparsers.add_parser("add-reviewer", help="Add a PR reviewer")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--pr-id", required=True, type=int)
    p.add_argument("--reviewer-id", required=True)
    p.add_argument("--vote", type=int, default=0, help="Vote: 10=approve, 5=approve-with-suggestions, 0=no-vote, -5=wait, -10=reject")

    # list-threads
    p = subparsers.add_parser("list-threads", help="List PR comment threads")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--pr-id", required=True, type=int)

    # create-thread
    p = subparsers.add_parser("create-thread", help="Create a PR comment thread")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--pr-id", required=True, type=int)
    p.add_argument("--content", required=True)
    p.add_argument("--file-path", help="File to comment on")
    p.add_argument("--line", type=int, help="Line number")
    p.add_argument("--status", default="active", choices=["active", "fixed", "wontFix", "closed", "byDesign", "pending"])

    # add-thread-comment
    p = subparsers.add_parser("add-thread-comment", help="Reply to a PR thread")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--pr-id", required=True, type=int)
    p.add_argument("--thread-id", required=True, type=int)
    p.add_argument("--content", required=True)

    # complete-pr
    p = subparsers.add_parser("complete-pr", help="Complete (merge) a PR")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--pr-id", required=True, type=int)
    p.add_argument("--merge-strategy", default="squash", choices=["squash", "noFastForward", "rebase", "rebaseMerge"])
    p.add_argument("--keep-source", action="store_true", help="Don't delete source branch")

    # abandon-pr
    p = subparsers.add_parser("abandon-pr", help="Abandon a PR")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--pr-id", required=True, type=int)

    # get-diff
    p = subparsers.add_parser("get-diff", help="Get diff between versions")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--base", required=True, help="Base version (commit/branch)")
    p.add_argument("--target", required=True, help="Target version (commit/branch)")
    p.add_argument("--base-version-type", default="branch", choices=["branch", "commit", "tag"])
    p.add_argument("--target-version-type", default="branch", choices=["branch", "commit", "tag"])

    # list-files
    p = subparsers.add_parser("list-files", help="List files in a repo path")
    p.add_argument("--project", required=True)
    p.add_argument("--repo", required=True)
    p.add_argument("--path", default="/")
    p.add_argument("--branch")

    args = parser.parse_args()
    result = _dispatch(args)

    print(json.dumps(result, indent=2))
    if isinstance(result, dict) and "error" in result:
        sys.exit(1)


def _dispatch(args):
    if args.command == "list":
        return list_repos(args.project)
    elif args.command == "get":
        return get_repo(args.project, args.repo)
    elif args.command == "list-branches":
        return list_branches(args.project, args.repo)
    elif args.command == "create-branch":
        return create_branch(args.project, args.repo, args.name, args.source)
    elif args.command == "search-commits":
        return search_commits(args.project, args.repo, args.item_path, args.author, args.top)
    elif args.command == "list-prs":
        return list_prs(args.project, args.repo, args.status, args.top, args.creator, args.reviewer)
    elif args.command == "get-pr":
        return get_pr(args.project, args.repo, args.pr_id)
    elif args.command == "create-pr":
        return create_pr(args.project, args.repo, args.source, args.target, args.title,
                        args.description, args.draft, args.reviewers)
    elif args.command == "update-pr":
        fields = {}
        if args.title:
            fields["title"] = args.title
        if args.description:
            fields["description"] = args.description
        if args.status:
            fields["status"] = args.status
        return update_pr(args.project, args.repo, args.pr_id, fields)
    elif args.command == "list-reviewers":
        return list_reviewers(args.project, args.repo, args.pr_id)
    elif args.command == "add-reviewer":
        return add_reviewer(args.project, args.repo, args.pr_id, args.reviewer_id, args.vote)
    elif args.command == "list-threads":
        return list_threads(args.project, args.repo, args.pr_id)
    elif args.command == "create-thread":
        return create_thread(args.project, args.repo, args.pr_id, args.content,
                           args.file_path, args.line, args.status)
    elif args.command == "add-thread-comment":
        return add_thread_comment(args.project, args.repo, args.pr_id, args.thread_id, args.content)
    elif args.command == "complete-pr":
        return complete_pr(args.project, args.repo, args.pr_id, args.merge_strategy,
                         not args.keep_source)
    elif args.command == "abandon-pr":
        return abandon_pr(args.project, args.repo, args.pr_id)
    elif args.command == "get-diff":
        return get_diff(args.project, args.repo, args.base, args.target,
                       args.base_version_type, args.target_version_type)
    elif args.command == "list-files":
        return list_files(args.project, args.repo, args.path, args.branch)
    return {"error": f"Unknown command: {args.command}"}


if __name__ == "__main__":
    main()