Coverage for src/receptiviti/norming.py: 77%

1"""Interact with the norming endpoint.""" 

2 

3import json 

4import os 

5import re 

6import warnings 

7from typing import Dict, List, Union 

8 

9import pandas 

10import requests 

11import requests.auth 

12 

13from receptiviti.manage_request import _manage_request, _resolve_request_def 

14 

15 

16def norming( 

17 name: Union[str, None] = None, 

18 text: Union[str, List[str], pandas.DataFrame, None] = None, 

19 options: Union[Dict[str, Union[str, int]], None] = None, 

20 delete: bool = False, 

21 name_only: bool = False, 

22 dotenv: Union[bool, str] = True, 

23 key: str = os.getenv("RECEPTIVITI_KEY", ""), 

24 secret: str = os.getenv("RECEPTIVITI_SECRET", ""), 

25 url: str = os.getenv("RECEPTIVITI_URL", ""), 

26 verbose: bool = True, 

27 **kwargs, 

28) -> Union[ 

29 None, List[str], pandas.DataFrame, pandas.Series, Dict[str, Union[str, pandas.Series, pandas.DataFrame, None]] 

30]: 

31 """ 

32 View or Establish Custom Norming Contexts. 

33 

34 Custom norming contexts can be used to process later texts by specifying the 

35 `custom_context` API argument in the `receptiviti.request` function (e.g., 

36 `receptiviti.request("text to score", version="v2", custom_context="norm_name")`, 

37 where `norm_name` is the name you set here). 

38 

39 Args: 

40 name (str): Name of a new norming context, to be established from the provided 'text'. 

41 Not providing a name will list the previously created contexts. 

42 text (str): Text to be processed and used as the custom norming context. 

43 Not providing text will return the status of the named norming context. 

44 options (dict): Options to set for the norming context (e.g., 

45 `{"min_word_count": 350, "max_punctuation": .25}`). 

46 delete (bool): If `True`, will request removal of the `name` context. 

47 name_only (bool): If `True`, will return a list of context names only, including those of 

48 build-in contexts. 

49 dotenv (bool | str): Path to a .env file to read environment variables from. By default, 

50 will for a file in the current directory or `~/Documents`. 

51 Passed to `readin_env` as `path`. 

52 key (str): Your API key. 

53 secret (str): Your API secret. 

54 url (str): The URL of the API; defaults to `https://api.receptiviti.com`. 

55 verbose (bool): If `False`, will not show status messages. 

56 **kwargs (Any): Additional arguments to specify how texts are read in and processed 

57 (excluding than `cores`); 

58 see [receptiviti.request][receptiviti.request]. 

59 

60 Returns: 

61 Nothing if `delete` is `True`. 

62 If `name_only` is `True`, a `list` containing context names (built-in and custom). 

63 Otherwise, either a `pandas.DataFrame` containing all existing custom context statuses 

64 (if no `name` is specified), a `pandas.Series` containing the the status of 

65 `name` (if `text` is not specified), a dictionary: 

66 

67 - `initial_status`: Initial status of the context. 

68 - `first_pass`: Response after texts are sent the first time, or 

69 `None` if the initial status is `pass_two`. 

70 - `second_pass`: Response after texts are sent the second time. 

71 

72 Examples: 

73 ```python 

74 # list all available contexts: 

75 receptiviti.norming() 

76 

77 # create or get the status of a single context: 

78 receptiviti.norming("new_context") 

79 ``` 

80 

81 Send texts to establish the context, just like 

82 the [receptiviti.request][receptiviti.request] function. 

83 ```python 

84 ## such as directly: 

85 receptiviti.norming("new_context", ["text to send", "another text"]) 

86 

87 ## or from a file: 

88 receptiviti.norming("new_context", "./path/to/file.csv", text_column = "text") 

89 

90 ## delete the new context: 

91 receptiviti.norming("new_context", delete=True) 

92 ``` 

93 """ 

94 kwargs["cores"] = 1 

95 _, url, key, secret = _resolve_request_def(url, key, secret, dotenv) 

96 auth = requests.auth.HTTPBasicAuth(key, secret) 

97 if name_only: 

98 if verbose: 

99 print("requesting list of existing custom norming contests") 

100 req = requests.get(url + "/v2/norming/", auth=auth, timeout=9999) 

101 if req.status_code != 200: 

102 msg = f"failed to make norming list request: {req.status_code} {req.reason}" 

103 raise RuntimeError(msg) 

104 norms = req.json() 

105 if norms and verbose: 

106 custom_prefix = re.compile("^custom/") 

107 print("available norming context(s): " + ", ".join([custom_prefix.sub("", name) for name in norms])) 

108 return norms 

109 

110 url += "/v2/norming/custom/" 

111 if name and re.search("[^a-z0-9_.-]", name): 

112 msg = "`name` can only include lowercase letters, numbers, hyphens, underscores, or periods" 

113 raise RuntimeError(msg) 

114 

115 # list current context 

116 if verbose: 

117 print("requesting list of existing custom norming contests") 

118 req = requests.get(url, auth=auth, timeout=9999) 

119 if req.status_code != 200: 

120 msg = f"failed to make custom norming list request: {req.status_code} {req.reason}" 

121 raise RuntimeError(msg) 

122 norms = pandas.json_normalize(req.json()) 

123 if len(norms) and "name" not in norms: 

124 msg = "`name` column not found in response with columns " + ", ".join(norms.columns) 

125 raise RuntimeError(msg) 

126 if not name: 

127 if len(norms): 

128 if verbose: 

129 custom_prefix = re.compile("^custom/") 

130 print( 

131 "custom norming context(s) found: " 

132 + ", ".join([custom_prefix.sub("", name) for name in norms["name"]]) 

133 ) 

134 elif verbose: 

135 print("no custom norming contexts found") 

136 return norms 

137 context_id = "custom/" + name 

138 if len(norms) and context_id in norms["name"].values: 

139 if delete: 

140 res = requests.delete(url + name, auth=auth, timeout=9999) 

141 content = res.json() if res.text[:1] == "[" else {"message": res.text} 

142 if res.status_code != 200: 

143 msg = f"Request Error ({res.status_code!s})" + ( 

144 (" (" + str(content["code"]) + ")" if "code" in content else "") + ": " + content["message"] 

145 ) 

146 raise RuntimeError(msg) 

147 return None 

148 status = norms[norms["name"] == context_id].iloc[0] 

149 if options: 

150 warnings.warn(UserWarning(f"context {name} already exists, so options do not apply"), stacklevel=2) 

151 elif delete: 

152 print(f"context {name} does not exist") 

153 return None 

154 else: 

155 if verbose: 

156 print(f"requesting creation of context {name}") 

157 req = requests.post(url, json.dumps({"name": name, **(options if options else {})}), auth=auth, timeout=9999) 

158 if req.status_code != 200: 

159 msg = f"failed to make norming creation request: {req.json().get('error', 'reason unknown')}" 

160 raise RuntimeError(msg) 

161 status = pandas.json_normalize(req.json()).iloc[0] 

162 if options: 

163 for param, value in options.items(): 

164 if param in status and value != status[param]: 

165 warnings.warn(UserWarning(f"set option {param} does not match the requested value"), stacklevel=2) 

166 if verbose: 

167 print(f"status of {name}:") 

168 print(status) 

169 if text is None: 

170 return status 

171 status_step = status["status"] 

172 if status_step != "created": 

173 warnings.warn(UserWarning("status is not `created`, so cannot send text"), stacklevel=2) 

174 return {"initial_status": status, "first_pass": None, "second_pass": None} 

175 if verbose: 

176 print(f"sending first-pass sample for {name}") 

177 _, first_pass, _ = _manage_request( 

178 text=text, 

179 **kwargs, 

180 dotenv=dotenv, 

181 key=key, 

182 secret=secret, 

183 url=f"{url}{name}/one", 

184 to_norming=True, 

185 ) 

186 second_pass = None 

187 if first_pass is not None and (first_pass["analyzed_samples"] == 0).all(): 

188 warnings.warn( 

189 UserWarning("no texts were successfully analyzed in the first pass, so second pass was skipped"), 

190 stacklevel=2, 

191 ) 

192 else: 

193 if verbose: 

194 print(f"sending second-pass samples for {name}") 

195 _, second_pass, _ = _manage_request( 

196 text=text, 

197 **kwargs, 

198 dotenv=dotenv, 

199 key=key, 

200 secret=secret, 

201 url=f"{url}{name}/two", 

202 to_norming=True, 

203 ) 

204 if second_pass is None or (second_pass["analyzed_samples"] == 0).all(): 

205 warnings.warn(UserWarning("no texts were successfully analyzed in the second pass"), stacklevel=2) 

206 return {"initial_stats": status, "first_pass": first_pass, "second_pass": second_pass}