1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120
|
# pylint: disable=line-too-long,useless-suppression
import functools
import json
from devtools_testutils import AzureRecordedTestCase, EnvironmentVariableLoader, recorded_by_proxy
from azure.core.credentials import AzureKeyCredential
from azure.core.exceptions import HttpResponseError
from azure.ai.language.conversations.authoring import ConversationAuthoringClient
from azure.ai.language.conversations.authoring.models import (
CreateProjectOptions,
ProjectKind,
ProjectSettings,
ExportedProject,
ExportedProjectFormat,
ConversationExportedProjectAsset,
ConversationExportedIntent,
ConversationExportedEntity,
ConversationExportedUtterance,
ExportedUtteranceEntityLabel,
)
ConversationsPreparer = functools.partial(
EnvironmentVariableLoader,
"authoring",
authoring_endpoint="https://Sanitized.cognitiveservices.azure.com/",
authoring_key="fake_key",
)
class TestConversations(AzureRecordedTestCase):
# Start with any helper functions you might need, for example a client creation method:
def create_client(self, endpoint, key):
credential = AzureKeyCredential(key)
client = ConversationAuthoringClient(endpoint, credential)
return client
# ...
class TestConversationsImportCase(TestConversations):
@ConversationsPreparer()
@recorded_by_proxy
def test_import_project(self, authoring_endpoint, authoring_key):
authoring_client = self.create_client(authoring_endpoint, authoring_key)
project_name = "PythonImportProject0820"
project_client = authoring_client.get_project_client(project_name)
# ----- Build assets using objects -----
intents = [
ConversationExportedIntent(category="None"),
ConversationExportedIntent(category="Buy"),
]
entities = [
ConversationExportedEntity(
category="product",
composition_mode="combineComponents",
)
]
u1 = ConversationExportedUtterance(
text="I want to buy a house",
intent="Buy",
language="en-us",
dataset="Train",
entities=[ExportedUtteranceEntityLabel(category="product", offset=16, length=5)],
)
u2 = ConversationExportedUtterance(
text="I want to buy surface pro",
intent="Buy",
language="en-us",
dataset="Train",
entities=[ExportedUtteranceEntityLabel(category="product", offset=14, length=11)],
)
u3 = ConversationExportedUtterance(
text="I want to buy xbox",
intent="Buy",
language="en-us",
dataset="Train",
entities=[ExportedUtteranceEntityLabel(category="product", offset=14, length=4)],
)
assets = ConversationExportedProjectAsset(
intents=intents,
entities=entities,
utterances=[u1, u2, u3],
)
metadata = CreateProjectOptions(
project_kind=ProjectKind.CONVERSATION,
project_name=project_name, # required
language="en-us", # required
settings=ProjectSettings(confidence_threshold=0.0),
multilingual=False,
description="",
)
exported_project = ExportedProject(
project_file_version="2025-05-15-preview",
string_index_type="Utf16CodeUnit",
metadata=metadata,
assets=assets,
)
# ----- Act: begin import (LRO) -----
poller = project_client.project.begin_import(
body=exported_project,
exported_project_format=ExportedProjectFormat.CONVERSATION,
)
try:
poller.result()
except HttpResponseError as e:
print(f"Operation failed: {e.message}")
print(e.error)
raise
print(f"Import completed. done={poller.done()} status={poller.status()}")
|