diff --git a/src/azure-cli/azure/cli/command_modules/find/custom.py b/src/azure-cli/azure/cli/command_modules/find/custom.py index 2912dc70688..d46bcf71aa9 100644 --- a/src/azure-cli/azure/cli/command_modules/find/custom.py +++ b/src/azure-cli/azure/cli/command_modules/find/custom.py @@ -2,35 +2,51 @@ # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License.txt in the project root for license information. # -------------------------------------------------------------------------------------------- -from collections import namedtuple import hashlib -import random import json +import platform import re import sys -import platform + import requests import colorama # pylint: disable=import-error - from azure.cli.core import telemetry as telemetry_core from azure.cli.core import __version__ as core_version from azure.cli.core.style import Style, format_styled_text from packaging.version import parse from knack.log import get_logger + logger = get_logger(__name__) -WAIT_MESSAGE = ['Finding examples and documentation...'] +WAIT_MESSAGE = '\nFinding examples and documentation...' -# Display limits -MAX_DOC_RESULTS = 2 +# Number of entries printed in each section MAX_CODE_RESULTS = 3 +MAX_DOC_RESULTS = 2 + +# Hints appended to the docs query so the semantic search stays Azure CLI specific +DOCS_QUERY_HINTS = ['Azure CLI', 'az command'] -Example = namedtuple("Example", "title snippet") +# Doc summary length and the marker shown when a summary is cut short +MAX_SUMMARY_LENGTH = 150 +MIN_SUMMARY_LENGTH = 40 +CONTINUATION_MARKER = ' ... (see link for the full article)' + +# Filler words that carry no search signal and would otherwise skew filtering +STOP_WORDS = { + 'about', 'all', 'also', 'an', 'and', 'any', 'are', 'as', 'at', 'be', 'been', 'being', 'but', + 'by', 'can', 'could', 'did', 'do', 'does', 'for', 'from', 'had', 'has', 'have', 'here', 'how', + 'if', 'in', 'into', 'is', 'it', 'its', 'may', 'me', 'might', 'must', 'my', 'no', 'not', 'of', + 'on', 'or', 'our', 'over', 'please', 'shall', 'should', 'so', 'some', 'such', 'than', 'that', + 'the', 'their', 'them', 'then', 'there', 'these', 'they', 'this', 'those', 'to', 'up', 'us', + 'via', 'want', 'was', 'we', 'were', 'what', 'when', 'where', 'which', 'while', 'who', 'why', + 'will', 'with', 'would', 'you', 'your', +} class MCPClient: - """Lightweight MCP client for Microsoft Learn MCP Server. + """Lightweight MCP client for the Microsoft Learn MCP server. Implements the minimum JSON-RPC 2.0 over Streamable HTTP protocol needed for single-invocation tool calls (initialize → notify → tools/call). @@ -49,30 +65,21 @@ def __init__(self, client_version=None): self._context = self._build_telemetry_context() def _build_telemetry_context(self): - """Build telemetry context sent alongside MCP requests. + """Build the telemetry context sent alongside MCP requests. - Mirrors the dev branch behavior: the hashed installation id is always - sent as the ``X-UserId`` header (used for DDOS protection and rate - limiting), while the remaining contextual values are only included when - the user has consented to telemetry. + The hashed installation id is always sent as the ``X-UserId`` header + (used for DDOS protection and rate limiting); the remaining contextual + values are only included when the user has consented to telemetry. """ - # Used for DDOS protection and rate limiting user_id = telemetry_core._get_installation_id() # pylint: disable=protected-access - hashed_user_id = hashlib.sha256(user_id.encode('utf-8')).hexdigest() - self._headers["X-UserId"] = hashed_user_id + self._headers["X-UserId"] = hashlib.sha256(user_id.encode('utf-8')).hexdigest() - context = { - "versionNumber": self.client_version, - } + context = {"versionNumber": self.client_version} - # Only pull in the contextual values if we have consent if telemetry_core.is_telemetry_enabled(): - correlation_id = telemetry_core._session.correlation_id # pylint: disable=protected-access - event_id = telemetry_core._session.event_id # pylint: disable=protected-access + context["correlationId"] = telemetry_core._session.correlation_id # pylint: disable=protected-access + context["eventId"] = telemetry_core._session.event_id # pylint: disable=protected-access subscription_id = telemetry_core._get_azure_subscription_id() # pylint: disable=protected-access - - context["correlationId"] = correlation_id - context["eventId"] = event_id if subscription_id is not None: context["subscriptionId"] = subscription_id @@ -87,7 +94,7 @@ def _next_id(self): return self._request_id def initialize(self): - """Send initialize request and store session ID.""" + """Send the initialize request and store the session ID.""" body = { "jsonrpc": "2.0", "id": self._next_id(), @@ -109,20 +116,12 @@ def initialize(self): return self._parse_sse(resp.text) def notify_initialized(self): - """Send initialized notification to confirm client readiness.""" + """Send the initialized notification to confirm client readiness.""" body = {"jsonrpc": "2.0", "method": "notifications/initialized"} requests.post(self.MCP_ENDPOINT, json=body, headers=self._headers, params=self._params, timeout=10) def call_tool(self, tool_name, arguments): - """Call an MCP tool and return parsed results. - - Args: - tool_name: Name of the MCP tool (e.g., 'microsoft_docs_search'). - arguments: Dict of tool arguments. - - Returns: - Parsed JSON result from the tool's text content. - """ + """Call an MCP tool (e.g. 'microsoft_docs_search') and return its parsed result.""" body = { "jsonrpc": "2.0", "id": self._next_id(), @@ -134,110 +133,112 @@ def call_tool(self, tool_name, arguments): } resp = requests.post(self.MCP_ENDPOINT, json=body, headers=self._headers, params=self._params, timeout=30) resp.raise_for_status() - result = self._parse_sse(resp.text) - content_list = result.get("result", {}).get("content", []) - if content_list and content_list[0].get("text"): - return json.loads(content_list[0]["text"]) + content = self._parse_sse(resp.text).get("result", {}).get("content", []) + if content and content[0].get("text"): + return json.loads(content[0]["text"]) return {} @staticmethod def _parse_sse(text): - """Parse single-event SSE response. - - The MCP server returns responses in SSE format with a single 'data:' event. - """ + """Parse a single-event SSE response, which carries one 'data:' line.""" for line in text.split("\n"): if line.startswith("data: "): return json.loads(line[6:]) return {} -def _extract_query_command(query): - """Extract the CLI command group/name from a query string. +def _get_query_keywords(query): + """Extract the meaningful keywords of a query. + + Words of 2+ characters are kept so short but meaningful Azure terms such as + 'vm', 'ad' or 'k8s' survive; the 'az' prefix and filler words (articles, + auxiliaries, interrogatives) are dropped because they carry no signal. + """ + words = re.findall(r'[a-z0-9]{2,}', query.lower()) + return {w for w in words if w != 'az' and w not in STOP_WORDS} + + +def _stem(word): + """Reduce a word to a crude stem so morphological variants match. + + Strips common inflectional suffixes and a trailing 'e' so that 'creating', + 'creates', 'created' and 'create' all collapse to 'creat'. A lightweight, + dependency-free approximation of a real stemmer, good enough for matching + query terms against documentation and code samples. + """ + word = word.lower() + for suffix in ('ings', 'ing', 'ies', 'ied', 'es', 'ed', 's'): + if word.endswith(suffix) and len(word) - len(suffix) >= 3: + word = word[:-len(suffix)] + break + if len(word) > 3 and word.endswith('e'): + word = word[:-1] + return word - Examples: - 'az vm create' → 'az vm create' - 'az vm' → 'az vm' - 'vm create' → 'az vm create' - 'deploy arm template' → None (not a CLI command pattern) - Returns: - Normalized command string or None if not a CLI command pattern. +def _matches_keywords(text, query_words): + """Check whether any query keyword appears in the text, comparing stems.""" + text_stems = {_stem(w) for w in re.findall(r'[a-z0-9]+', text.lower())} + return any(_stem(word) in text_stems for word in query_words) + + +def _has_keyword_overlap(result, query_words): + """Check whether a doc result's title or content shares a keyword with the query.""" + combined = result.get("title", "") + " " + result.get("content", "")[:500] + return _matches_keywords(combined, query_words) + + +def _extract_query_command(query): + """Normalize a query into an `az` command, or None if it isn't one. + + 'az vm create' → 'az vm create'; 'vm create' → 'az vm create'; + 'deploy/arm template' → None (doesn't look like a command). """ query = query.strip().lower() if query.startswith('az '): return query - # If the query looks like a CLI subcommand (single words that could be a group) + parts = query.split() - if parts and not any(c in parts[0] for c in ' ./-'): + if parts and not any(c in parts[0] for c in './-'): return 'az ' + query return None def _is_cli_command_relevant(title, query): - """Check if a CLI command result title is relevant to the query. - - For CLI command results (titles starting with 'az '), checks that - the result command shares the same command group as the query. + """Check whether a CLI command title belongs to the same command group as the query. - Examples (query='az vm create'): - 'az vm create' → True (exact match) - 'az vm run-command create' → True (same group) - 'az lab vm create' → False (different group: 'lab' vs 'vm') - 'az connectedvmware vm create' → False (different group) + Only titles that look like a command ('az ...') are judged; anything else + (tutorials, concept articles) is always considered relevant. - Args: - title: The result title string. - query: The user's query string. - - Returns: - True if the result is relevant, False otherwise. + With query='az vm create': 'az vm create' and 'az vm run-command create' + are relevant, while 'az lab vm create' and 'az connectedvmware vm create' + belong to other groups. """ title_lower = title.strip().lower() - - # Only filter CLI command titles (starting with 'az ') if not title_lower.startswith('az '): return True - cmd = _extract_query_command(query) - if not cmd: + command = _extract_query_command(query) + if not command: return True - # Extract the command group (first word after 'az') - cmd_parts = cmd.split() - if len(cmd_parts) < 2: - return True - - query_group = cmd_parts[1] # e.g., 'vm' from 'az vm create' - + command_parts = command.split() title_parts = title_lower.split() - if len(title_parts) < 2: + if len(command_parts) < 2 or len(title_parts) < 2: return True - title_group = title_parts[1] # e.g., 'lab' from 'az lab vm create' - - # The result's first command group must match the query's command group - return title_group == query_group + # e.g. 'vm' from 'az vm create' vs 'lab' from 'az lab vm create' + return title_parts[1] == command_parts[1] def _filter_results(results, query): - """Filter MCP results for relevance to the query. + """Drop doc results that belong to another command group or share no keyword. - Removes CLI command results that belong to different command groups, - and filters out results whose titles have no meaningful word overlap - with the query (to discard noise from semantic search on gibberish queries). - - Args: - results: List of MCP doc result dicts. - query: The user's query string. - - Returns: - Filtered list of result dicts. + The keyword check discards the noise a semantic search returns for + gibberish queries. """ filtered = [r for r in results if _is_cli_command_relevant(r.get("title", ""), query)] - # For non-empty queries, check that at least some results are genuinely relevant - # by verifying word overlap between the query and result titles/content query_words = _get_query_keywords(query) if query_words: filtered = [r for r in filtered if _has_keyword_overlap(r, query_words)] @@ -245,127 +246,148 @@ def _filter_results(results, query): return filtered -def _get_query_keywords(query): - """Extract meaningful keywords from the query (words with 3+ chars, excluding 'az'). +def _build_docs_query(query): + """Append Azure CLI hints to a query so the docs search stays CLI specific. - Args: - query: The user's query string. + Without them the semantic search happily returns portal, PowerShell or SDK + articles. Hints already present in the query are not repeated. + """ + query = (query or "").strip() + lowered = query.lower() + hints = [hint for hint in DOCS_QUERY_HINTS if hint.lower() not in lowered] + return " ".join([query] + hints) if hints else query - Returns: - Set of lowercase keyword strings. + +def _cli_doc_score(result): + """Score how Azure CLI specific a doc result is. + + 3 for the `az` command reference, 2 for content showing `az` invocations, + 1 for content merely mentioning the Azure CLI, 0 for everything else. """ - words = re.findall(r'[a-zA-Z]{3,}', query.lower()) - stop_words = {'the', 'and', 'for', 'with', 'from', 'that', 'this', 'are', 'was', 'has', 'have'} - return {w for w in words if w != 'az' and w not in stop_words} + if "/cli/azure" in (result.get("contentUrl", "") or "").lower(): + return 3 + content = (result.get("content", "") or "") + " " + (result.get("title", "") or "") + lowered = content.lower() + if "azurecli" in lowered or re.search(r'(?m)^\s*az\s+[a-z][\w-]*', content): + return 2 -def _has_keyword_overlap(result, query_words): - """Check if a result has meaningful keyword overlap with the query. + return 1 if "azure cli" in lowered else 0 - Args: - result: A doc result dict with 'title' and optionally 'content'. - query_words: Set of query keywords to match against. - Returns: - True if at least one query keyword appears in the result's title or content. +def _prefer_cli_docs(results): + """Keep the Azure CLI related doc results, most CLI specific first. + + The server's relative ranking is preserved within the same score. If no + result looks CLI specific at all, the original list is returned so the user + still gets something back. """ - title = result.get("title", "").lower() - content = result.get("content", "").lower()[:500] # Only check first 500 chars of content - combined = title + " " + content + scored = [(score, i, result) for i, result in enumerate(results) + if (score := _cli_doc_score(result)) > 0] + if not scored: + return results - return any(word in combined for word in query_words) + scored.sort(key=lambda item: (-item[0], item[1])) + return [result for _, _, result in scored] def search_mslearn(query): - """Search Microsoft Learn via MCP for docs and code samples. - - Calls two MCP tools: - 1. microsoft_docs_search - for command reference and documentation - 2. microsoft_code_sample_search - for runnable CLI examples + """Search Microsoft Learn for docs and code samples matching the query. - Args: - query: Search query string (e.g., 'az vm delete'). - - Returns: - Tuple of (docs_results, code_results) where each is a list of dicts. + Returns a (docs_results, code_results) tuple, each a list of result dicts + already filtered down to what is relevant to the Azure CLI. """ client = MCPClient() - client.initialize() client.notify_initialized() docs_response = client.call_tool( "microsoft_docs_search", - {"query": query} + {"query": _build_docs_query(query)} ) - code_response = client.call_tool( "microsoft_code_sample_search", {"query": query, "language": "azurecli"} ) - docs_results = docs_response.get("results", []) - code_results = code_response.get("results", []) - - # Filter out irrelevant CLI commands (e.g., 'az lab vm' when searching 'az vm') - docs_results = _filter_results(docs_results, query) + docs_results = _prefer_cli_docs(_filter_results(docs_response.get("results", []), query)) - # Filter code results by keyword overlap too + code_results = code_response.get("results", []) query_words = _get_query_keywords(query) if query_words: code_results = [r for r in code_results - if any(word in (r.get("codeSnippet", "") + " " + - r.get("description", "")).lower() - for word in query_words)] + if _matches_keywords(r.get("codeSnippet", "") + " " + r.get("description", ""), + query_words)] return docs_results, code_results -def _extract_summary(content): - """Extract a clean summary from MCP doc content. +def _clean_markdown(text): + """Strip markdown noise (images, links, emphasis) so text reads as plain prose.""" + text = re.sub(r'!\[[^\]]*\]\([^)]*\)', '', text) + text = re.sub(r'\[([^\]]+)\]\([^)]*\)', r'\1', text) + text = re.sub(r'[*_`]+', '', text) + return re.sub(r'\s+', ' ', text).strip() - MCP returns markdown content with headers. Extract the Summary section - or first meaningful paragraph. - Args: - content: Raw markdown content string. +def _shorten(text): + """Trim text to MAX_SUMMARY_LENGTH, never cutting mid-word. - Returns: - Clean summary string, max 150 chars. + A sentence boundary is preferred as the cut point. Whenever text is + dropped, CONTINUATION_MARKER is appended so it's clear the rest of the + article lives behind the link. """ + text = text.strip() + if len(text) <= MAX_SUMMARY_LENGTH: + return text + + window = text[:MAX_SUMMARY_LENGTH + 1] + + # Only cut at a sentence boundary if that still keeps most of the window. + sentence_end = max(window.rfind('. '), window.rfind('! '), window.rfind('? ')) + if sentence_end >= MAX_SUMMARY_LENGTH // 2: + return window[:sentence_end + 1] + CONTINUATION_MARKER + + cut = window.rfind(' ') + if cut <= 0: + cut = MAX_SUMMARY_LENGTH + return window[:cut].rstrip(' ,;:-') + CONTINUATION_MARKER + + +def _extract_summary(content): + """Build a short, readable summary from a doc result's markdown content.""" if not content: return "" - # Try to find a "### Summary" section summary_match = re.search(r'###\s*Summary\s*\n(.+?)(?:\n###|\Z)', content, re.DOTALL) if summary_match: - summary = summary_match.group(1).strip() - # Take first sentence/line - first_line = summary.split('\n')[0].strip() - if first_line: - return first_line[:150] + summary = _clean_markdown(summary_match.group(1).split('\n\n')[0]) + if summary: + return _shorten(summary) - # Try first non-header, non-empty line + candidates = [] for line in content.split('\n'): line = line.strip() - if line and not line.startswith('#') and not line.startswith('---'): - return line[:150] - - return content[:150] - + if not line or line.startswith(('#', '---', '|', '```', '>')): + continue + cleaned = _clean_markdown(line) + if cleaned: + candidates.append(cleaned) + if len(candidates) >= 10: + break -def _clean_title(title): - """Normalize a title into a real sentence. + if candidates: + # Prefer a line that reads as a complete thought over a lead-in fragment. + best = next((c for c in candidates + if len(c) >= MIN_SUMMARY_LENGTH and not c.endswith((':', ';', ','))), + candidates[0]) + return _shorten(best) - Strips leading markdown header markers ('#') and ensures the title - ends with a sentence mark (period, question mark, or exclamation mark). + return _shorten(_clean_markdown(content)) - Args: - title: Raw title string. - Returns: - Cleaned title string, or empty string if no title. - """ +def _clean_title(title): + """Normalize a title into a sentence: no leading '#', always end-punctuated.""" if not title: return "" @@ -378,16 +400,8 @@ def _clean_title(title): def _to_imperative(text): """Convert a leading third-person-singular verb to imperative mood. - Examples: - 'Deploys the template.' -> 'Deploy the template.' - 'Creates a virtual machine.' -> 'Create a virtual machine.' - 'Specifies the name.' -> 'Specify the name.' - - Args: - text: A description sentence. - - Returns: - The sentence with its first word converted to imperative mood. + 'Deploys the template.' → 'Deploy the template.' + 'Specifies the name.' → 'Specify the name.' """ if not text: return text @@ -403,17 +417,10 @@ def _to_imperative(text): def _extract_description(description): - """Extract the human-readable description from a code sample's metadata. - - The MCP code sample 'description' field is a metadata blob such as: - 'description: Deploys the ARM template ...\\nlanguage: azurecli\\n' - This extracts just the description text as a clean sentence. + """Pull the human-readable sentence out of a code sample's metadata blob. - Args: - description: Raw description metadata string. - - Returns: - Cleaned description sentence, or 'Example.' as a fallback. + The raw field looks like 'description: Deploys the ARM template\\nlanguage: + azurecli\\n'. """ if description: for line in description.split('\n'): @@ -421,7 +428,6 @@ def _extract_description(description): if line.lower().startswith('description:'): return _clean_title(_to_imperative(line[len('description:'):].strip())) - # Fallback: first non-empty, non-metadata line for line in description.split('\n'): line = line.strip() if line and not line.lower().startswith(('language:', 'package:')): @@ -431,17 +437,10 @@ def _extract_description(description): def _extract_command(snippet): - """Extract the `az` command block from a code snippet. - - Returns the snippet lines starting from the first line that begins with - 'az', preserving the server's original formatting. Stops at a blank line - that isn't a shell line-continuation, so a single example is returned. + """Extract the `az` commands from a code snippet, one per line. - Args: - snippet: Raw code snippet string. - - Returns: - List of command lines (server formatting preserved), or empty list. + Shell line-continuations are collapsed so a command that the docs wrapped + over many lines is printed as a single copy-pasteable line. """ if not snippet: return [] @@ -451,108 +450,134 @@ def _extract_command(snippet): if start is None: return [] - # If the command block is indented, strip the leading indent of the first - # `az` line from every line, preserving relative indentation. - indent = len(lines[start]) - len(lines[start].lstrip()) - result = [] + commands = [] + pending = None for line in lines[start:]: - if not line.strip(): + line = line.strip() + if not line: continue - # Remove up to `indent` leading whitespace chars, keeping deeper indents. - stripped = line - for _ in range(indent): - if stripped[:1] in (' ', '\t'): - stripped = stripped[1:] - else: - break - result.append(stripped.rstrip()) - return result + # '/' is a typo for '\' seen in some docs; require preceding whitespace + # so that trailing slashes in URLs and paths are left alone. + continued = line.endswith(('\\', '^', '`')) or bool(re.search(r'\s/$', line)) + if continued: + line = line[:-1].rstrip() -def format_results(query, docs_results, code_results): - """Format and print search results to stdout. + pending = line if pending is None else (pending + ' ' + line).strip() + if not continued: + commands.append(pending) + pending = None + + if pending: + commands.append(pending) - Displays results in two sections: - 1. Examples - from microsoft_code_sample_search (shown first) - 2. Documentation - from microsoft_docs_search + return commands - Args: - query: Original search query (for display). - docs_results: List of doc result dicts from MCP. - code_results: List of code sample dicts from MCP. + +def _dedupe_key(text): + """Normalize text into a comparison key, ignoring case and punctuation.""" + return re.sub(r'[^a-z0-9]+', ' ', (text or "").lower()).strip() + + +def _build_example_entry(result): + """Turn a code sample result into a (title, command lines, url) entry. + + Returns None when the sample contains no `az` command. """ - if not docs_results and not code_results: - print("\nSorry I am not able to help with [" + query + "]." - "\nTry typing the beginning of a command e.g., " + style_message('az vm') + ".", file=sys.stderr) + command_lines = _extract_command(result.get("codeSnippet", "")) + if not command_lines: + return None + + return (_extract_description(result.get("description", "")), + command_lines, + result.get("link", "")) + + +def _build_doc_entry(result): + """Turn a doc result into a (title, summary, url) entry.""" + return (_clean_title(result.get("title", "")), + _extract_summary(result.get("content", "")), + result.get("contentUrl", "")) + + +def _collect_unique(results, limit, build_entry): + """Build up to `limit` entries, skipping empty ones and duplicates. + + A result is skipped when its URL or its normalized title was already used, + so the same article never shows up twice under different URL fragments. + """ + entries = [] + seen_urls = set() + seen_titles = set() + + for result in results: + entry = build_entry(result) + if not entry: + continue + + title, _, url = entry + title_key = _dedupe_key(title) + if (url and url in seen_urls) or (title_key and title_key in seen_titles): + continue + + seen_urls.add(url) + seen_titles.add(title_key) + entries.append(entry) + if len(entries) >= limit: + break + + return entries + + +def _print_entry(title, body_lines, url): + """Print one result: a highlighted title, its body, then the source link.""" + print(format_styled_text((Style.HIGHLIGHT, " - " + title))) + for line in body_lines: + print(" " + line) + if url: + print(" " + format_styled_text((Style.SECONDARY, url))) + print() + + +def format_results(query, docs_results, code_results): + """Print the runnable examples first, then the documentation links.""" + # Collect first: results can still end up empty here, for instance when no + # code sample contains an actual `az` command. + examples = _collect_unique(code_results, MAX_CODE_RESULTS, _build_example_entry) + docs = _collect_unique(docs_results, MAX_DOC_RESULTS, _build_doc_entry) + + if not examples and not docs: + print('\nSorry I am not able to help with [' + query + '].' + '\nTry typing the beginning of a command, e.g., "az vm create".\n', file=sys.stderr) return print("\nHere is what I found for [" + query + "]: \n", file=sys.stderr) - if code_results: - examples = [] - seen_urls = set() - for result in code_results: - command_lines = _extract_command(result.get("codeSnippet", "")) - if not command_lines: - continue - url = result.get("link", "") - # Skip duplicate URLs, keeping only the first occurrence. - if url and url in seen_urls: - continue - if url: - seen_urls.add(url) - examples.append(( - _extract_description(result.get("description", "")), - command_lines, - url - )) - if len(examples) >= MAX_CODE_RESULTS: - break - - if examples: - print("Examples") - for title, command_lines, url in examples: - print(format_styled_text((Style.ACTION, " - " + title))) - for line in command_lines: - print(" " + line) - if url: - print(" " + format_styled_text((Style.SECONDARY, url))) - print() - - if docs_results: - docs = [] - seen_urls = set() - for result in docs_results: - url = result.get("contentUrl", "") - # Skip duplicate URLs, keeping only the first occurrence. - if url and url in seen_urls: - continue - if url: - seen_urls.add(url) - docs.append(( - _clean_title(result.get("title", "")), - _extract_summary(result.get("content", "")), - url - )) - if len(docs) >= MAX_DOC_RESULTS: - break - - if docs: - print("Documentation") - for title, summary, url in docs: - print(format_styled_text((Style.ACTION, " - " + title))) - if summary: - print(" " + summary) - if url: - print(" " + format_styled_text((Style.SECONDARY, url))) - print() + if examples: + print("Examples") + for title, command_lines, url in examples: + _print_entry(title, command_lines, url) + + if docs: + print("Documentation") + for title, summary, url in docs: + _print_entry(title, [summary] if summary else [], url) + + +def should_enable_styling(): + """Check whether output is going to a terminal that can render styling.""" + try: + return bool(sys.stdout.isatty()) + except AttributeError: + return False def process_query(cli_term): + """Entry point for `az find`.""" if not cli_term: - logger.error('Please provide a search term e.g. az find "vm"') + logger.error('Please provide a search term, e.g., az find "az vm create".') else: - print(random.choice(WAIT_MESSAGE), file=sys.stderr) + print(WAIT_MESSAGE, file=sys.stderr) try: docs_results, code_results = search_mslearn(cli_term) @@ -572,47 +597,3 @@ def process_query(cli_term): from azure.cli.core.util import show_updates_available show_updates_available() - - -def get_generated_examples(cli_term): - """Get generated examples for a CLI term. - - Returns list of Example namedtuples for backward compatibility. - """ - examples = [] - try: - docs_results, code_results = search_mslearn(cli_term) - - for result in docs_results: - title = result.get("title", "") - summary = _extract_summary(result.get("content", "")) - examples.append(Example(title, summary)) - - for result in code_results: - snippet = result.get("codeSnippet", "") - desc = result.get("description", "") - examples.append(Example(desc[:100] if desc else "Example", snippet)) - - except requests.exceptions.RequestException: - pass - - return examples - - -def style_message(msg): - if should_enable_styling(): - try: - msg = colorama.Style.BRIGHT + msg + colorama.Style.RESET_ALL - except KeyError: - pass - return msg - - -def should_enable_styling(): - try: - # Style if tty stream is available - if sys.stdout.isatty(): - return True - except AttributeError: - pass - return False diff --git a/src/azure-cli/azure/cli/command_modules/find/tests/latest/test_find.py b/src/azure-cli/azure/cli/command_modules/find/tests/latest/test_find.py index 57282b671ae..36f5a3195d7 100644 --- a/src/azure-cli/azure/cli/command_modules/find/tests/latest/test_find.py +++ b/src/azure-cli/azure/cli/command_modules/find/tests/latest/test_find.py @@ -9,11 +9,11 @@ from io import StringIO from azure.cli.command_modules.find.custom import ( - Example, MCPClient, search_mslearn, format_results, - get_generated_examples, process_query, _extract_summary, + MCPClient, search_mslearn, format_results, process_query, _extract_summary, _is_cli_command_relevant, _extract_query_command, _filter_results, - _get_query_keywords, _has_keyword_overlap, - MAX_DOC_RESULTS, MAX_CODE_RESULTS + _get_query_keywords, _has_keyword_overlap, _stem, _matches_keywords, + _extract_command, _build_docs_query, _prefer_cli_docs, _dedupe_key, + MAX_DOC_RESULTS, MAX_CODE_RESULTS, MAX_SUMMARY_LENGTH, CONTINUATION_MARKER ) @@ -69,7 +69,7 @@ def _make_tool_response(results, request_id=2): }, { "title": "Delete a VM and attached resources", - "content": "# Delete a VM and attached resources\nYou can change the behavior when you delete a VM.", + "content": "# Delete a VM and attached resources\nYou can change the behavior when you delete a VM.\n\naz vm delete --resource-group myResourceGroup --name myVM", "contentUrl": "https://learn.microsoft.com/azure/virtual-machines/delete" } ] @@ -210,7 +210,27 @@ def test_extract_summary_empty(self): def test_extract_summary_truncates(self): long_content = "A" * 200 summary = _extract_summary(long_content) - self.assertEqual(len(summary), 150) + self.assertTrue(summary.endswith(CONTINUATION_MARKER)) + self.assertEqual(len(summary), MAX_SUMMARY_LENGTH + len(CONTINUATION_MARKER)) + + def test_extract_summary_marks_truncated_sentence(self): + content = "Create a VM with the Azure CLI. " + ("word " * 60) + summary = _extract_summary(content) + self.assertTrue(summary.startswith("Create a VM with the Azure CLI.")) + self.assertTrue(summary.endswith(CONTINUATION_MARKER)) + + def test_extract_summary_not_truncated_has_no_marker(self): + self.assertNotIn(CONTINUATION_MARKER, _extract_summary("A short summary.")) + + def test_extract_summary_strips_markdown_links(self): + content = "# Title\n[New-AZVM](https://learn.microsoft.com/powershell) creates the resources you need." + self.assertEqual(_extract_summary(content), + "New-AZVM creates the resources you need.") + + def test_extract_summary_skips_fragment_lines(self): + content = "# Title\nIf you plan to use Cloud Shell:\nCloud Shell is an interactive shell that you run in your browser." + self.assertEqual(_extract_summary(content), + "Cloud Shell is an interactive shell that you run in your browser.") class TestRelevanceFiltering(unittest.TestCase): @@ -275,10 +295,16 @@ def test_filter_results_gibberish_query(self): self.assertEqual(len(filtered), 0) def test_get_query_keywords(self): - self.assertEqual(_get_query_keywords("az vm create"), {"create"}) # 'az' excluded, 'vm' < 3 chars + # 'az' is dropped, short-but-meaningful terms like 'vm' are kept + self.assertEqual(_get_query_keywords("az vm create"), {"vm", "create"}) self.assertEqual(_get_query_keywords("az storage blob list"), {"storage", "blob", "list"}) self.assertEqual(_get_query_keywords("deploy arm template"), {"deploy", "arm", "template"}) + def test_get_query_keywords_drops_filler_words(self): + # Regression: natural-language queries must keep their subject + self.assertEqual(_get_query_keywords("what is vm"), {"vm"}) + self.assertEqual(_get_query_keywords("how do i create a vm"), {"vm", "create"}) + def test_has_keyword_overlap_match(self): result = {"title": "az vm create", "content": "Create a virtual machine"} self.assertTrue(_has_keyword_overlap(result, {"create"})) @@ -287,9 +313,96 @@ def test_has_keyword_overlap_no_match(self): result = {"title": "Albanian Keyboard", "content": "KLID code"} self.assertFalse(_has_keyword_overlap(result, {"alskdn1k2lenasd"})) + def test_stem_verb_variants_collapse(self): + # Inflected verb forms should collapse to the same stem + self.assertEqual(_stem("creating"), _stem("create")) + self.assertEqual(_stem("creates"), _stem("create")) + self.assertEqual(_stem("created"), _stem("create")) + self.assertEqual(_stem("deleting"), _stem("delete")) -class TestSearchMslearn(unittest.TestCase): + def test_matches_keywords_morphological_variant(self): + # 'creating' in the query should match 'create' in the sample text + self.assertTrue(_matches_keywords("az vm create --name myvm", {"creating"})) + self.assertTrue(_matches_keywords("az vm delete --name myvm", {"deleting"})) + + def test_matches_keywords_no_match(self): + self.assertFalse(_matches_keywords("az vm create", {"storage"})) + + def test_has_keyword_overlap_morphological_variant(self): + # Regression: 'creating a vm' should still surface 'az vm create' results + result = {"title": "az vm create", "content": "Create a virtual machine"} + self.assertTrue(_has_keyword_overlap(result, _get_query_keywords("creating a vm"))) + + +class TestCliScoping(unittest.TestCase): + + def test_build_docs_query_appends_hints(self): + query = _build_docs_query("create a vm") + self.assertTrue(query.startswith("create a vm")) + self.assertIn("Azure CLI", query) + + def test_build_docs_query_skips_existing_hint(self): + self.assertEqual(_build_docs_query("azure cli vm create").count("Azure CLI"), 0) + + def test_prefer_cli_docs_drops_non_cli(self): + results = [ + {"title": "Create a VM in the portal", "content": "Select Create.", "contentUrl": "https://x/portal"}, + {"title": "az vm create", "content": "Create a VM.", "contentUrl": "https://learn.microsoft.com/cli/azure/vm"}, + ] + kept = _prefer_cli_docs(results) + self.assertEqual([r["title"] for r in kept], ["az vm create"]) + + def test_prefer_cli_docs_ranks_reference_first(self): + results = [ + {"title": "Tutorial", "content": "Run:\naz vm create --name myVM", "contentUrl": "https://x/tutorial"}, + {"title": "az vm create", "content": "Create a VM.", "contentUrl": "https://learn.microsoft.com/cli/azure/vm"}, + ] + self.assertEqual([r["title"] for r in _prefer_cli_docs(results)], ["az vm create", "Tutorial"]) + + def test_prefer_cli_docs_falls_back_when_none_match(self): + results = [{"title": "Portal", "content": "Select Create.", "contentUrl": "https://x/portal"}] + self.assertEqual(_prefer_cli_docs(results), results) + + +class TestExtractCommand(unittest.TestCase): + + def test_collapses_backslash_continuations(self): + snippet = "az vm create \\\n -n myVM \\\n --image myImage" + self.assertEqual(_extract_command(snippet), ["az vm create -n myVM --image myImage"]) + + def test_collapses_caret_and_backtick_continuations(self): + self.assertEqual(_extract_command("az vm create ^\n -n myVM"), ["az vm create -n myVM"]) + self.assertEqual(_extract_command("az vm create `\n -n myVM"), ["az vm create -n myVM"]) + + def test_collapses_mistyped_slash_continuation(self): + snippet = "az storage account create -n acc /\n -g rg --sku Standard_GRS" + self.assertEqual(_extract_command(snippet), ["az storage account create -n acc -g rg --sku Standard_GRS"]) + + def test_keeps_trailing_slash_in_values(self): + snippet = "az storage blob upload --url https://x.blob.core.windows.net/c/" + self.assertEqual(_extract_command(snippet), [snippet]) + def test_keeps_separate_commands_on_separate_lines(self): + self.assertEqual(_extract_command("az vm create -n x\n\naz vm show -n x"), + ["az vm create -n x", "az vm show -n x"]) + + def test_no_command_returns_empty(self): + self.assertEqual(_extract_command("Write-Host hello"), []) + self.assertEqual(_extract_command(""), []) + + +class TestDedupeKey(unittest.TestCase): + + def test_ignores_case_and_punctuation(self): + self.assertEqual(_dedupe_key("Create an Azure storage account"), + _dedupe_key("Create an Azure Storage Account.")) + + def test_empty_input(self): + self.assertEqual(_dedupe_key(""), "") + self.assertEqual(_dedupe_key(None), "") + + +class TestSearchMslearn(unittest.TestCase): @mock.patch('azure.cli.command_modules.find.custom.telemetry_core') @mock.patch('requests.post') def test_search_returns_docs_and_code(self, mock_post, mock_telemetry): @@ -351,6 +464,18 @@ def test_format_empty_results(self, mock_stderr, _): output = mock_stderr.getvalue() self.assertIn("Sorry I am not able to help with", output) + @mock.patch('azure.cli.command_modules.find.custom.should_enable_styling', return_value=False) + @mock.patch('sys.stdout', new_callable=StringIO) + @mock.patch('sys.stderr', new_callable=StringIO) + def test_format_unusable_results(self, mock_stderr, mock_stdout, _): + # Regression: code samples without an `az` command produce no entries, + # so the apology must be shown instead of an empty result header. + code = [dict(SAMPLE_CODE_RESULTS[0], codeSnippet="Remove-AzVM -Name myVM")] + format_results("what is rm", [], code) + self.assertIn("Sorry I am not able to help with", mock_stderr.getvalue()) + self.assertNotIn("Here is what I found", mock_stderr.getvalue()) + self.assertEqual(mock_stdout.getvalue(), "") + @mock.patch('azure.cli.command_modules.find.custom.should_enable_styling', return_value=False) @mock.patch('sys.stdout', new_callable=StringIO) def test_format_docs_only(self, mock_stdout, _): @@ -370,15 +495,17 @@ def test_format_code_only(self, mock_stdout, _): @mock.patch('azure.cli.command_modules.find.custom.should_enable_styling', return_value=False) @mock.patch('sys.stdout', new_callable=StringIO) def test_format_caps_results(self, mock_stdout, _): - # Create more (distinct-URL) results than the max. + # Create more (distinct URL and title) results than the max. many_docs = [] for i in range(MAX_DOC_RESULTS + 5): doc = dict(SAMPLE_DOC_RESULTS[0]) + doc["title"] = "az vm delete %d" % i doc["contentUrl"] = "https://learn.microsoft.com/doc/%d" % i many_docs.append(doc) many_code = [] for i in range(MAX_CODE_RESULTS + 5): code = dict(SAMPLE_CODE_RESULTS[0]) + code["description"] = "Deletes an Azure virtual machine, variant %d." % i code["link"] = "https://learn.microsoft.com/code/%d" % i many_code.append(code) format_results("az vm", many_docs, many_code) @@ -403,6 +530,7 @@ def test_format_dedupes_by_url(self, mock_stdout, _): dup_code = dict(SAMPLE_CODE_RESULTS[0]) extra_code = dict(SAMPLE_CODE_RESULTS[1]) + extra_code["description"] = "Deletes a virtual machine and its disks in Azure." extra_code["link"] = "https://learn.microsoft.com/code/extra" code = [SAMPLE_CODE_RESULTS[0], dup_code, SAMPLE_CODE_RESULTS[1], extra_code] @@ -416,6 +544,29 @@ def test_format_dedupes_by_url(self, mock_stdout, _): self.assertIn("https://learn.microsoft.com/doc/extra", output) self.assertIn("https://learn.microsoft.com/code/extra", output) + @mock.patch('azure.cli.command_modules.find.custom.should_enable_styling', return_value=False) + @mock.patch('sys.stdout', new_callable=StringIO) + def test_format_dedupes_by_title(self, mock_stdout, _): + # Same article under different URL fragments must appear only once. + dup_doc = dict(SAMPLE_DOC_RESULTS[0]) + dup_doc["title"] = "AZ VM Delete." # same title, different casing/punctuation + dup_doc["contentUrl"] = "https://learn.microsoft.com/cli/azure/vm#delete" + other_doc = dict(SAMPLE_DOC_RESULTS[1]) + + dup_code = dict(SAMPLE_CODE_RESULTS[0]) + dup_code["link"] = "https://learn.microsoft.com/code/other-fragment" + + format_results("az vm", [SAMPLE_DOC_RESULTS[0], dup_doc, other_doc], + [SAMPLE_CODE_RESULTS[0], dup_code, SAMPLE_CODE_RESULTS[1]]) + output = mock_stdout.getvalue() + + self.assertEqual(output.count("az vm delete."), 1) + self.assertNotIn("https://learn.microsoft.com/cli/azure/vm#delete", output) + self.assertNotIn("https://learn.microsoft.com/code/other-fragment", output) + # The next unique results still fill the freed slots. + self.assertIn(SAMPLE_DOC_RESULTS[1]["contentUrl"], output) + self.assertIn(SAMPLE_CODE_RESULTS[1]["link"], output) + class TestProcessQuery(unittest.TestCase): @@ -460,40 +611,5 @@ def test_process_query_empty_term(self, _): process_query(None) -class TestGetGeneratedExamples(unittest.TestCase): - - @mock.patch('azure.cli.command_modules.find.custom.telemetry_core') - @mock.patch('requests.post') - def test_get_generated_examples(self, mock_post, mock_telemetry): - mock_telemetry._get_installation_id.return_value = "test-install-id" - mock_telemetry.is_telemetry_enabled.return_value = False - - mock_post.side_effect = [ - _make_init_response(), - _make_notify_response(), - _make_tool_response(SAMPLE_DOC_RESULTS), - _make_tool_response(SAMPLE_CODE_RESULTS), - ] - - examples = get_generated_examples("az vm delete") - - self.assertGreater(len(examples), 0) - # Should return Example namedtuples - self.assertIsInstance(examples[0], Example) - self.assertEqual(examples[0].title, "az vm delete") - - @mock.patch('azure.cli.command_modules.find.custom.telemetry_core') - @mock.patch('requests.post') - def test_get_generated_examples_network_error(self, mock_post, mock_telemetry): - mock_telemetry._get_installation_id.return_value = "test-install-id" - mock_telemetry.is_telemetry_enabled.return_value = False - - import requests as req - mock_post.side_effect = req.exceptions.ConnectionError("fail") - - examples = get_generated_examples("az vm delete") - self.assertEqual(len(examples), 0) - - if __name__ == '__main__': unittest.main()