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

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

3import json

4import os

5import re

6import warnings

7from typing import 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[

28 None, "list[str]", pandas.DataFrame, pandas.Series, "dict[str, Union[pandas.Series, pandas.DataFrame, None]]"

29]:

30 """

31 View or Establish Custom Norming Contexts.

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

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

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

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

38 Args:

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

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

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

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

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

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

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

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

47 build-in contexts.

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

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

50 Passed to `readin_env` as `path`.

51 key (str): Your API key.

52 secret (str): Your API secret.

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

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

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

56 see [receptiviti.request][receptiviti.request].

58 Returns:

59 Nothing if `delete` is `True`.

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

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

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

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

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

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

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

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

70 Examples:

71 ```

72 # list all available contexts:

73 receptiviti.norming()

75 # list current custom contexts:

76 receptiviti.norming()

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

79 receptiviti.norming("new_context")

80 ```

82 Send tests to establish the context, just like

83 the [receptiviti.request][receptiviti.request] function.

84 ```

85 ## such as directly:

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

88 ## or from a file:

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

91 ## delete the new context:

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

93 ```

94 """

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 not name:

124 if len(norms):

125 if verbose:

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

127 print(

128 "custom norming context(s) found: "

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

130 )

131 elif verbose:

132 print("no custom norming contexts found")

133 return norms

134 context_id = "custom/" + name

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

136 if delete:

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

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

139 if res.status_code != 200:

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

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

142 )

143 raise RuntimeError(msg)

144 return None

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

146 if options:

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

148 elif delete:

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

150 return None

151 else:

152 if verbose:

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

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

155 if req.status_code != 200:

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

157 raise RuntimeError(msg)

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

159 if options:

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

161 if param not in status:

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

163 elif value != status[param]:

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

165 if verbose:

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

167 print(status)

168 if not text:

169 return status

170 status_step = status["status"]

171 if status_step == "completed":

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

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

174 if status_step == "pass_two":

175 first_pass = None

176 else:

177 if verbose:

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

179 _, first_pass, _ = _manage_request(

180 text=text,

181 **kwargs,

182 dotenv=dotenv,

183 key=key,

184 secret=secret,

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

186 to_norming=True,

187 )

188 second_pass = None

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

190 warnings.warn(

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

192 stacklevel=2,

193 )

194 else:

195 if verbose:

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

197 _, second_pass, _ = _manage_request(

198 text=text,

199 **kwargs,

200 dotenv=dotenv,

201 key=key,

202 secret=secret,

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

204 to_norming=True,

205 )

206 if second_pass is None or (second_pass["analyzed"] == 0).all():

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

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