1+ import os .path
2+ from pprint import pprint
3+ import time
4+
5+ from azure .cognitiveservices .vision .contentmoderator import ContentModeratorClient
6+ from azure .cognitiveservices .vision .contentmoderator .models import (
7+ APIErrorException ,
8+ TermList ,
9+ Terms ,
10+ TermsData ,
11+ RefreshIndex ,
12+ Screen
13+ )
14+ from msrest .authentication import CognitiveServicesCredentials
15+
16+ SUBSCRIPTION_KEY_ENV_NAME = "CONTENTMODERATOR_SUBSCRIPTION_KEY"
17+ CONTENTMODERATOR_LOCATION = os .environ .get ("CONTENTMODERATOR_LOCATION" , "westcentralus" )
18+
19+ # The number of minutes to delay after updating the search index before
20+ # performing image match operations against the list.
21+ LATENCY_DELAY = 0.5
22+
23+
24+ def terms_lists (subscription_key ):
25+ """TermsList.
26+
27+ This will screen text using a term list.
28+ """
29+
30+ client = ContentModeratorClient (
31+ CONTENTMODERATOR_LOCATION + '.api.cognitive.microsoft.com' ,
32+ CognitiveServicesCredentials (subscription_key )
33+ )
34+
35+ #
36+ # Create list
37+ #
38+
39+ print ("\n Creating list" )
40+ custom_list = client .list_management_term_lists .create (
41+ "application/json" ,
42+ {
43+ "name" : "Term list name" ,
44+ "description" : "Term list description" ,
45+ }
46+ )
47+ print ("List created:" )
48+ assert isinstance (custom_list , TermList )
49+ pprint (custom_list .as_dict ())
50+ list_id = custom_list .id
51+
52+ #
53+ # Update list details
54+ #
55+ print ("\n Updating details for list {}" .format (list_id ))
56+ updated_list = client .list_management_term_lists .update (
57+ list_id ,
58+ "application/json" ,
59+ {
60+ "name" : "New name" ,
61+ "description" : "New description"
62+ }
63+ )
64+ assert isinstance (updated_list , TermList )
65+ pprint (updated_list .as_dict ())
66+
67+ #
68+ # Add terms
69+ #
70+ print ("\n Adding terms to list {}" .format (list_id ))
71+ client .list_management_term .add_term (list_id , "term1" , "eng" )
72+ client .list_management_term .add_term (list_id , "term2" , "eng" )
73+
74+ #
75+ # Get all terms ids
76+ #
77+ print ("\n Getting all term IDs for list {}" .format (list_id ))
78+ terms = client .list_management_term .get_all_terms (list_id , "eng" )
79+ assert isinstance (terms , Terms )
80+ terms_data = terms .data
81+ assert isinstance (terms_data , TermsData )
82+ pprint (terms_data .as_dict ())
83+
84+ #
85+ # Refresh the index
86+ #
87+ print ("\n Refreshing the search index for list {}" .format (list_id ))
88+ refresh_index = client .list_management_term_lists .refresh_index_method (list_id , "eng" )
89+ assert isinstance (refresh_index , RefreshIndex )
90+ pprint (refresh_index .as_dict ())
91+
92+ print ("\n Waiting {} minutes to allow the server time to propagate the index changes." .format (LATENCY_DELAY ))
93+ time .sleep (LATENCY_DELAY * 60 )
94+
95+ #
96+ # Screen text
97+ #
98+ text = 'This text contains the terms "term1" and "term2".'
99+ print ('\n Screening text "{}" using term list {}' .format (text , list_id ))
100+ screen = client .text_moderation .screen_text (
101+ "eng" ,
102+ "text/plain" ,
103+ text ,
104+ autocorrect = False ,
105+ pii = False ,
106+ list_id = list_id
107+ )
108+ assert isinstance (screen , Screen )
109+ pprint (screen .as_dict ())
110+
111+
112+ #
113+ # Remove terms
114+ #
115+ term_to_remove = "term1"
116+ print ("\n Remove term {} from list {}" .format (term_to_remove , list_id ))
117+ client .list_management_term .delete_term (
118+ list_id ,
119+ term_to_remove ,
120+ "eng"
121+ )
122+
123+ #
124+ # Refresh the index
125+ #
126+ print ("\n Refreshing the search index for list {}" .format (list_id ))
127+ refresh_index = client .list_management_term_lists .refresh_index_method (list_id , "eng" )
128+ assert isinstance (refresh_index , RefreshIndex )
129+ pprint (refresh_index .as_dict ())
130+
131+ print ("\n Waiting {} minutes to allow the server time to propagate the index changes." .format (LATENCY_DELAY ))
132+ time .sleep (LATENCY_DELAY * 60 )
133+
134+ #
135+ # Re-Screen text
136+ #
137+ text = 'This text contains the terms "term1" and "term2".'
138+ print ('\n Screening text "{}" using term list {}' .format (text , list_id ))
139+ screen = client .text_moderation .screen_text (
140+ "eng" ,
141+ "text/plain" ,
142+ text ,
143+ autocorrect = False ,
144+ pii = False ,
145+ list_id = list_id
146+ )
147+ assert isinstance (screen , Screen )
148+ pprint (screen .as_dict ())
149+
150+ #
151+ # Delete all terms
152+ #
153+ print ("\n Delete all terms in the image list {}" .format (list_id ))
154+ client .list_management_term .delete_all_terms (list_id , "eng" )
155+
156+ #
157+ # Delete list
158+ #
159+ print ("\n Delete the term list {}" .format (list_id ))
160+ client .list_management_term_lists .delete (list_id )
161+
162+
163+ if __name__ == "__main__" :
164+ import sys , os .path
165+ sys .path .append (os .path .abspath (os .path .join (__file__ , ".." , ".." )))
166+ from tools import execute_samples
167+ execute_samples (globals (), SUBSCRIPTION_KEY_ENV_NAME )
0 commit comments