export ADO_ORG="https://dev.azure.com/my-org"
export ADO_PROJECT="MyProject"
export ADO_PAT="<personal-access-token>"

import requests, base64, os

pat = os.environ["ADO_PAT"]
token = base64.b64encode(f":{pat}".encode()).decode()
HEADERS = {
    "Authorization": f"Basic {token}",
    "Content-Type": "application/json-patch+json",
    "Accept": "application/json",
}
BASE = os.environ["ADO_ORG"]
PROJECT = os.environ["ADO_PROJECT"]

def create_bug(title, description_html, priority, severity, tags=""):
    body = [
        {"op": "add", "path": "/fields/System.Title", "value": title},
        {"op": "add", "path": "/fields/System.Description",
         "value": description_html},
        {"op": "add", "path": "/fields/Microsoft.VSTS.Common.Priority",
         "value": priority},
        {"op": "add", "path": "/fields/Microsoft.VSTS.Common.Severity",
         "value": severity},
        {"op": "add", "path": "/fields/System.Tags", "value": tags},
    ]
    r = requests.post(
        f"{BASE}/{PROJECT}/_apis/wit/workitems/$Bug?api-version=7.1",
        json=body, headers=HEADERS,
    )
    r.raise_for_status()
    item = r.json()
    return item["id"], item["url"]

def set_state(work_item_id, target_state, current_rev=None):
    body = []
    if current_rev is not None:
        body.append({"op": "test", "path": "/rev", "value": current_rev})
    body.append(
        {"op": "add", "path": "/fields/System.State", "value": target_state}
    )
    r = requests.patch(
        f"{BASE}/{PROJECT}/_apis/wit/workitems/{work_item_id}?api-version=7.1",
        json=body, headers=HEADERS,
    )
    r.raise_for_status()
    return r.json()["fields"]["System.State"]

body += [
    {"op": "add", "path": "/fields/Microsoft.VSTS.Common.ResolvedReason",
     "value": "Fixed"},
    {"op": "add", "path": "/fields/System.History",
     "value": "Resolved in PR #1234"},
]

def wiql_search(query, top=50):
    r = requests.post(
        f"{BASE}/{PROJECT}/_apis/wit/wiql?$top={top}&api-version=7.1",
        json={"query": query},
        headers={**HEADERS, "Content-Type": "application/json"},
    )
    r.raise_for_status()
    return r.json().get("workItems", [])   # [{id, url}, ...]

# Triage queue: all open Bugs ordered by priority then created date
triage_items = wiql_search(
    "SELECT [System.Id] FROM WorkItems "
    "WHERE [System.WorkItemType] = 'Bug' "
    "AND [System.State] NOT IN ('Resolved', 'Closed') "
    "ORDER BY [Microsoft.VSTS.Common.Priority] ASC, "
    "[System.CreatedDate] DESC"
)

# Duplicate candidate search
dupes = wiql_search(
    f"SELECT [System.Id] FROM WorkItems "
    f"WHERE [System.WorkItemType] = 'Bug' "
    f"AND [System.Title] CONTAINS WORDS 'SSO redirect' "
    f"AND [System.State] <> 'Closed'"
)

def get_work_items(ids, fields=None):
    if not ids:
        return []
    fields_param = ",".join(fields) if fields else (
        "System.Id,System.Title,System.State,"
        "Microsoft.VSTS.Common.Priority,Microsoft.VSTS.Common.Severity"
    )
    ids_param = ",".join(str(i["id"]) for i in ids)
    r = requests.get(
        f"{BASE}/_apis/wit/workitems"
        f"?ids={ids_param}&fields={fields_param}&api-version=7.1",
        headers={**HEADERS, "Content-Type": "application/json"},
    )
    r.raise_for_status()
    return r.json()["value"]

def create_or_comment(title, body_html, priority="2", severity="2 - High"):
    hits = wiql_search(
        f"SELECT [System.Id] FROM WorkItems "
        f"WHERE [System.WorkItemType] = 'Bug' "
        f"AND [System.Title] CONTAINS WORDS '{title[:60]}' "
        f"AND [System.State] <> 'Closed'",
        top=5,
    )
    if hits:
        existing_id = hits[0]["id"]
        add_comment(existing_id, f"Recurred: {body_html[:500]}")
        return existing_id, False   # (id, created)
    item_id, _ = create_bug(title, body_html, priority, severity,
                            tags="ci-failure; auto-filed")
    return item_id, True

def add_comment(work_item_id, text_html):
    r = requests.post(
        f"{BASE}/{PROJECT}/_apis/wit/workItems/{work_item_id}"
        f"/comments?api-version=7.1-preview.3",
        json={"text": text_html},
        headers={**HEADERS, "Content-Type": "application/json"},
    )
    r.raise_for_status()

def link_related(source_id, target_id, comment=""):
    target_url = f"{BASE}/_apis/wit/workItems/{target_id}"
    body = [{
        "op": "add",
        "path": "/relations/-",
        "value": {
            "rel": "System.LinkTypes.Related",
            "url": target_url,
            "attributes": {"comment": comment},
        }
    }]
    r = requests.patch(
        f"{BASE}/{PROJECT}/_apis/wit/workitems/{source_id}?api-version=7.1",
        json=body, headers=HEADERS,
    )
    r.raise_for_status()

def link_pull_request(work_item_id, org_name, project_id, repo_id, pr_id):
    # vstfs artifact URI format for a Git PR:
    # vstfs:///Git/PullRequestId/{projectId}/{repoId}/{prId}
    artifact_url = (
        f"vstfs:///Git/PullRequestId/{project_id}/{repo_id}/{pr_id}"
    )
    body = [{
        "op": "add",
        "path": "/relations/-",
        "value": {
            "rel": "ArtifactLink",
            "url": artifact_url,
            "attributes": {
                "name": "Pull Request",
                "comment": f"Fixing PR !{pr_id}",
            },
        }
    }]
    r = requests.patch(
        f"{BASE}/{PROJECT}/_apis/wit/workitems/{work_item_id}?api-version=7.1",
        json=body, headers=HEADERS,
    )
    r.raise_for_status()

resolved = wiql_search(
    "SELECT [System.Id] FROM WorkItems "
    "WHERE [System.WorkItemType] = 'Bug' "
    "AND [System.State] = 'Resolved' "
    "AND [System.IterationPath] UNDER 'MyProject\\\\Sprint 42'"
)
for item in resolved:
    set_state(item["id"], "Closed")

# Create a Bug
az boards work-item create \
  --type Bug \
  --title "Login fails on SSO redirect" \
  --priority 2 \
  --org "$ADO_ORG" \
  --project "$ADO_PROJECT"

# Update state
az boards work-item update --id 4210 --state Active \
  --org "$ADO_ORG" --project "$ADO_PROJECT"

# Link two work items
az boards work-item relation add --id 4210 \
  --relation-type Related --target-id 4205 \
  --org "$ADO_ORG"

# WIQL query (returns JSON)
az boards query \
  --wiql "SELECT [System.Id],[System.Title] FROM WorkItems \
    WHERE [System.WorkItemType]='Bug' AND [System.State]='New'" \
  --org "$ADO_ORG" --project "$ADO_PROJECT"

permalink = f"{BASE}/{PROJECT}/_workitems/edit/{work_item_id}"

# azure-pipelines.yml (excerpt)
- task: PythonScript@0
  displayName: "File ADO bug on test failure"
  condition: failed()
  inputs:
    scriptSource: filePath
    scriptPath: scripts/file-ado-bug.py
  env:
    ADO_ORG: $(System.CollectionUri)
    ADO_PROJECT: $(System.TeamProject)
    ADO_PAT: $(ADO_PAT_SECRET)
    BUILD_ID: $(Build.BuildId)
    BUILD_URL: $(System.CollectionUri)$(System.TeamProject)/_build/results?buildId=$(Build.BuildId)

export GITHUB_TOKEN="ghp_..."  # personal access token, classic or fine-grained
export GITHUB_REPO="owner/repo"

import requests, os

HEADERS = {
    "Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}",
    "Accept": "application/vnd.github+json",
    "X-GitHub-Api-Version": "2026-03-10",
}
BASE = f"https://api.github.com/repos/{os.environ['GITHUB_REPO']}"

def create_bug(title, body, severity, priority, labels=None):
    payload = {
        "title": title,
        "body": body,
        "labels": (labels or []) + [
            "bug",
            f"severity:{severity}",
            f"priority:{priority}",
        ],
    }
    r = requests.post(f"{BASE}/issues", json=payload, headers=HEADERS)
    r.raise_for_status()
    return r.json()

def close(issue_number, reason="completed"):
    """reason: completed | not_planned | duplicate"""
    r = requests.patch(
        f"{BASE}/issues/{issue_number}",
        json={"state": "closed", "state_reason": reason},
        headers=HEADERS,
    )
    r.raise_for_status()
    result = r.json()
    # verify the destructive transition landed; a stale state means a concurrent edit won
    assert result["state"] == "closed" and result["state_reason"] == reason, result
    return result

def reopen(issue_number):
    r = requests.patch(
        f"{BASE}/issues/{issue_number}",
        json={"state": "open", "state_reason": "reopened"},
        headers=HEADERS,
    )
    r.raise_for_status()
    return r.json()

def search_issues(q):
    r = requests.get(
        "https://api.github.com/search/issues",
        params={"q": f"repo:{os.environ['GITHUB_REPO']} {q}"},
        headers=HEADERS,
    )
    r.raise_for_status()
    return r.json()["items"]

dupes = search_issues(
    f'type:issue is:open label:bug "{title_safe}" in:title,body'
)

def add_comment(issue_number, body):
    r = requests.post(
        f"{BASE}/issues/{issue_number}/comments",
        json={"body": body}, headers=HEADERS)
    r.raise_for_status()
    return r.json()

def create_or_attach(title, body):
    dupes = search_issues(f'is:open label:bug "{title}" in:title')
    if dupes:
        add_comment(dupes[0]["number"], f"Recurred: {body[:500]}")
        return dupes[0]["number"]
    issue = create_bug(title, body, severity="medium", priority="p3")
    return issue["number"]

PROJECTS_MUTATION = """
mutation MoveItem($projectId: ID!, $itemId: ID!, $fieldId: ID!, $optionId: String!) {
  updateProjectV2ItemFieldValue(
    input: { projectId: $projectId, itemId: $itemId,
             fieldId: $fieldId, value: { singleSelectOptionId: $optionId } }
  ) { projectV2Item { id } }
}
"""
# Discovery of projectId, itemId, fieldId, optionId via the matching queries.

# Create
gh issue create \
  --title "Checkout fails for promo X" \
  --body-file failure.md \
  --label bug,severity:high,priority:p2

# Close with reason
gh issue close 1234 --reason completed
gh issue close 1234 --reason "not planned"

# Search
gh issue list --search 'is:open label:bug "checkout fails"'

# .github/workflows/test.yml
- name: Run tests
  id: tests
  run: pytest --junitxml=results.xml
  continue-on-error: true

- name: File issue on test failure
  if: steps.tests.outcome == 'failure'
  env:
    GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
    GITHUB_REPO: ${{ github.repository }}
  run: python scripts/file-github-bug.py results.xml

curl -u "$JIRA_EMAIL:$JIRA_TOKEN" \
     "$JIRA_BASE/rest/api/3/field" \
     | jq '.[] | select(.name=="Severity") | {id, name}'
# {"id": "customfield_10039", "name": "Severity"}

"customfield_10039": {"value": severity},  # "Critical" | "High" | ...

def update_priority(issue_key, priority_name):
    r = requests.put(
        f"{BASE}/rest/api/3/issue/{issue_key}",
        json={"fields": {"priority": {"name": priority_name}}},
        headers=HEADERS,
    )
    r.raise_for_status()

url = f"{BASE}/browse/{issue_key}"

# Personal API key (lin_api_*)
export LINEAR_KEY="lin_api_xxxxxxxxxxxxxxxxxx"

# Or OAuth bearer token
export LINEAR_TOKEN="<oauth-access-token>"

HEADERS_KEY = {
    "Authorization": os.environ["LINEAR_KEY"],   # personal key, no Bearer
    "Content-Type": "application/json",
}
HEADERS_OAUTH = {
    "Authorization": f"Bearer {os.environ['LINEAR_TOKEN']}",
    "Content-Type": "application/json",
}
ENDPOINT = "https://api.linear.app/graphql"

import requests, os

QUERY = """
mutation IssueCreate($input: IssueCreateInput!) {
  issueCreate(input: $input) {
    success
    issue { id identifier url state { name } }
  }
}
"""

def create_bug(team_id, title, description_md, priority, state_id, label_ids=None):
    variables = {
        "input": {
            "teamId": team_id,
            "title": title,
            "description": description_md,  # Markdown supported
            "priority": priority,            # 0=No, 1=Urgent, 2=High, 3=Med, 4=Low
            "stateId": state_id,             # initial state (e.g., "Backlog" or "Todo")
            "labelIds": label_ids or [],
        }
    }
    r = requests.post(ENDPOINT, json={"query": QUERY, "variables": variables},
                      headers=HEADERS_KEY)
    r.raise_for_status()
    data = r.json()
    if data.get("errors"):
        raise RuntimeError(data["errors"])
    return data["data"]["issueCreate"]["issue"]

STATES_QUERY = """
query Workflow($teamId: String!) {
  workflowStates(filter: { team: { id: { eq: $teamId } } }) {
    nodes { id name type }
  }
}
"""

def get_states(team_id):
    r = requests.post(ENDPOINT,
        json={"query": STATES_QUERY, "variables": {"teamId": team_id}},
        headers=HEADERS_KEY)
    r.raise_for_status()
    return r.json()["data"]["workflowStates"]["nodes"]

def transition_to_completed(issue_id, team_id):
    done = next(s for s in get_states(team_id) if s["type"] == "completed")
    return transition(issue_id, done["id"])

UPDATE_QUERY = """
mutation IssueUpdate($id: String!, $input: IssueUpdateInput!) {
  issueUpdate(id: $id, input: $input) {
    success
    issue { id state { name } }
  }
}
"""

def transition(issue_id, new_state_id):
    variables = {"id": issue_id, "input": {"stateId": new_state_id}}
    r = requests.post(ENDPOINT, json={"query": UPDATE_QUERY, "variables": variables},
                      headers=HEADERS_KEY)
    r.raise_for_status()
    data = r.json()
    return data["data"]["issueUpdate"]["success"]

SEARCH_QUERY = """
query SearchIssues($filter: IssueFilter!) {
  issues(filter: $filter, first: 50) {
    nodes { id identifier title state { name } priority }
  }
}
"""

def find_dupes(team_id, title_text):
    r = requests.post(ENDPOINT,
        json={"query": SEARCH_QUERY, "variables": {"filter": {
            "team": {"id": {"eq": team_id}},
            "title": {"contains": title_text},
            "state": {"type": {"neq": "completed"}},
        }}},
        headers=HEADERS_KEY)
    r.raise_for_status()
    return r.json()["data"]["issues"]["nodes"]

def create_or_attach(team_id, title, description):
    dupes = find_dupes(team_id, title)
    if dupes:
        # Comment on the existing issue rather than duplicate
        add_comment(dupes[0]["id"], f"Recurred: {description[:500]}")
        return dupes[0]["identifier"]
    todo_state = next(s for s in get_states(team_id) if s["type"] == "unstarted")
    return create_bug(team_id, title, description, priority=3,
                      state_id=todo_state["id"])["identifier"]

- name: File Linear bug on failure
  if: failure()
  env:
    LINEAR_KEY: ${{ secrets.LINEAR_KEY }}
    LINEAR_TEAM_ID: ${{ vars.LINEAR_TEAM_ID }}
  run: python scripts/file-linear-bug.py results.xml