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

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

3import json

4import os

5import re

6import warnings

7from typing import Dict, List, Union

9import pandas

10import requests

12from receptiviti.manage_request import _manage_request, _resolve_request_def

15def norming(

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

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

18 options: Union[dict, None] = None,

19 delete=False,

20 name_only=False,

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

22 key=os.getenv("RECEPTIVITI_KEY", ""),

23 secret=os.getenv("RECEPTIVITI_SECRET", ""),

24 url=os.getenv("RECEPTIVITI_URL", ""),

25 verbose=True,

26 **kwargs,

27) -> Union[None, List[str], pandas.DataFrame, pandas.Series, Dict[str, Union[pandas.Series, pandas.DataFrame, None]]]:

28 """

29 View or Establish Custom Norming Contexts.

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

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

33 `receptiviti.request("text to score", version = "v2", options = {"custom_context": "norm_name"})`,

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

36 Args:

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

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

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

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

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

42 `{"word_count_filter": 350, "punctuation_filter": .25}`).

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

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

45 build-in contexts.

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

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

48 Passed to `readin_env` as `path`.

49 key (str): Your API key.

50 secret (str): Your API secret.

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

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

53 **kwargs (Any): Additional arguments to specify how tests are read in and processed;

54 see [receptiviti.request][receptiviti.request].

56 Returns:

57 Nothing if `delete` is `True`.

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

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

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

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

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

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

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

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

68 Examples:

69 ```

70 # list all available contexts:

71 receptiviti.norming()

73 # list current custom contexts:

74 receptiviti.norming()

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

77 receptiviti.norming("new_context")

78 ```

80 Send tests to establish the context, just like

81 the [receptiviti.request][receptiviti.request] function.

82 ```

83 ## such as directly:

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

86 ## or from a file:

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

89 ## delete the new context:

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

91 ```

92 """

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

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

95 if name_only:

96 if verbose:

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

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

99 if req.status_code != 200:

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

101 raise RuntimeError(msg)

102 norms = req.json()

103 if norms and verbose:

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

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

106 return norms

107

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

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

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

111 raise RuntimeError(msg)

112

113 # list current context

114 if verbose:

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

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

117 if req.status_code != 200:

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

119 raise RuntimeError(msg)

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

121 if not name:

122 if len(norms):

123 if verbose:

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

125 print(

126 "custom norming context(s) found: "

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

128 )

129 elif verbose:

130 print("no custom norming contexts found")

131 return norms

132 context_id = "custom/" + name

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

134 if delete:

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

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

137 if res.status_code != 200:

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

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

140 )

141 raise RuntimeError(msg)

142 return None

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

144 if options:

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

146 elif delete:

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

148 return None

149 else:

150 if verbose:

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

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

153 if req.status_code != 200:

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

155 raise RuntimeError(msg)

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

157 if options:

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

159 if param not in status:

160 warnings.warn(UserWarning(f"option {param} was not set"), stacklevel=2)

161 elif value != status[param]:

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

163 if verbose:

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

165 print(status)

166 if not text:

167 return status

168 status_step = status["status"]

169 if status_step == "completed":

170 warnings.warn(UserWarning("status is `completes`, so cannot send text"), stacklevel=2)

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

172 if status_step == "pass_two":

173 first_pass = None

174 else:

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"] == 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"] == 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}