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", 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 texts are read in and processed;

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

56 Returns:

57 Nothing if `delete` is `True`.

58 If `name_only` 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 ```python

70 # list all available contexts:

71 receptiviti.norming()

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

74 receptiviti.norming("new_context")

75 ```

77 Send texts to establish the context, just like

78 the [receptiviti.request][receptiviti.request] function.

79 ```python

80 ## such as directly:

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

83 ## or from a file:

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

86 ## delete the new context:

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

88 ```

89 """

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

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

92 if name_only:

93 if verbose:

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

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

96 if req.status_code != 200:

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

98 raise RuntimeError(msg)

99 norms = req.json()

100 if norms and verbose:

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

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

103 return norms

104

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

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

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

108 raise RuntimeError(msg)

109

110 # list current context

111 if verbose:

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

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

114 if req.status_code != 200:

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

116 raise RuntimeError(msg)

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

118 if not name:

119 if len(norms):

120 if verbose:

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

122 print(

123 "custom norming context(s) found: "

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

125 )

126 elif verbose:

127 print("no custom norming contexts found")

128 return norms

129 context_id = "custom/" + name

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

131 if delete:

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

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

134 if res.status_code != 200:

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

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

137 )

138 raise RuntimeError(msg)

139 return None

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

141 if options:

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

143 elif delete:

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

145 return None

146 else:

147 if verbose:

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

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

150 if req.status_code != 200:

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

152 raise RuntimeError(msg)

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

154 if options:

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

156 if param not in status:

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

158 elif value != status[param]:

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

160 if verbose:

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

162 print(status)

163 if not text:

164 return status

165 status_step = status["status"]

166 if status_step == "completed":

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

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

169 if status_step == "pass_two":

170 first_pass = None

171 else:

172 if verbose:

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

174 _, first_pass, _ = _manage_request(

175 text=text,

176 **kwargs,

177 dotenv=dotenv,

178 key=key,

179 secret=secret,

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

181 to_norming=True,

182 )

183 second_pass = None

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

185 warnings.warn(

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

187 stacklevel=2,

188 )

189 else:

190 if verbose:

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

192 _, second_pass, _ = _manage_request(

193 text=text,

194 **kwargs,

195 dotenv=dotenv,

196 key=key,

197 secret=secret,

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

199 to_norming=True,

200 )

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

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

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