datasources.upload.import_csv

  1"""
  2Custom data upload to create bespoke datasets
  3"""
  4import secrets
  5import hashlib
  6import time
  7import csv
  8import re
  9import io
 10
 11import datasources.upload.import_formats as import_formats
 12
 13from dateutil.parser import parse as parse_datetime
 14from datetime import datetime
 15
 16from backend.lib.processor import BasicProcessor
 17from common.lib.exceptions import QueryParametersException, QueryNeedsFurtherInputException, \
 18    QueryNeedsExplicitConfirmationException, CsvDialectException
 19from common.lib.helpers import strip_tags, sniff_encoding, UserInput, HashCache
 20
 21
 22class SearchCustom(BasicProcessor):
 23    type = "upload-search"  # job ID
 24    category = "Search"  # category
 25    title = "Custom Dataset Upload"  # title displayed in UI
 26    description = "Upload your own CSV file to be used as a dataset"  # description displayed in UI
 27    extension = "csv"  # extension of result file, used internally and in UI
 28    is_local = False  # Whether this datasource is locally scraped
 29    is_static = False  # Whether this datasource is still updated
 30
 31    max_workers = 1
 32    options = {
 33        "intro": {
 34            "type": UserInput.OPTION_INFO,
 35            "help": "You can upload a CSV or TAB file here that, after upload, will be available for further analysis "
 36                    "and processing. Files need to be [UTF-8](https://en.wikipedia.org/wiki/UTF-8)-encoded and must "
 37                    "contain a header row.\n\n"
 38                    "You can indicate what format the file has or upload one with arbitrary structure. In the latter "
 39                    "case, for each item, columns describing its ID, author, timestamp, and content are expected. You "
 40                    "can select which column holds which value after uploading the file."
 41        },
 42        "data_upload": {
 43            "type": UserInput.OPTION_FILE,
 44            "help": "File"
 45        },
 46        "format": {
 47            "type": UserInput.OPTION_CHOICE,
 48            "help": "CSV format",
 49            "options": {
 50                tool: info["name"] for tool, info in import_formats.tools.items()
 51            },
 52            "default": "custom"
 53        },
 54        "strip_html": {
 55            "type": UserInput.OPTION_TOGGLE,
 56            "help": "Strip HTML?",
 57            "default": False,
 58            "tooltip": "Removes HTML tags from the column identified as containing the item content ('body' by default)"
 59        }
 60    }
 61
 62    def process(self):
 63        """
 64        Process uploaded CSV file
 65
 66        Applies the provided mapping and makes sure the file is in a format
 67        4CAT will understand.
 68        """
 69        tool_format = import_formats.tools.get(self.parameters.get("format"))
 70        temp_file = self.dataset.get_results_path().with_suffix(".importing")
 71        with temp_file.open("rb") as infile:
 72            # detect encoding - UTF-8 with or without BOM
 73            encoding = sniff_encoding(infile)
 74
 75        # figure out the csv dialect
 76        # the sniffer is not perfect and sometimes makes mistakes
 77        # for some formats we already know the dialect, so we can override its
 78        # guess and set the properties as defined in import_formats.py
 79        infile = temp_file.open("r", encoding=encoding)
 80        sample = infile.read(1024 * 1024)
 81        try:
 82            possible_dialects = [csv.Sniffer().sniff(sample, delimiters=(",", ";", "\t"))]
 83        except csv.Error:
 84            possible_dialects = csv.list_dialects()
 85        if tool_format.get("csv_dialect", {}):
 86            # Known dialects are defined in import_formats.py
 87            dialect = csv.Sniffer().sniff(sample, delimiters=(",", ";", "\t"))
 88            for prop in tool_format.get("csv_dialect", {}):
 89                setattr(dialect, prop, tool_format["csv_dialect"][prop])
 90            possible_dialects.append(dialect)
 91
 92        while possible_dialects:
 93            # With validated csvs, save as is but make sure the raw file is sorted
 94            infile.seek(0)
 95            dialect = possible_dialects.pop() # Use the last dialect first
 96            self.dataset.log(f"Importing CSV file with dialect: {vars(dialect) if type(dialect) == csv.Dialect else dialect}")
 97            reader = csv.DictReader(infile, dialect=dialect)
 98
 99            if tool_format.get("columns") and not tool_format.get("allow_user_mapping") and set(reader.fieldnames) & \
100                    set(tool_format["columns"]) != set(tool_format["columns"]):
101                raise QueryParametersException("Not all columns are present")
102
103            # hasher for pseudonymisation
104            salt = secrets.token_bytes(16)
105            hasher = hashlib.blake2b(digest_size=24, salt=salt)
106            hash_cache = HashCache(hasher)
107
108            # write the resulting dataset
109            writer = None
110            done = 0
111            skipped = 0
112            with self.dataset.get_results_path().open("w", encoding="utf-8", newline="") as output_csv:
113                # mapper is defined in import_formats
114                try:
115                    for i, item in enumerate(tool_format["mapper"](reader, tool_format["columns"], self.dataset, self.parameters)):
116                        if isinstance(item, import_formats.InvalidImportedItem):
117                            # if the mapper returns this class, the item is not written
118                            skipped += 1
119                            if hasattr(item, "reason"):
120                                self.dataset.log(f"Skipping item ({item.reason})")
121                            continue
122
123                        if not writer:
124                            writer = csv.DictWriter(output_csv, fieldnames=list(item.keys()))
125                            writer.writeheader()
126
127                        if self.parameters.get("strip_html") and "body" in item:
128                            item["body"] = strip_tags(item["body"])
129
130                        # pseudonymise or anonymise as needed
131                        filtering = self.parameters.get("pseudonymise")
132                        try:
133                            if filtering:
134                                for field, value in item.items():
135                                    if field is None:
136                                        # This would normally be caught when writerow is called
137                                        raise CsvDialectException("Field is None")
138                                    if field.startswith("author"):
139                                        if filtering == "anonymise":
140                                            item[field] = "REDACTED"
141                                        elif filtering == "pseudonymise":
142                                            item[field] = hash_cache.update_cache(value)
143
144                            writer.writerow(item)
145                        except ValueError as e:
146                            if not possible_dialects:
147                                self.dataset.log(f"Error ({e}) writing item {i}: {item}")
148                                return self.dataset.finish_with_error("Could not parse CSV file. Have you selected the correct "
149                                                                      "format or edited the CSV after exporting? Try importing "
150                                                                      "as custom format.")
151                            else:
152                                raise CsvDialectException(f"Error ({e}) writing item {i}: {item}")
153
154                        done += 1
155
156                except import_formats.InvalidCustomFormat as e:
157                    self.log.warning(f"Unable to import improperly formatted file for {tool_format['name']}. See dataset "
158                                     "log for details.")
159                    infile.close()
160                    temp_file.unlink()
161                    return self.dataset.finish_with_error(str(e))
162
163                except UnicodeDecodeError as e:
164                    infile.close()
165                    temp_file.unlink()
166                    return self.dataset.finish_with_error("The uploaded file is not encoded with the UTF-8 character set. "
167                                                          "Make sure the file is encoded properly and try again.")
168
169                except CsvDialectException as e:
170                    self.dataset.log(f"Error with CSV dialect: {vars(dialect)}")
171                    continue
172
173            # done!
174            infile.close()
175            # We successfully read the CSV, no need to try other dialects
176            break
177
178        if skipped:
179            self.dataset.update_status(
180                f"CSV file imported, but {skipped:,} items were skipped because their date could not be parsed.",
181                is_final=True)
182
183        temp_file.unlink()
184        self.dataset.delete_parameter("filename")
185        if skipped and not done:
186            self.dataset.finish_with_error("No valid items could be found in the uploaded file. The column containing "
187                                           "the item's timestamp may be in a format that cannot be parsed properly.")
188        else:
189            self.dataset.finish(done)
190
191    def validate_query(query, request, user):
192        """
193        Validate custom data input
194
195        Confirms that the uploaded file is a valid CSV or tab file and, if so, returns
196        some metadata.
197
198        :param dict query:  Query parameters, from client-side.
199        :param request:  Flask request
200        :param User user:  User object of user who has submitted the query
201        :return dict:  Safe query parameters
202        """
203        # do we have an uploaded file?
204        if "option-data_upload" not in request.files:
205            raise QueryParametersException("No file was offered for upload.")
206
207        file = request.files["option-data_upload"]
208        if not file:
209            raise QueryParametersException("No file was offered for upload.")
210
211        if query.get("format") not in import_formats.tools:
212            raise QueryParametersException(f"Cannot import CSV from tool {query.get('format')}")
213
214        # content_length seems unreliable, so figure out the length by reading
215        # the file...
216        upload_size = 0
217        while True:
218            bit = file.read(1024)
219            if len(bit) == 0:
220                break
221            upload_size += len(bit)
222
223        file.seek(0)
224        encoding = sniff_encoding(file)
225        tool_format = import_formats.tools.get(query.get("format"))
226
227        try:
228            # try reading the file as csv here
229            # never read more than 128 kB (to keep it quick)
230            sample_size = min(upload_size, 128 * 1024)  # 128 kB is sent from the frontend at most
231            wrapped_file = io.TextIOWrapper(file, encoding=encoding)
232            sample = wrapped_file.read(sample_size)
233
234            if not csv.Sniffer().has_header(sample) and not query.get("frontend-confirm"):
235                # this may be intended, or the check may be bad, so allow user to continue
236                raise QueryNeedsExplicitConfirmationException(
237                    "The uploaded file does not seem to have a header row. Continue anyway?")
238
239            wrapped_file.seek(0)
240            dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
241
242            # override the guesses for specific formats if defined so in
243            # import_formats.py
244            for prop in tool_format.get("csv_dialect", {}):
245                setattr(dialect, prop, tool_format["csv_dialect"][prop])
246
247        except UnicodeDecodeError as e:
248            raise QueryParametersException("The uploaded file does not seem to be a CSV file encoded with UTF-8. "
249                                           "Save the file in the proper format and try again.")
250        except csv.Error:
251            raise QueryParametersException("Uploaded file is not a well-formed, UTF 8-encoded CSV or TAB file.")
252
253        # With validated csvs, save as is but make sure the raw file is sorted
254        reader = csv.DictReader(wrapped_file, dialect=dialect)
255
256        # we know that the CSV file is a CSV file now, next verify whether
257        # we know what each column means
258        try:
259            fields = reader.fieldnames
260        except UnicodeDecodeError:
261            raise QueryParametersException("The uploaded file is not a well-formed, UTF 8-encoded CSV or TAB file.")
262
263        incomplete_mapping = list(tool_format["columns"])
264        for field in tool_format["columns"]:
265            if tool_format.get("allow_user_mapping", False) and "option-mapping-%s" % field in request.form:
266                incomplete_mapping.remove(field)
267            elif not tool_format.get("allow_user_mapping", False) and field in fields:
268                incomplete_mapping.remove(field)
269
270        # offer the user a number of select boxes where they can indicate the
271        # mapping for each column
272        column_mapping = {}
273        if tool_format.get("allow_user_mapping", False):
274            magic_mappings = {
275                "id": {"__4cat_auto_sequence": "[generate sequential IDs]"},
276                "thread_id": {"__4cat_auto_sequence": "[generate sequential IDs]"},
277                "empty": {"__4cat_empty_value": "[empty]"},
278                "timestamp": {"__4cat_now": "[current date and time]"}
279            }
280            if incomplete_mapping:
281                raise QueryNeedsFurtherInputException({
282                    "mapping-info": {
283                        "type": UserInput.OPTION_INFO,
284                        "help": "Please confirm which column in the CSV file maps to each required value."
285                    },
286                    **{
287                        "mapping-%s" % mappable_column: {
288                            "type": UserInput.OPTION_CHOICE,
289                            "options": {
290                                "": "",
291                                **magic_mappings.get(mappable_column, magic_mappings["empty"]),
292                                **{column: column for column in fields}
293                            },
294                            "default": mappable_column if mappable_column in fields else "",
295                            "help": mappable_column,
296                            "tooltip": tool_format["columns"][mappable_column]
297                        } for mappable_column in incomplete_mapping
298                    }})
299
300            # the mappings do need to point to a column in the csv file
301            missing_mapping = []
302            for field in tool_format["columns"]:
303                mapping_field = "option-mapping-%s" % field
304                provided_field = request.form.get(mapping_field)
305                if (provided_field not in fields and not provided_field.startswith("__4cat")) or not provided_field:
306                    missing_mapping.append(field)
307                else:
308                    column_mapping["mapping-" + field] = request.form.get(mapping_field)
309
310            if missing_mapping:
311                raise QueryParametersException(
312                    "You need to indicate which column in the CSV file holds the corresponding value for the following "
313                    "columns: %s" % ", ".join(missing_mapping))
314
315        elif incomplete_mapping:
316            raise QueryParametersException("The CSV file does not contain all required columns. The following columns "
317                                           "are missing: %s" % ", ".join(incomplete_mapping))
318
319        # the timestamp column needs to be parseable
320        timestamp_column = request.form.get("mapping-timestamp")
321        try:
322            row = reader.__next__()
323            if timestamp_column not in row:
324                # incomplete row because we are analysing a sample
325                # stop parsing because no complete rows will follow
326                raise StopIteration
327
328            try:
329                if row[timestamp_column].isdecimal():
330                    datetime.fromtimestamp(float(row[timestamp_column]))
331                else:
332                    parse_datetime(row[timestamp_column])
333            except (ValueError, OSError):
334                raise QueryParametersException(
335                    "Your 'timestamp' column does not use a recognisable format (yyyy-mm-dd hh:mm:ss is recommended)")
336
337        except StopIteration:
338            pass
339
340        # ok, we're done with the file
341        wrapped_file.detach()
342
343        # Whether to strip the HTML tags
344        strip_html = False
345        if query.get("strip_html"):
346            strip_html = True
347
348        # return metadata - the filename is sanitised and serves no purpose at
349        # this point in time, but can be used to uniquely identify a dataset
350        disallowed_characters = re.compile(r"[^a-zA-Z0-9._+-]")
351        return {
352            "filename": disallowed_characters.sub("", file.filename),
353            "time": time.time(),
354            "datasource": "upload",
355            "board": query.get("format", "custom").replace("_", "-"),
356            "format": query.get("format"),
357            "strip_html": strip_html,
358            **column_mapping,
359        }
360
361    def after_create(query, dataset, request):
362        """
363        Hook to execute after the dataset for this source has been created
364
365        In this case, put the file in a temporary location so it can be
366        processed properly by the related Job later.
367
368        :param dict query:  Sanitised query parameters
369        :param DataSet dataset:  Dataset created for this query
370        :param request:  Flask request submitted for its creation
371        """
372        file = request.files["option-data_upload"]
373        file.seek(0)
374        with dataset.get_results_path().with_suffix(".importing").open("wb") as outfile:
375            while True:
376                chunk = file.read(1024)
377                if len(chunk) == 0:
378                    break
379                outfile.write(chunk)