diff --git a/backend/lib/preset.py b/backend/lib/preset.py
index af9d2beeb..ee4d5635e 100644
--- a/backend/lib/preset.py
+++ b/backend/lib/preset.py
@@ -23,13 +23,19 @@ def process(self):
pipeline = self.format_linear_pipeline(pipeline)
+ # the deeper pipeline steps are checked as they are created in
+ # after_process; the first step is created here, so check it here
+ self.warn_unexpected_parameters(
+ self.modules.processors.get(pipeline[0]["type"]), pipeline[0]["parameters"],
+ self.dataset, self.config, self.log)
+
analysis_pipeline = DataSet(
parameters=pipeline[0]["parameters"],
db=self.db,
type=pipeline[0]["type"],
owner=self.dataset.creator,
is_private=self.dataset.is_private,
- parent=self.dataset.key,
+ parent=self.dataset.key,
modules=self.modules)
# give same ownership as parent dataset
diff --git a/backend/lib/processor.py b/backend/lib/processor.py
index 616ffccaf..5122bee37 100644
--- a/backend/lib/processor.py
+++ b/backend/lib/processor.py
@@ -114,6 +114,29 @@ def is_compatible_with(cls, module=None, config=None):
#: evaluated from it.
compatibility = None
+ #: Keys the framework injects into validate_query's input that are not
+ #: dataset parameters and must never be stored. validate_query may read
+ #: them, but get_validated_query strips them from the result. These are
+ #: only the transient queue-time signals that pass *through* validate_query;
+ #: functional keys added elsewhere (datasource, type, email-complete,
+ #: pseudonymise) and provenance keys (copied_from, producer_type) are not
+ #: listed here because they are meaningful and are added downstream of this.
+ TRANSIENT_QUERY_KEYS = ("frontend-confirm",)
+
+ #: Parameter keys the framework itself uses on a dataset (as opposed to a
+ #: processor's own options), so they are expected on a queued processor
+ #: even though no get_options() declares them. Used by
+ #: warn_unexpected_parameters(); keys starting with `_` (internal
+ #: cross-processor plumbing) are also treated as expected.
+ FRAMEWORK_PARAMETER_KEYS = ("next", "attach_to", "copy_to")
+
+ #: Parameters this processor reads but does not (always) declare in
+ #: get_options() — e.g. an option offered only under certain config, or a
+ #: value a preset deliberately passes it. Override per-processor to state
+ #: this input surface; these keys are exempt from
+ #: warn_unexpected_parameters().
+ accepted_parameters = tuple()
+
def work(self):
"""
Process a dataset
@@ -198,22 +221,33 @@ def work(self):
# Add source dataset to cleanup list to remove disposable files
self.for_cleanup.append(self.source_dataset)
- # get parameters
- # if possible, fill defaults where parameters are not provided
+ # build the single runtime view of this dataset's parameters: every
+ # stored parameter, plus the declared default for any option that was
+ # not stored, so worker code can read any declared option without
+ # checking whether it exists. The stored parameters themselves are
+ # kept as self.given_parameters: they hold only the options that were
+ # actually part of the submission (see option_given()). Worker code
+ # should read options from self.parameters, and not re-read them from
+ # the dataset mid-run — stored parameters may change during a run,
+ # but these dictionaries stay stable.
given_parameters = self.dataset.parameters.copy()
all_parameters = self.get_options(self.source_dataset, config=self.config)
+ self.given_parameters = given_parameters
self.parameters = {
param: given_parameters.get(param, all_parameters.get(param, {}).get("default"))
for param in [*all_parameters.keys(), *given_parameters.keys()]
}
- # now the parameters have been loaded into memory, clear any sensitive
- # ones. This has a side-effect that a processor may not run again
- # without starting from scratch, but this is the price of progress
- options = self.get_options(self.dataset.get_parent(), config=self.config)
- for option, option_settings in options.items():
- if option_settings.get("sensitive"):
- self.dataset.delete_parameter(option)
+ # the values of sensitive options (e.g. API keys) stay available in
+ # self.parameters for the duration of this run, but are removed from
+ # the stored dataset record right away, so they spend as little time
+ # in the database as possible. The deliberate trade-off is that a run
+ # that is interrupted and retried later no longer has these values.
+ try:
+ self.dataset.remove_sensitive_parameters(config=self.config)
+ except Exception as e:
+ # failing to remove these values should not fail the dataset
+ self.log.warning(f"Could not remove sensitive parameters of dataset {self.dataset.key}: {e}")
if self.interrupted:
self.dataset.log("Processing interrupted, trying again later")
@@ -223,7 +257,7 @@ def work(self):
try:
self.process()
self.after_process()
-
+
# processors should usually finish their jobs by themselves, but if
# the worker finished without errors, the job can be finished in
# any case
@@ -299,6 +333,8 @@ def after_process(self):
next_type, self.dataset.key))
elif next_type in available_processors:
+ self.warn_unexpected_parameters(
+ available_processors[next_type], next_parameters, self.dataset, self.config, self.log)
next_analysis = DataSet(
parameters=next_parameters,
type=next_type,
@@ -341,6 +377,14 @@ def after_process(self):
"Cannot find preset's source dataset for dataset %s" % self.dataset.key)
break
+ # the follow-up datasets now hold their own copy of any sensitive
+ # values that were queued for them, so remove those values from this
+ # dataset's stored `next` chain, where they would otherwise remain
+ try:
+ self.dataset.remove_sensitive_parameters_from_next(config=self.config)
+ except Exception as e:
+ self.log.warning(f"Could not remove sensitive next parameters of dataset {self.dataset.key}: {e}")
+
# see if we need to register the result somewhere
if "copy_to" in self.parameters:
# copy the results to an arbitrary place that was passed
@@ -431,6 +475,29 @@ def after_process(self):
except (SMTPException, ConnectionRefusedError, socket.timeout):
self.log.error("Error sending email to %s" % owner)
+ def option_given(self, option):
+ """
+ Check whether an option was actually part of this dataset's submission
+
+ `self.parameters` always contains every option this worker declares,
+ with defaults filled in for anything that was not stored. That is
+ convenient for reading values, but it cannot tell you whether the
+ option was really part of the submission. This method can: it checks
+ the parameters as they were stored when the dataset was created. For
+ datasets queued via the web interface, an option is only stored if
+ the user could see it when submitting (i.e. its `requires` condition
+ was met); for datasets queued by other code (e.g. presets), an option
+ is only stored if the calling code explicitly set it.
+
+ Note that a stored option may still hold its default value: a user
+ who saw an option and left it untouched still counts as having been
+ given it.
+
+ :param str option: Option name, as used in `get_options()`
+ :return bool: Whether the option was part of the stored parameters
+ """
+ return option in self.given_parameters
+
def clean_up_on_error(self):
try:
# ensure proxied requests are stopped
@@ -1051,6 +1118,150 @@ class attribute is still supported for backwards compatibility (returned
return copy.deepcopy(cls.options) if hasattr(cls, "options") else {}
+ @staticmethod
+ def validate_query(query, request, config):
+ """
+ Check and finalise user input before a dataset is created
+
+ The web tool calls this before creating a dataset for this worker.
+ `query` contains the values the user entered for the options from
+ `get_options()`: parsed, with defaults filled in for options the user
+ did not touch, plus `frontend-confirm` (True when the user has already
+ answered a confirmation pop-up for this submission). Options that were
+ hidden because their `requires` condition was not met are absent.
+
+ Implementations can check these values and:
+
+ - raise QueryParametersException("message") to reject the input; the
+ message is shown next to the form;
+ - raise QueryNeedsExplicitConfirmationException("question") to show an
+ OK/cancel pop-up; when the user accepts, the form is re-submitted
+ with `frontend-confirm` set, so only raise this when that key is not
+ set;
+ - raise QueryNeedsFurtherInputException(config={...}) to add extra
+ fields to the form (same format as get_options() values), after
+ which it is re-submitted.
+
+ Whatever this method returns is stored as the new dataset's
+ parameters. 4CAT adds a few keys of its own, and the values of options
+ marked `sensitive` are removed from the stored record when the dataset
+ starts running. At run time, the worker reads all of this back through
+ `self.parameters`, with declared option defaults filled in for any
+ missing keys; `option_given()` tells a worker whether an option was
+ really part of the stored submission.
+
+ When rebuilding the returned dictionary instead of editing the given
+ one, copy optional keys only when they are present in the input: a
+ pattern like `"option": query.get("option")` stores a meaningless
+ None for options the user was never given. Keys you leave untouched
+ in an edited dictionary keep this property automatically.
+
+ By default nothing is checked and the input is stored as-is. Search
+ workers must define their own version: doing so is what makes a
+ datasource queryable from the web interface.
+
+ :param dict query: Parsed user input for this worker's options
+ :param request: Flask request the input arrived in, for e.g. access
+ to uploaded files
+ :param config: Configuration reader, scoped to the queueing user
+ :return dict: The parameters to store with the new dataset
+ """
+ return query
+
+ @classmethod
+ def get_validated_query(cls, query, request, config, log=None):
+ """
+ Run this worker's validate_query and check the result
+
+ Code that turns user input into a dataset should call this instead of
+ calling validate_query directly, so that the checks below apply no
+ matter where a dataset is created from. Future rules about
+ validate_query itself (for example, warning when a worker defines no
+ validation of its own) also belong here, so that they automatically
+ cover every place datasets are created.
+
+ Currently checked: a returned key that was not part of the submitted
+ input and holds None is almost always `query.get()` on an option the
+ user was never given, which stores a meaningless value in the dataset
+ record. Such keys are logged so the worker can be fixed.
+
+ :param dict query: Parsed user input, as passed to validate_query
+ :param request: Flask request the input arrived in
+ :param config: Configuration reader, scoped to the queueing user
+ :param log: Logger to report suspicious results to, if any
+ :return dict: The parameters to store with the new dataset
+ """
+ sanitised = cls.validate_query(query, request, config)
+
+ if sanitised is None:
+ raise ProcessorException(f"validate_query of {cls.type} returned nothing; it must return the "
+ "dictionary of parameters to store with the dataset")
+
+ # drop framework-injected signals that validate_query may have read but
+ # that are not parameters; storing them would clutter the record
+ for key in cls.TRANSIENT_QUERY_KEYS:
+ sanitised.pop(key, None)
+
+ if log:
+ junk = [option for option, value in sanitised.items() if value is None and option not in query]
+ if junk:
+ log.warning(f"validate_query of {cls.type} stored None for options that were not part of the "
+ f"submission ({', '.join(junk)}); such keys should only be stored when they are given")
+
+ return sanitised
+
+ @staticmethod
+ def warn_unexpected_parameters(processor, parameters, parent_dataset, config, log):
+ """
+ Warn when a processor is queued with parameters it does not use
+
+ Datasets created programmatically (preset pipelines and `next` chains)
+ are not validated the way user-submitted queries are, so a parameter
+ meant for a different processor - or simply misspelled - is silently
+ ignored at run time. This logs a warning for any parameter key the
+ target processor does not declare in get_options(), so such mistakes
+ are visible. It never raises: the value is developer-authored, and
+ dropping a stale key is better than breaking the chain.
+
+ Keys the framework itself uses (FRAMEWORK_PARAMETER_KEYS), keys the
+ target lists in its `accepted_parameters` (things it reads but does not
+ always offer as an option), and keys starting with `_` (internal
+ cross-processor plumbing, not user options) are expected and not warned
+ about.
+
+ This only runs when the `dev.mode` setting is enabled: the warnings are
+ a developer aid (a processor may legitimately read a parameter it does
+ not declare), so they are off by default and can be toggled live in the
+ control panel.
+
+ :param processor: The processor class the dataset is queued for.
+ :param dict parameters: The parameters the dataset is queued with.
+ :param parent_dataset: The dataset the processor will run on, used to
+ resolve options whose availability depends on it.
+ :param config: Configuration reader.
+ :param log: Logger to warn to.
+ """
+ if not processor or not log:
+ return
+
+ if not config or not config.get("dev.mode", False):
+ return
+
+ try:
+ known = set(processor.get_options(parent_dataset, config=config))
+ except Exception:
+ # if the options can't be determined, don't guess at what's unexpected
+ return
+
+ accepted = set(getattr(processor, "accepted_parameters", ()) or ())
+
+ for key in parameters:
+ if key in known or key in accepted \
+ or key in BasicProcessor.FRAMEWORK_PARAMETER_KEYS or key.startswith("_"):
+ continue
+ log.warning(f"Processor {processor.type} was queued with parameter '{key}', which it does not "
+ "use; the value will be ignored.")
+
@classmethod
def get_status(cls):
"""
diff --git a/backend/lib/search.py b/backend/lib/search.py
index c79849f51..1f8858dc1 100644
--- a/backend/lib/search.py
+++ b/backend/lib/search.py
@@ -59,7 +59,12 @@ def process(self):
are marked as finished.
"""
- query_parameters = self.dataset.get_parameters()
+ # search workers read all parameters from a single dictionary: the
+ # dataset's stored parameters, with the declared default filled in for
+ # any option that was not stored (prepared in work()). get_items()
+ # receives this same dictionary as its `query` argument, so `query`
+ # and `self.parameters` always agree.
+ query_parameters = self.parameters
results_file = self.dataset.get_results_path()
self.log.info("Querying: %s" % str({k: v for k, v in query_parameters.items() if not self.get_options(
@@ -106,7 +111,9 @@ def search(self, query):
class. This method just provides some scaffolding and processing
of results via `after_search()`, if it is defined.
- :param dict query: Query parameters
+ :param dict query: Query parameters: the dataset's stored parameters
+ with declared option defaults filled in for missing keys. This is
+ the same dictionary as `self.parameters`.
:return: Iterable of matching items, or None if there are no results.
"""
items = self.get_items(query)
@@ -128,7 +135,10 @@ def get_items(self, query):
To be implemented by descending classes!
- :param dict query: Query parameters
+ :param dict query: Query parameters: the dataset's stored parameters
+ with declared option defaults filled in for missing keys. This is
+ the same dictionary as `self.parameters`; read options from either,
+ they always agree.
:return Generator: A generator or iterable that returns items
collected according to the provided parameters.
"""
diff --git a/backend/workers/expire_items.py b/backend/workers/expire_items.py
index 70657d238..48df33c3c 100644
--- a/backend/workers/expire_items.py
+++ b/backend/workers/expire_items.py
@@ -35,7 +35,12 @@ class ThingExpirer(BasicWorker):
type = "expire-datasets"
max_workers = 1
- expiry_notification_after_days = 7
+ expiry_notification_after_days = 7
+
+ # datasets that have not finished after this many seconds are considered
+ # abandoned: their sensitive option values (e.g. API keys) are removed,
+ # which normally happens as soon as a dataset starts running
+ abandoned_dataset_age = 7 * 86400
@classmethod
def ensure_job(cls, config=None):
@@ -57,6 +62,7 @@ def work(self):
self.expire_datasets()
self.expire_users()
self.expire_notifications()
+ self.expire_sensitive_parameters()
self.job.finish()
@@ -115,6 +121,51 @@ def expire_datasets(self):
finally:
self.db.commit()
+ def expire_sensitive_parameters(self):
+ """
+ Remove sensitive option values from abandoned datasets
+
+ Sensitive option values (e.g. API keys) are removed from a dataset's
+ stored parameters as soon as it starts running (top-level values), and
+ from its `next` chain once its follow-up datasets are created. A
+ dataset that is created but never picked up by a worker - for example
+ because the backend was stopped for good before it could run, or one
+ that keeps crashing before finishing - would keep those values forever,
+ so remove both here once the dataset is old enough to be considered
+ abandoned. An abandoned dataset never created its follow-ups, so
+ clearing its `next` chain is safe.
+ """
+ cutoff = int(time.time()) - self.abandoned_dataset_age
+ abandoned = self.db.fetchall(
+ "SELECT * FROM datasets WHERE is_finished = FALSE AND timestamp < %s",
+ (cutoff,))
+ self.log.debug(f"Found {len(abandoned)} abandoned datasets; cleaning up sensitive parameters if present.")
+
+ wrappers = {}
+ for dataset_data in abandoned:
+ if self.interrupted:
+ raise WorkerInterruptedException("Interrupted while cleaning up abandoned datasets")
+
+ try:
+ # which options exist (and which are sensitive) can depend on
+ # the dataset owner's configuration context
+ if dataset_data["creator"] not in wrappers:
+ wrappers[dataset_data["creator"]] = ConfigWrapper(
+ self.config, user=User.get_by_name(self.db, dataset_data["creator"])
+ )
+
+ dataset = DataSet(data=dataset_data, db=self.db, modules=self.modules, check_owners=False)
+ dataset.remove_sensitive_parameters(config=wrappers[dataset_data["creator"]])
+ dataset.remove_sensitive_parameters_from_next(config=wrappers[dataset_data["creator"]])
+
+ except DataSetNotFoundException:
+ # deleted in the meantime
+ continue
+
+ except Exception as e:
+ # cleaning up must not crash the worker; try again next run
+ self.log.warning(f"Could not remove sensitive parameters of dataset {dataset_data['key']}: {e}")
+
def expire_users(self):
"""
Delete expired users
diff --git a/common/lib/config_definition.py b/common/lib/config_definition.py
index 3de7b3f99..821b92bb3 100644
--- a/common/lib/config_definition.py
+++ b/common/lib/config_definition.py
@@ -339,6 +339,15 @@
"tooltip": "Slack callback URL to use for alerts",
"global": True
},
+ # developer options
+ "dev.mode": {
+ "type": UserInput.OPTION_TOGGLE,
+ "default": False,
+ "help": "Developer mode",
+ "tooltip": "Enables extra developer-facing diagnostics that are not needed for normal operation, e.g. additional "
+ "logging useful when creating new data sources or processors.",
+ "global": True
+ },
"mail.admin_email": {
"type": UserInput.OPTION_TEXT,
"default": "",
diff --git a/common/lib/dataset.py b/common/lib/dataset.py
index f29081399..88f042451 100644
--- a/common/lib/dataset.py
+++ b/common/lib/dataset.py
@@ -1552,6 +1552,8 @@ def change_datasource(self, datasource):
data={"parameters": json.dumps(self.parameters)},
where={"key": self.key},
)
+ # keep data["parameters"] consistent, since get_parameters() reads it
+ self.data["parameters"] = json.dumps(self.parameters)
return datasource
def reserve_result_file(self, parameters=None, extension="csv"):
@@ -1866,10 +1868,108 @@ def delete_parameter(self, parameter, instant=True):
)
if instant:
+ # keep both in-memory stores consistent with the database:
+ # get_parameters() reads data["parameters"], so updating only
+ # self.parameters would leave it returning the deleted value
self.parameters = parameters
+ self.data["parameters"] = json.dumps(parameters)
return updated > 0
+ def remove_sensitive_parameters(self, config):
+ """
+ Delete stored values of options marked as sensitive
+
+ Workers can mark an option as "sensitive" in their option definitions.
+ This is used for values such as API keys, which should spend as little
+ time in the database as possible. This method deletes those values
+ from the stored parameters. It is called as soon as a dataset starts
+ running (the worker keeps the values in memory for the run), and
+ periodically for datasets that were created but never picked up by a
+ worker.
+
+ :param config: Configuration reader, used to determine the worker's
+ options. Which options exist (and which are sensitive) can depend on
+ the configuration.
+ """
+ # check the options of both the processor that created this dataset
+ # and the processor matching its current type. These can differ after
+ # adopt_type() has rewritten the type. get_own_processor is purely
+ # defensive: the record is scrubbed before a type is ever adopted, and
+ # the periodic cleanup only sees datasets that never ran - but this
+ # keeps the check correct for future callers.
+ for worker in {self.get_producer_processor(), self.get_own_processor()}:
+ if not worker:
+ continue
+
+ for option, settings in worker.get_options(self.get_parent(), config=config).items():
+ if settings.get("sensitive"):
+ self.delete_parameter(option)
+
+ def remove_sensitive_parameters_from_next(self, config):
+ """
+ Remove sensitive option values from queued follow-up parameters
+
+ A dataset's parameters can contain a `next` chain: parameters for
+ follow-up processors that run once this dataset finishes. Those
+ parameters may hold sensitive values (e.g. an API key a follow-up
+ processor needs). Once the follow-up datasets have been created they
+ hold their own copy and remove it themselves when they run, so those
+ values are no longer needed here and can be removed from this dataset's
+ stored `next` chain, where they would otherwise remain indefinitely.
+
+ Call this *after* the follow-up datasets have been created (at the end
+ of after_process), not before, or the follow-ups will not receive the
+ values.
+
+ :param config: Configuration reader, used to determine each follow-up
+ processor's options (which options are sensitive can depend on it).
+ """
+ if not isinstance(self.parameters.get("next"), list):
+ return
+
+ if self._remove_sensitive_from_next_steps(self.parameters["next"], config):
+ self.db.update("datasets", where={"key": self.key},
+ data={"parameters": json.dumps(self.parameters)})
+ self.data["parameters"] = json.dumps(self.parameters)
+
+ def _remove_sensitive_from_next_steps(self, steps, config):
+ """
+ Recursively remove sensitive option values from a list of `next` steps.
+
+ :param list steps: A list of `next` steps, each a dict with a `type`
+ and `parameters`.
+ :param config: Configuration reader.
+ :return bool: Whether anything was removed.
+ """
+ removed = False
+ for step in steps:
+ parameters = step.get("parameters")
+ if not isinstance(parameters, dict):
+ continue
+
+ processor = self.modules.processors.get(step.get("type")) if self.modules else None
+ if processor:
+ try:
+ # the immediate follow-ups run on this dataset, so this
+ # dataset is their parent context; pass self so options
+ # depedant on the parent resolve.
+ options = processor.get_options(self, config=config)
+ except Exception:
+ # never let a follow-up's get_options failure block cleanup
+ options = {}
+ for option, settings in options.items():
+ if settings.get("sensitive") and option in parameters:
+ del parameters[option]
+ removed = True
+
+ # a step can itself queue further steps
+ if isinstance(parameters.get("next"), list):
+ if self._remove_sensitive_from_next_steps(parameters["next"], config):
+ removed = True
+
+ return removed
+
def get_version_url(self, file):
"""
Get a versioned github URL for the version this dataset was processed with
diff --git a/common/lib/user_input.py b/common/lib/user_input.py
index 55aac477f..cc5505aa2 100644
--- a/common/lib/user_input.py
+++ b/common/lib/user_input.py
@@ -109,7 +109,13 @@ def parse_all(options, input, silently_correct=True):
# ignored
continue
- elif settings.get("type") == UserInput.OPTION_DATERANGE:
+ if settings.get("requires") and not UserInput.requirements_met(settings["requires"], parsed_input):
+ # the option's condition is not met, so the user never saw it:
+ # leave it out instead of storing a default for it. This holds
+ # whether or not the (hidden) form field was submitted.
+ continue
+
+ if settings.get("type") == UserInput.OPTION_DATERANGE:
# special case, since it combines two inputs
option_min = option + "-min"
option_max = option + "-max"
@@ -128,7 +134,7 @@ def parse_all(options, input, silently_correct=True):
if before and after and after > before:
if not silently_correct:
- raise QueryParametersException("End of date range must be after beginning of date range.")
+ raise QueryParametersException("The start of the date range must be before its end.")
else:
before = after
@@ -234,6 +240,43 @@ def parse_all(options, input, silently_correct=True):
return parsed_input
+ @staticmethod
+ def requirements_met(requires, other_input):
+ """
+ Check whether an option's "requires" condition is satisfied
+
+ `requires` may be:
+ - a single string: "field==value"
+ - an &&-joined string: "field1==v1 && field2==v2" (all must be true)
+ - a ||-joined string: "field1==v1 || field2==v2" (any must be true)
+ - a list/tuple of strings: every requirement must be true
+
+ :param requires: The option's "requires" setting
+ :param dict other_input: Values to check the condition against; a
+ condition that refers to a field missing from this dictionary is
+ not met
+ :return bool: True if the condition is satisfied
+ """
+ if isinstance(requires, (list, tuple)):
+ # a list always means: every requirement must be satisfied
+ req_list = list(requires)
+ combine_and = True
+ elif "||" in requires:
+ # pipe-separated alternatives: any one is sufficient
+ req_list = [r.strip() for r in requires.split("||")]
+ combine_and = False
+ elif "&&" in requires:
+ # ampersand-separated conditions: all must hold
+ req_list = [r.strip() for r in requires.split("&&")]
+ combine_and = True
+ else:
+ # single requirement string
+ req_list = [requires]
+ combine_and = True
+
+ results = [UserInput._requirement_met(r, other_input) for r in req_list]
+ return all(results) if combine_and else any(results)
+
@staticmethod
def _requirement_met(req, other_input):
"""
@@ -337,42 +380,9 @@ def parse_value(settings, choice, other_input=None, silently_correct=True):
:return: Validated and parsed input
"""
# short-circuit if there is a requirement for the field to be parsed
- # and the requirement isn't met.
- # 'requires' may be:
- # - a single string: "field==value"
- # - an &&-joined string: "field1==v1 && field2==v2" (all must be true)
- # - a ||-joined string: "field1==v1 || field2==v2" (any must be true)
- # - a list/tuple of strings: AND semantics (all must be true)
- if settings.get("requires"):
- reqs = settings["requires"]
-
- # normalise to a list of individual requirement strings plus a flag
- # indicating whether they combine with AND (True) or OR (False)
- if isinstance(reqs, (list, tuple)):
- # a list always uses AND semantics: every requirement must be satisfied
- req_list = list(reqs)
- combine_and = True
- elif "||" in reqs:
- # pipe-separated alternatives: any one is sufficient (OR)
- req_list = [r.strip() for r in reqs.split("||")]
- combine_and = False
- elif "&&" in reqs:
- # ampersand-separated conditions: all must hold (AND)
- req_list = [r.strip() for r in reqs.split("&&")]
- combine_and = True
- else:
- # single requirement string — original behaviour
- req_list = [reqs]
- combine_and = True
-
- results = [UserInput._requirement_met(r, other_input) for r in req_list]
-
- if combine_and and not all(results):
- # AND mode: every requirement must be satisfied
- raise RequirementsNotMetException()
- elif not combine_and and not any(results):
- # OR mode: at least one requirement must be satisfied
- raise RequirementsNotMetException()
+ # and the requirement isn't met
+ if settings.get("requires") and not UserInput.requirements_met(settings["requires"], other_input):
+ raise RequirementsNotMetException()
input_type = settings.get("type", "")
if input_type in UserInput.OPTIONS_COSMETIC:
diff --git a/datasources/audio_to_text/audio_to_text.py b/datasources/audio_to_text/audio_to_text.py
index ead409088..8089eb2c3 100644
--- a/datasources/audio_to_text/audio_to_text.py
+++ b/datasources/audio_to_text/audio_to_text.py
@@ -32,7 +32,11 @@ def validate_query(query, request, config):
# We need SearchMedia's validate_query to upload the media
media_query = SearchMedia.validate_query(query, request, config)
- # Here's the real trick: act like a preset and add another processor to the pipeline
- media_query["next"] = [{"type": "audio-to-text",
- "parameters": query.copy()}]
- return media_query
\ No newline at end of file
+ # Here's the real trick: act like a preset and add another processor to
+ # the pipeline. Pass it only its own options -- the media-upload options
+ # SearchMedia already consumed are not the audio-to-text processor's, so
+ # copying the whole query would leave stray keys in its parameters.
+ audio_options = AudioToText.get_options(config=config)
+ audio_parameters = {key: value for key, value in query.items() if key in audio_options}
+ media_query["next"] = [{"type": "audio-to-text", "parameters": audio_parameters}]
+ return media_query
diff --git a/datasources/bsky/search_bsky.py b/datasources/bsky/search_bsky.py
index 3454bc426..0e1ca7313 100644
--- a/datasources/bsky/search_bsky.py
+++ b/datasources/bsky/search_bsky.py
@@ -151,10 +151,8 @@ def validate_query(query, request, config):
# sanitize query
sanitized_query = [q.strip() for q in query.get("query").replace("\n", ",").split(",") if q.strip()]
- # the dates need to make sense as a range to search within
+ # a reversed date range is already rejected by parse_all()
min_date, max_date = query.get("daterange")
- if min_date and max_date and min_date > max_date:
- raise QueryParametersException("The start date must be before the end date.")
# Only check this if not already confirmed by the frontend
posts_per_second = 55 # gathered from simply checking start/end times of logs
@@ -171,15 +169,20 @@ def validate_query(query, request, config):
elif max_posts == 0:
raise QueryNeedsExplicitConfirmationException("No maximum number of posts set! This query may take a long time to complete. Do you want to continue?")
- return {
- "max_posts": query.get("max_posts"),
- "query": ",".join(sanitized_query),
- "username": query.get("username"),
- "password": query.get("password"),
- "session_id": session_id,
- "min_date": min_date,
- "max_date": max_date,
- }
+ # store the submitted parameters back, with the query normalised and
+ # the login session added; username/password were already normalised
+ # in place above
+ query["query"] = ",".join(sanitized_query)
+ query["session_id"] = session_id
+
+ # only store date bounds that were actually set
+ if min_date:
+ query["min_date"] = min_date
+ if max_date:
+ query["max_date"] = max_date
+ del query["daterange"]
+
+ return query
def get_items(self, query):
"""
diff --git a/datasources/douban/search_douban.py b/datasources/douban/search_douban.py
index 37ff7bbb0..5c7eb2c5e 100644
--- a/datasources/douban/search_douban.py
+++ b/datasources/douban/search_douban.py
@@ -101,8 +101,8 @@ def get_items(self, query):
"""
groups = query["groups"].split(",")
max_topics = min(convert_to_int(query["amount"], 100), 500)
- start = query["min_date"]
- end = query["max_date"]
+ start = query.get("min_date")
+ end = query.get("max_date")
strip = bool(query["strip"])
topics_processed = 0
posts_processed = 0
@@ -253,6 +253,7 @@ def get_douban_url(self, url, **kwargs):
return requests.get(url, **kwargs)
+ @staticmethod
def validate_query(query, request, config):
"""
Validate input for a dataset query on the Douban data source.
@@ -262,14 +263,8 @@ def validate_query(query, request, config):
:param ConfigManager|None config: Configuration reader (context-aware)
:return dict: Safe query parameters
"""
- filtered_query = {}
-
- # the dates need to make sense as a range to search within
+ # a reversed date range is already rejected by parse_all()
after, before = query.get("daterange")
- if before and after and before < after:
- raise QueryParametersException("Date range must start before it ends")
-
- filtered_query["min_date"], filtered_query["max_date"] = (after, before)
# normalize groups to just their IDs, even if a URL was provided, and
# limit to 25
@@ -279,12 +274,16 @@ def validate_query(query, request, config):
if not any(groups):
raise QueryParametersException("No valid groups were provided.")
- filtered_query["groups"] = ",".join(groups)
-
+ # store the submitted parameters back, with groups normalised, the
+ # topic count clamped, and the date range replaced by the bounds set
+ query["groups"] = ",".join(groups)
# max amount of topics is 200 because after that Douban starts throwing 429s
- filtered_query["amount"] = max(min(convert_to_int(query["amount"], 10), 200), 1)
+ query["amount"] = max(min(convert_to_int(query["amount"], 10), 200), 1)
- # strip HTML from posts?
- filtered_query["strip"] = bool(query.get("strip", False))
+ if after:
+ query["min_date"] = after
+ if before:
+ query["max_date"] = before
+ del query["daterange"]
- return filtered_query
+ return query
diff --git a/datasources/fourchan/search_4chan.py b/datasources/fourchan/search_4chan.py
index abc9e722c..ef8409d27 100644
--- a/datasources/fourchan/search_4chan.py
+++ b/datasources/fourchan/search_4chan.py
@@ -364,7 +364,7 @@ def get_options(cls, parent_dataset=None, config=None):
"United Nations": " United Nations",
"White Supremacist": " White Supremacist",
},
- "default": ""
+ "default": []
},
"divider": {
"type": UserInput.OPTION_DIVIDER
@@ -839,6 +839,7 @@ def get_thread_sizes(self, thread_ids, min_length):
return thread_sizes
+ @staticmethod
def validate_query(query, request, config):
"""
Validate input for a dataset query on the 4chan data source.
@@ -861,7 +862,12 @@ def validate_query(query, request, config):
and query.get("search_scope", "") != "match-ids":
raise QueryParametersException("Please provide a message or subject search query")
- query["min_date"], query["max_date"] = query["daterange"]
+ # only store date bounds that were actually set
+ after, before = query["daterange"]
+ if after:
+ query["min_date"] = after
+ if before:
+ query["max_date"] = before
del query["daterange"]
if query.get("search_scope") not in ("dense-threads",):
diff --git a/datasources/telegram/search_telegram.py b/datasources/telegram/search_telegram.py
index 5f80d11a5..f7120225d 100644
--- a/datasources/telegram/search_telegram.py
+++ b/datasources/telegram/search_telegram.py
@@ -126,7 +126,7 @@ def get_options(cls, parent_dataset=None, config=None):
f"ranges have **no** effect for [hashtag searches](https://telegram.org/blog/message-effects-and-more)"
f", which will always simply return all matching messages in reverse "
f"chronological order.")
-},
+ },
"query": {
"type": UserInput.OPTION_TEXT_LARGE,
"help": "Entities to scrape",
@@ -308,9 +308,9 @@ async def execute_queries(self):
return []
# ready our parameters
- parameters = self.dataset.get_parameters()
- queries = [query.strip() for query in parameters.get("query", "").split(",")]
- max_items = convert_to_int(parameters.get("items", 10), 10)
+ parameters = self.parameters
+ queries = [query.strip() for query in parameters["query"].split(",")]
+ max_items = convert_to_int(parameters["max_posts"], 10)
# If any query is a numeric ID, pre-fetch the dialog list so Telethon
# caches access_hash values for every channel/group the account is in.
@@ -380,7 +380,6 @@ async def gather_posts(self, queries, max_items, min_date, max_date):
crawl_max_depth = self.parameters.get("crawl-depth", 0)
crawl_msg_threshold = self.parameters.get("crawl-threshold", 10)
crawl_via_links = self.parameters.get("crawl-via-links", False)
-
self.dataset.log(f"Max crawl depth: {crawl_max_depth}")
self.dataset.log(f"Crawl threshold: {crawl_msg_threshold}")
@@ -1038,13 +1037,22 @@ def serialize_obj(input_obj):
@staticmethod
def validate_query(query, request, config):
"""
- Validate Telegram query
+ Validate Telegram query. Checks for required parameters and sanitizes the query.
+
+ Updated values:
+ - "query": Sanitized query string, with whitespace removed and newlines replaced by commas.
+ - "max_posts": Number of posts to query, limited by config settings.
+
+ Added keys:
+ - "min_date" (optional): Minimum date for the query, from daterange.
+ - "max_date" (optional): Maximum date for the query, from daterange.
+
+ Deleted keys:
+ - "daterange": Removed after extracting min_date and max_date.
- :param config:
:param dict query: Query parameters, from client-side.
:param request: Flask request
- :param User user: User object of user who has submitted the query
- :param ConfigManager config: Configuration reader (context-aware)
+ :param config:
:return dict: Safe query parameters
"""
# no query 4 u
@@ -1169,22 +1177,19 @@ def validate_query(query, request, config):
"help": "Security code",
"sensitive": True
}})
+
+ # Update query with sanitized items and date bounds
+ query["max_posts"] = num_items
+ query["query"] = ",".join(sanitized_items)
- # simple!
- return {
- "items": num_items,
- "query": ",".join(sanitized_items),
- "api_id": query.get("api_id"),
- "api_hash": query.get("api_hash"),
- "api_phone": query.get("api_phone"),
- "save-session": query.get("save-session"),
- "resolve-entities": query.get("resolve-entities"),
- "min_date": min_date,
- "max_date": max_date,
- "crawl-depth": query.get("crawl-depth"),
- "crawl-threshold": query.get("crawl-threshold"),
- "crawl-via-links": query.get("crawl-via-links")
- }
+ # only store date bounds that were actually set
+ if min_date:
+ query["min_date"] = min_date
+ if max_date:
+ query["max_date"] = max_date
+ del query["daterange"]
+
+ return query
async def iter_hashtag_messages(self, entity, offset_date=None):
"""
diff --git a/datasources/tumblr/search_tumblr.py b/datasources/tumblr/search_tumblr.py
index d87531f03..984110679 100644
--- a/datasources/tumblr/search_tumblr.py
+++ b/datasources/tumblr/search_tumblr.py
@@ -212,7 +212,7 @@ def get_items(self, query):
"""
# ready our parameters
- parameters = self.dataset.get_parameters()
+ parameters = self.parameters
queries = re.split(",|\n", parameters.get("query", ""))
get_notes = parameters.get("get_notes", False)
get_reblogs = parameters.get("get_reblogs", False)
@@ -931,6 +931,7 @@ def connect_to_tumblr(self):
return self.client
+ @staticmethod
def validate_query(query, request, config):
"""
Validate custom data input
@@ -964,10 +965,16 @@ def validate_query(query, request, config):
items = ", ".join([item.strip() for item in items if item])
# the dates need to make sense as a range to search within
- query["min_date"], query["max_date"] = query.get("daterange")
- if any(query.get("daterange")) and not all(query.get("daterange")):
+ after, before = query.get("daterange")
+ if any((after, before)) and not all((after, before)):
raise QueryParametersException("When providing a date range, set both an upper and lower limit.")
+ # only store date bounds that were actually set
+ if after:
+ query["min_date"] = after
+ if before:
+ query["max_date"] = before
+
del query["daterange"]
query["query"] = items
diff --git a/datasources/twitterv2/search_twitter.py b/datasources/twitterv2/search_twitter.py
index a813e3d50..94701af19 100644
--- a/datasources/twitterv2/search_twitter.py
+++ b/datasources/twitterv2/search_twitter.py
@@ -599,22 +599,27 @@ def validate_query(query, request, config):
else:
twitter_query = query.get("query")
- # the dates need to make sense as a range to search within
- # but, on Twitter, you can also specify before *or* after only
+ # on Twitter you can also specify before *or* after only; a reversed
+ # date range is already rejected by parse_all()
after, before = query.get("daterange")
- if before and after and before < after:
- raise QueryParametersException("Date range must start before it ends")
# if we made it this far, the query can be executed
params = {
"query": twitter_query,
- "api_bearer_token": query.get("api_bearer_token"),
"api_type": query.get("api_type", "all"),
"query_type": query.get("query_type", "query"),
- "min_date": after,
- "max_date": before
}
+ # the bearer token is only asked for when the server has no key of
+ # its own; only store it when it was given. Same for date bounds:
+ # only store those that were actually set
+ if "api_bearer_token" in query:
+ params["api_bearer_token"] = query["api_bearer_token"]
+ if after:
+ params["min_date"] = after
+ if before:
+ params["max_date"] = before
+
# never query more tweets than allowed
tweets_to_collect = convert_to_int(query.get("amount"), 10)
diff --git a/datasources/upload/import_csv.py b/datasources/upload/import_csv.py
index a76081475..c42c8568c 100644
--- a/datasources/upload/import_csv.py
+++ b/datasources/upload/import_csv.py
@@ -213,6 +213,7 @@ def process(self):
else:
self.dataset.finish(done)
+ @staticmethod
def validate_query(query, request, config):
"""
Validate custom data input
diff --git a/datasources/vk/search_vk.py b/datasources/vk/search_vk.py
index 7fcbebe66..b9263a66d 100644
--- a/datasources/vk/search_vk.py
+++ b/datasources/vk/search_vk.py
@@ -340,27 +340,22 @@ def validate_query(query, request, config):
if not query.get("query", None):
raise QueryParametersException("Please provide a query.")
- # the dates need to make sense as a range to search within
- # but, on VK, you can also specify before *or* after only
+ # on VK you can also specify before *or* after only; a reversed date
+ # range is already rejected by parse_all()
after, before = query.get("daterange")
- if before and after and before < after:
- raise QueryParametersException("Date range must start before it ends")
# TODO: test username and password?
- # if we made it this far, the query can be executed
- params = {
- "query": query.get("query"),
- "query_type": query.get("query_type"),
- "amount": query.get("amount"),
- "include_comments": query.get("include_comments"),
- "min_date": after,
- "max_date": before,
- "username": query.get("username"),
- "password": query.get("password"),
- }
+ # if we made it this far, the query can be executed; store the
+ # submitted parameters back, replacing the date range with the bounds
+ # that were actually set
+ if after:
+ query["min_date"] = after
+ if before:
+ query["max_date"] = before
+ del query["daterange"]
- return params
+ return query
@staticmethod
def map_item(item):
diff --git a/processors/filtering/column_filter.py b/processors/filtering/column_filter.py
index c681f3ca7..c0ac7aff1 100644
--- a/processors/filtering/column_filter.py
+++ b/processors/filtering/column_filter.py
@@ -9,6 +9,7 @@
from processors.filtering.base_filter import BaseFilter
from common.lib.helpers import UserInput, convert_to_int
from common.lib.compatibility import Compatibility
+from common.lib.exceptions import QueryParametersException
__author__ = "Stijn Peeters"
__credits__ = ["Stijn Peeters", "Dale Wahl"]
@@ -60,7 +61,7 @@ def get_options(cls, parent_dataset=None, config=None) -> dict:
"top-top": "is in the top n results for this attribute",
"top-bottom": "is in the bottom n results for this attribute"
},
- "default": "exact"
+ "default": "value-equals"
},
"strict-top": {
"type": UserInput.OPTION_TOGGLE,
@@ -129,6 +130,40 @@ def get_options(cls, parent_dataset=None, config=None) -> dict:
return options
+ @staticmethod
+ def validate_query(query, request, config):
+ """
+ Check that the value to compare with fits the chosen match type
+
+ Numerical and date comparisons can only work if the value to compare
+ with is a number or a date. Checking this here means the user gets
+ immediate feedback in the form, instead of a dataset that fails while
+ running. The value fields are only part of the input when the chosen
+ match type actually uses them, so they can be checked directly.
+
+ :param dict query: Parsed user input
+ :param request: Flask request the input arrived in
+ :param config: Configuration reader
+ :return dict: The parameters to store
+ """
+ if query.get("match-style") in ("value-less-than", "value-greater-than"):
+ try:
+ [float(value) for value in query.get("match-value", "").split(",")]
+ except ValueError:
+ raise QueryParametersException("Comparing as a number requires the value to compare with to be "
+ f"a number; '{query.get('match-value')}' is not.")
+
+ if query.get("match-style") in ("date-after", "date-before"):
+ match_date = query.get("match-date", "").strip()
+ if not re.match(r"[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}", match_date):
+ try:
+ int(match_date)
+ except ValueError:
+ raise QueryParametersException("Comparing by date requires the value to compare with to be a "
+ "date, either as YYYY-MM-DD hh:mm:ss or as a unix timestamp.")
+
+ return query
+
def filter_items(self):
"""
Create a generator to iterate through items that can be passed to create either a csv or ndjson. Use
diff --git a/processors/machine_learning/google_vision_api.py b/processors/machine_learning/google_vision_api.py
index 73cae0d2a..1f6f6eeea 100644
--- a/processors/machine_learning/google_vision_api.py
+++ b/processors/machine_learning/google_vision_api.py
@@ -63,6 +63,7 @@ def get_options(cls, parent_dataset=None, config=None) -> dict:
},
"api_key": {
"type": UserInput.OPTION_TEXT,
+ "sensitive": True,
"help": "API Key",
"tooltip": "The API Key for the Google API account you want to query with. You can generate and find this"
"key on console.cloud.google.com. You also need to enable billing and Vision API."
@@ -98,8 +99,9 @@ def process(self):
with one column with image hashes, one with the first file name used
for the image, and one with the amount of times the image was used
"""
+ # marked sensitive in get_options, so the stored value is already
+ # removed by the time this runs; the in-memory value is still available
api_key = self.parameters.get("api_key")
- self.dataset.delete_parameter("api_key") # sensitive, delete after use
features = self.parameters.get("features")
features = [{"type": feature} for feature in features]
diff --git a/processors/metrics/rank_attribute.py b/processors/metrics/rank_attribute.py
index a47c350dd..bc11298cf 100644
--- a/processors/metrics/rank_attribute.py
+++ b/processors/metrics/rank_attribute.py
@@ -113,7 +113,7 @@ def get_options(cls, parent_dataset=None, config=None):
},
"negate-filter": {
"type": UserInput.OPTION_TOGGLE,
- "default": "",
+ "default": False,
"help": "Negate filter",
"tooltip": "Only match items that do *not* match the filter configured above"
},
diff --git a/processors/presets/annotate-images.py b/processors/presets/annotate-images.py
index 829c5d8ac..57acbc55e 100644
--- a/processors/presets/annotate-images.py
+++ b/processors/presets/annotate-images.py
@@ -47,6 +47,7 @@ def get_options(cls, parent_dataset=None, config=None) -> dict:
},
"api_key": {
"type": UserInput.OPTION_TEXT,
+ "sensitive": True,
"help": "API Key",
"tooltip": "The API Key for your Google API account. You can generate and find this "
"key on the API dashboard."
@@ -81,26 +82,24 @@ def get_processor_pipeline(self):
is converted to a CSV file for easy processing.
"""
amount = convert_to_int(self.parameters.get("amount", 10), 10)
+ # api_key is marked sensitive, so its stored value is already removed
+ # by the time this runs; the in-memory value is still available to pass
+ # to the pipeline, and is scrubbed from the follow-up datasets in turn
api_key = self.parameters.get("api_key", "")
features = self.parameters.get("features", "")
- self.dataset.delete_parameter("api_key") # sensitive, delete as soon as possible
-
pipeline = [
# first, extract top images
{
"type": "top-images",
- "parameters": {
- "overwrite": False
- }
+ "parameters": {}
},
# then, download the images we want to annotate
{
"type": "image-downloader",
"parameters": {
"amount": amount,
- "columns": "item",
- "overwrite": False
+ "columns": "item"
}
},
# then, annotate the downloaded images with the Google Vision API
@@ -114,7 +113,7 @@ def get_processor_pipeline(self):
},
# finally, create a simplified CSV file from the download NDJSON (which can also be retrieved later)
{
- "type": "convert-vision-to-csv",
+ "type": "convert-google-vision-to-csv",
"parameters": {}
}
]
diff --git a/processors/presets/neologisms.py b/processors/presets/neologisms.py
index 0b1e0179d..a0aa7d849 100644
--- a/processors/presets/neologisms.py
+++ b/processors/presets/neologisms.py
@@ -69,7 +69,6 @@ def get_processor_pipeline(self):
"type": "tokenise-posts",
"parameters": {
"stem": False,
- "strip_symbols": True,
"lemmatise": False,
"docs_per": timeframe,
"columns": columns,
@@ -85,7 +84,6 @@ def get_processor_pipeline(self):
{
"type": "vector-ranker",
"parameters": {
- "amount": True,
"top": 15,
}
}
diff --git a/processors/presets/similar-words.py b/processors/presets/similar-words.py
index 5c172776e..b63387f14 100644
--- a/processors/presets/similar-words.py
+++ b/processors/presets/similar-words.py
@@ -71,7 +71,7 @@ def get_processor_pipeline(self):
"stem": False,
"lemmatise": False,
"columns": "body",
- "timeframe": timeframe,
+ "docs_per": timeframe,
"grouping-per": "sentence",
"language": language
}
diff --git a/processors/text-analysis/collocations.py b/processors/text-analysis/collocations.py
index 147608334..e777dc499 100644
--- a/processors/text-analysis/collocations.py
+++ b/processors/text-analysis/collocations.py
@@ -40,7 +40,7 @@ def get_options(cls, parent_dataset=None, config=None) -> dict:
return {
"n_size": {
"type": UserInput.OPTION_CHOICE,
- "default": 2,
+ "default": "2",
"options": {
"2": "2 (bigrams)",
"3": "3 (trigrams)"},
diff --git a/processors/text-analysis/similar_words.py b/processors/text-analysis/similar_words.py
index 7db6573c6..ad8f7db8a 100644
--- a/processors/text-analysis/similar_words.py
+++ b/processors/text-analysis/similar_words.py
@@ -63,7 +63,7 @@ def get_options(cls, parent_dataset=None, config=None) -> dict:
},
"crawl_depth": {
"type": UserInput.OPTION_CHOICE,
- "default": 1,
+ "default": "1",
"options": {"1": 1, "2": 2, "3": 3},
"help": "The crawl depth. 1 only gets the neighbours of the input word(s), 2 also their neighbours, etc."
}
diff --git a/processors/text-analysis/tf_idf.py b/processors/text-analysis/tf_idf.py
index 4e7378c4f..46d5c90a6 100644
--- a/processors/text-analysis/tf_idf.py
+++ b/processors/text-analysis/tf_idf.py
@@ -99,7 +99,7 @@ def get_options(cls, parent_dataset=None, config=None) -> dict:
},
"n_size": {
"type": UserInput.OPTION_CHOICE,
- "default": "",
+ "default": "1",
"options": {"1":"unigrams (1)", "2": "bigrams (2)", "3": "trigrams", "1-2": "uni- and bigrams (1-2)", "1-3": "uni-, bi-, and trigrams (1-3)"},
"help": "[scikit-learn] Amount of words to return",
"tooltip": "Selecting a range can be useful to e.g. extract multi-word nouns like names.",
diff --git a/processors/visualisation/download_tiktok_video.py b/processors/visualisation/download_tiktok_video.py
index 97a809602..d017ebc22 100644
--- a/processors/visualisation/download_tiktok_video.py
+++ b/processors/visualisation/download_tiktok_video.py
@@ -81,23 +81,21 @@ def get_processor_pipeline(self):
"""
This queues the video-downloader with set options
"""
- # Check if an upload
- if self.source_dataset.type == "upload-search":
- # Variable column name
- column = self.parameters.get("column")
- else:
- column = "id"
-
amount = self.parameters.get("amount")
+ metadata_parameters = {
+ "amount": amount,
+ "_amount_leeway": 10 # extra metadata in case videos fail
+ }
+ # only an uploaded dataset needs to be told which column holds the post
+ # IDs; for TikTok datasources the metadata processor uses a fixed column
+ if self.source_dataset.type == "upload-search":
+ metadata_parameters["id_column"] = self.parameters.get("column")
+
pipeline = [
{
"type": "tiktok-video-downloader-metadata",
- "parameters": {
- "column": column,
- "amount": amount,
- "_amount_leeway": 10 # extra metadata in case videos fail
- }
+ "parameters": metadata_parameters
},
{
"type": "video-downloader",
diff --git a/processors/visualisation/download_videos.py b/processors/visualisation/download_videos.py
index 552eb4be8..6c32dee8e 100644
--- a/processors/visualisation/download_videos.py
+++ b/processors/visualisation/download_videos.py
@@ -96,6 +96,11 @@ class VideoDownloaderPlus(BasicProcessor):
extension = "zip" # extension of result file, used internally and in UI
media_type = "video" # media type of the processor
+ # `also_indirect` is offered as an option only when the admin allows
+ # indirect downloads, but it is always read, and the TikTok downloader
+ # preset deliberately passes it, so declare that it is accepted
+ accepted_parameters = ("also_indirect",)
+
# Shared list -- other download_* processors reuse this as VideoDownloaderPlus.followups
# (and preferred_followups below reuses it), so it stays a named attribute.
followups = ["audio-extractor", "metadata-viewer", "video-scene-detector", "preset-scene-timelines", "video-stack", "preset-video-hashes", "video-hasher-1", "video-frames"]
diff --git a/processors/visualisation/histwords.py b/processors/visualisation/histwords.py
index 1fbe4302c..0bcb28311 100644
--- a/processors/visualisation/histwords.py
+++ b/processors/visualisation/histwords.py
@@ -78,7 +78,7 @@ def get_options(cls, parent_dataset=None, config=None) -> dict:
"PCA": "PCA",
"TruncatedSVD": "Truncated SVD (randomised, 5 iterations)"
},
- "default": "tsne"
+ "default": "t-SNE"
},
"num-words": {
"type": UserInput.OPTION_TEXT,
diff --git a/processors/visualisation/image_wall.py b/processors/visualisation/image_wall.py
index ebbb25f2b..c0b2f7bb1 100644
--- a/processors/visualisation/image_wall.py
+++ b/processors/visualisation/image_wall.py
@@ -72,6 +72,8 @@ def get_options(cls, parent_dataset=None, config=None):
"kmeans-dominant": "Dominant K-means (precise, slow)",
"average-hsv": "Average colour (HSV; imprecise, fastest)",
}
+ # add default (inherits options from video wall)
+ options["sort-mode"]["default"] = ""
else:
# add some caveats for running this directly on a video dataset
options["sort-mode"]["tooltip"] = ("To sort by e.g. average colour, first extract frames as images and "
diff --git a/processors/visualisation/rankflow.py b/processors/visualisation/rankflow.py
index 99397321d..09f007dff 100644
--- a/processors/visualisation/rankflow.py
+++ b/processors/visualisation/rankflow.py
@@ -107,7 +107,7 @@ def get_options(cls, parent_dataset=None, config=None) -> dict:
"weight": "Value (items with a higher value are bigger)",
"none": "None (same size for all elements)",
},
- "default": "change",
+ "default": "weight",
"help": "Size according to",
},
"only_adjacent_flows": {
diff --git a/processors/visualisation/video_frames.py b/processors/visualisation/video_frames.py
index 9a24621fa..8f0cc7fc1 100644
--- a/processors/visualisation/video_frames.py
+++ b/processors/visualisation/video_frames.py
@@ -58,7 +58,7 @@ def get_options(cls, parent_dataset=None, config=None) -> dict:
},
"frame_size": {
"type": UserInput.OPTION_CHOICE,
- "default": "medium",
+ "default": "432x432",
"options": {
"no_modify": "Do not modify",
"144x144": "Tiny (144x144)",
diff --git a/processors/visualisation/video_hasher.py b/processors/visualisation/video_hasher.py
index 8271e7b69..3c127df43 100644
--- a/processors/visualisation/video_hasher.py
+++ b/processors/visualisation/video_hasher.py
@@ -159,6 +159,11 @@ def get_options(cls, parent_dataset=None, config=None):
"default": 1,
"min": 0,
"max": 5,
+ },
+ "save_annotations": {
+ "type": UserInput.OPTION_TOGGLE,
+ "help": "Add hashes to top dataset",
+ "default": False
}
}
diff --git a/processors/visualisation/video_scene_frames.py b/processors/visualisation/video_scene_frames.py
index 65e136fee..b6bf50ebc 100644
--- a/processors/visualisation/video_scene_frames.py
+++ b/processors/visualisation/video_scene_frames.py
@@ -51,7 +51,7 @@ def get_options(cls, parent_dataset=None, config=None) -> dict:
return {
"frame_size": {
"type": UserInput.OPTION_CHOICE,
- "default": "medium",
+ "default": "432x432",
"options": {
"no_modify": "Do not modify",
"144x144": "Tiny (144x144)",
diff --git a/processors/visualisation/video_scene_identifier.py b/processors/visualisation/video_scene_identifier.py
index 7fa6c03b8..0fc1b6a8b 100644
--- a/processors/visualisation/video_scene_identifier.py
+++ b/processors/visualisation/video_scene_identifier.py
@@ -203,7 +203,9 @@ def get_options(cls, parent_dataset=None, config=None) -> dict:
ffmpeg_path = config.get("video-downloader.ffmpeg_path")
if not ffmpeg_path or not os.path.exists(shutil.which(ffmpeg_path)):
del options["detector_type"]["options"]["ffmpeg_select"]
- options["detector_type"]["default"] = list(options["detector_type"]["options"].values())[0]
+ # fall back to the first remaining choice; the keys are the stored
+ # values, the dictionary values are display labels
+ options["detector_type"]["default"] = list(options["detector_type"]["options"])[0]
return options
diff --git a/tests/test_modules.py b/tests/test_modules.py
index 027a9c4ef..ebb09dd36 100644
--- a/tests/test_modules.py
+++ b/tests/test_modules.py
@@ -406,6 +406,313 @@ def test_datasources(logger, fourcat_modules, mock_job, mock_job_queue, mock_dat
logger.info("All datasources passed successfully.")
+@pytest.mark.dependency(depends=["test_module_collector"])
+def test_validate_query_declarations(logger, fourcat_modules):
+ """
+ validate_query is always called on the class, never on an instance, so
+ any worker that defines its own must make it a @staticmethod. A plain
+ method happens to work when called on the class, but binds the query
+ dictionary to `self` as soon as it is called on an instance, so enforce
+ the decorator here.
+ """
+ import inspect
+ from backend.lib.processor import BasicProcessor
+
+ offenders = []
+ for name, worker in fourcat_modules.processors.items():
+ if worker.validate_query is BasicProcessor.validate_query:
+ # inherits the store-as-is default; nothing declared to check
+ continue
+
+ declaration = inspect.getattr_static(worker, "validate_query")
+ if not isinstance(declaration, staticmethod):
+ offenders.append(name)
+
+ assert not offenders, (
+ "These workers define validate_query without @staticmethod; it is called on the "
+ f"class, so it must be a static method: {sorted(set(offenders))}"
+ )
+
+
+@pytest.mark.dependency(depends=["test_module_collector"])
+def test_option_declarations(logger, fourcat_modules, mock_dataset, mock_basic_config):
+ """
+ Option defaults are stored and filled in exactly as declared - they skip
+ the type parsing that user input gets - so a wrongly typed default shows
+ up as a confusing runtime value instead of an error. Check the
+ declarations themselves: toggles need a True/False default, a choice
+ default must be one of the choices, text options with a min/max are
+ treated as numbers so their default must be a number, and multi-choice
+ defaults must be lists.
+ """
+ from common.lib.helpers import UserInput
+
+ problems = []
+ for name, worker in fourcat_modules.processors.items():
+ try:
+ options = worker.get_options(parent_dataset=mock_dataset, config=mock_basic_config)
+ except Exception:
+ # get_options failures are already reported by test_processors
+ continue
+
+ for option, settings in (options or {}).items():
+ if not isinstance(settings, dict):
+ problems.append((name, option, "option settings are not a dictionary"))
+ continue
+
+ option_type = settings.get("type")
+ default = settings.get("default")
+
+ if option_type in (UserInput.OPTION_TOGGLE, UserInput.OPTION_ANNOTATION):
+ if "default" in settings and not isinstance(default, bool):
+ problems.append((name, option, f"toggle default should be True or False, not {default!r}"))
+
+ elif option_type == UserInput.OPTION_CHOICE:
+ choices = settings.get("options", {})
+ if isinstance(choices, dict) and choices and all(isinstance(choice, dict) for choice in choices.values()):
+ # choices grouped into categories: the real values are the inner keys
+ choices = [value for group in choices.values() for value in group]
+ if "default" in settings and choices and default not in choices:
+ problems.append((name, option, f"default {default!r} is not one of the choices"))
+
+ elif option_type in (UserInput.OPTION_TEXT, UserInput.OPTION_TEXT_LARGE, UserInput.OPTION_HUE):
+ if ("min" in settings or "max" in settings) and "default" in settings \
+ and not isinstance(default, (int, float)):
+ problems.append((name, option, f"option has a min/max so its value is treated as a number, but the default is {default!r}"))
+
+ elif option_type in (UserInput.OPTION_MULTI, UserInput.OPTION_MULTI_SELECT):
+ if "default" in settings and default is not None and not isinstance(default, (list, tuple)):
+ problems.append((name, option, f"default for a multiple-choice option should be a list, not {default!r}"))
+
+ if problems:
+ report = "\n".join(f"{name} / {option}: {problem}" for name, option, problem in sorted(problems))
+ pytest.fail(f"{len(problems)} option declaration(s) have defaults that do not match their type:\n{report}")
+ else:
+ logger.info("All option declarations look consistent.")
+
+
+def test_parse_all_gated_options():
+ """
+ Options with a "requires" condition are only part of the parsed input when
+ their condition is met - whether or not the (hidden) form field was
+ submitted. This keeps the stored parameters honest: a missing key means
+ the user never saw the option, a present key means they chose a value or
+ its default applies. At run time, self.parameters fills every declared
+ option regardless (so plain reads are always safe); the stored honesty is
+ exposed to workers through BasicProcessor.option_given().
+ """
+ from common.lib.user_input import UserInput
+
+ options = {
+ "gate": {"type": UserInput.OPTION_TOGGLE, "default": False},
+ "gated": {"type": UserInput.OPTION_TEXT, "default": ",", "requires": "gate==true"},
+ "gated_toggle": {"type": UserInput.OPTION_TOGGLE, "default": True, "requires": "gate==true"},
+ "plain": {"type": UserInput.OPTION_TEXT, "default": "x"},
+ }
+
+ # gate off, gated fields not submitted: gated options are absent, not defaulted
+ parsed = UserInput.parse_all(options, {"option-plain": "y"})
+ assert parsed == {"gate": False, "plain": "y"}
+
+ # gate off, gated field submitted anyway (e.g. hidden field still posts):
+ # still absent
+ parsed = UserInput.parse_all(options, {"option-gated": ";"})
+ assert "gated" not in parsed and "gated_toggle" not in parsed
+
+ # gate on: gated options are parsed (submitted value) or defaulted (absent)
+ parsed = UserInput.parse_all(options, {"option-gate": "on", "option-gated": ";"})
+ assert parsed["gated"] == ";" and parsed["gated_toggle"] is False and parsed["gate"] is True
+
+
+def test_get_validated_query_flags_stored_none():
+ """
+ get_validated_query is the doorway between validate_query and the stored
+ dataset parameters. It warns when a validate_query stores None for an
+ option that was not part of the submission (the query.get() mistake in a
+ rebuilt dictionary), and refuses a validate_query that returns nothing.
+ """
+ from backend.lib.processor import BasicProcessor
+ from common.lib.exceptions import ProcessorException
+
+ class Rebuilder(BasicProcessor):
+ type = "rebuilder-test"
+
+ def process(self):
+ pass
+
+ @staticmethod
+ def validate_query(query, request, config):
+ return {"kept": query.get("kept"), "junk": query.get("junk")}
+
+ log = MagicMock()
+ result = Rebuilder.get_validated_query({"kept": "value"}, None, None, log=log)
+ assert result == {"kept": "value", "junk": None}
+ log.warning.assert_called_once()
+ assert "junk" in log.warning.call_args[0][0]
+
+ log = MagicMock()
+ Rebuilder.get_validated_query({"kept": "value", "junk": "given"}, None, None, log=log)
+ log.warning.assert_not_called()
+
+ # framework-injected transient keys are stripped from the stored result,
+ # even when validate_query returns them (e.g. a modify-in-place worker)
+ class PassThrough(BasicProcessor):
+ type = "passthrough-test"
+
+ def process(self):
+ pass
+
+ @staticmethod
+ def validate_query(query, request, config):
+ return query
+
+ result = PassThrough.get_validated_query({"kept": "value", "frontend-confirm": True}, None, None)
+ assert result == {"kept": "value"}
+
+ class Forgetful(BasicProcessor):
+ type = "forgetful-test"
+
+ def process(self):
+ pass
+
+ @staticmethod
+ def validate_query(query, request, config):
+ pass
+
+ with pytest.raises(ProcessorException):
+ Forgetful.get_validated_query({}, None, None)
+
+
+def test_warn_unexpected_parameters():
+ """
+ Programmatically-queued datasets (preset pipelines, `next` chains) are not
+ validated like user queries, so a parameter meant for another processor or
+ misspelled is silently ignored. warn_unexpected_parameters logs a warning
+ for any key the target processor doesn't declare, while treating framework
+ keys and `_`-prefixed internal plumbing as expected. It never raises.
+ """
+ from backend.lib.processor import BasicProcessor
+
+ def config_with_dev_mode(enabled):
+ cfg = MagicMock()
+ cfg.get = MagicMock(side_effect=lambda key, default=None, **kw: enabled if key == "dev.mode" else default)
+ return cfg
+
+ dev_on = config_with_dev_mode(True)
+
+ processor = MagicMock()
+ processor.type = "tokenise-posts"
+ processor.get_options.return_value = {"docs_per": {}, "columns": {}}
+
+ # a key the processor doesn't declare is flagged; declared, framework, and
+ # underscore-prefixed keys are not
+ log = MagicMock()
+ BasicProcessor.warn_unexpected_parameters(
+ processor,
+ {"docs_per": "month", "columns": "body", "timeframe": "x",
+ "next": [], "attach_to": "k", "_internal": 1},
+ parent_dataset=None, config=dev_on, log=log)
+ log.warning.assert_called_once()
+ assert "timeframe" in log.warning.call_args[0][0]
+
+ # nothing unexpected -> no warning
+ log = MagicMock()
+ BasicProcessor.warn_unexpected_parameters(
+ processor, {"docs_per": "month"}, parent_dataset=None, config=dev_on, log=log)
+ log.warning.assert_not_called()
+
+ # a key the target reads but does not declare, listed in accepted_parameters,
+ # is not flagged (e.g. a config-gated option a preset deliberately passes)
+ processor.accepted_parameters = ("also_indirect",)
+ log = MagicMock()
+ BasicProcessor.warn_unexpected_parameters(
+ processor, {"docs_per": "month", "also_indirect": "all"},
+ parent_dataset=None, config=dev_on, log=log)
+ log.warning.assert_not_called()
+
+ # dev mode off -> no warning, even for an unexpected key
+ log = MagicMock()
+ BasicProcessor.warn_unexpected_parameters(
+ processor, {"timeframe": "x"}, parent_dataset=None, config=config_with_dev_mode(False), log=log)
+ log.warning.assert_not_called()
+
+ # a processor whose get_options raises must not raise here
+ broken = MagicMock()
+ broken.type = "broken"
+ broken.get_options.side_effect = RuntimeError("boom")
+ log = MagicMock()
+ BasicProcessor.warn_unexpected_parameters(
+ broken, {"anything": 1}, parent_dataset=None, config=dev_on, log=log)
+ log.warning.assert_not_called()
+
+
+def test_remove_sensitive_parameters_from_next(mock_dataset):
+ """
+ Sensitive values queued for follow-up processors in a dataset's `next`
+ chain (e.g. an API key a follow-up needs) must be removed from the stored
+ record once the follow-up datasets exist and hold their own copy. Nested
+ chains are cleaned recursively; non-sensitive values are left in place.
+ """
+ fake_processor = MagicMock()
+ fake_processor.get_options.return_value = {
+ "api_key": {"sensitive": True},
+ "amount": {},
+ }
+ mock_dataset.modules = MagicMock()
+ mock_dataset.modules.processors = {"fake-proc": fake_processor}
+
+ mock_dataset.parameters = {
+ "next": [{
+ "type": "fake-proc",
+ "parameters": {
+ "api_key": "SECRET",
+ "amount": 5,
+ "next": [{
+ "type": "fake-proc",
+ "parameters": {"api_key": "ALSO_SECRET", "amount": 9},
+ }],
+ },
+ }],
+ }
+
+ mock_dataset.remove_sensitive_parameters_from_next(config=None)
+
+ top = mock_dataset.parameters["next"][0]["parameters"]
+ assert "api_key" not in top, "top-level sensitive key should be removed"
+ assert top["amount"] == 5, "non-sensitive key should be kept"
+ nested = top["next"][0]["parameters"]
+ assert "api_key" not in nested, "nested sensitive key should be removed"
+ assert nested["amount"] == 9, "nested non-sensitive key should be kept"
+
+
+def test_remove_sensitive_parameters(mock_dataset):
+ """
+ Sensitive option values (e.g. API keys) must be removed from a dataset's
+ own stored parameters as soon as it runs. The options are resolved from
+ both the processor that produced the dataset and the one matching its
+ current type (these can differ once a filter has adopted a new type), so a
+ sensitive option declared by either is scrubbed while non-sensitive values
+ are left in place.
+ """
+ producer = MagicMock()
+ producer.get_options.return_value = {"api_key": {"sensitive": True}, "amount": {}}
+ own = MagicMock()
+ own.get_options.return_value = {"session_token": {"sensitive": True}}
+
+ mock_dataset.get_producer_processor = MagicMock(return_value=producer)
+ mock_dataset.get_own_processor = MagicMock(return_value=own)
+
+ deleted = []
+ mock_dataset.delete_parameter = MagicMock(side_effect=deleted.append)
+
+ mock_dataset.remove_sensitive_parameters(config=None)
+
+ # a sensitive option declared by either processor is removed; the
+ # non-sensitive one is left untouched
+ assert set(deleted) == {"api_key", "session_token"}
+ assert "amount" not in deleted
+
+
def test_dataset_finish_raises_on_double_finish(mock_dataset):
"""
Regression guard for common/lib/dataset.py:986.
diff --git a/webtool/views/api_tool.py b/webtool/views/api_tool.py
index 903f5fb75..47fdef7ed 100644
--- a/webtool/views/api_tool.py
+++ b/webtool/views/api_tool.py
@@ -23,6 +23,7 @@
from common.lib.helpers import UserInput, call_api, get_software_commit, get_software_version, get_git_branch
from common.lib.user import User
from backend.lib.worker import BasicWorker
+from backend.lib.processor import BasicProcessor
from common.lib.item_mapping import MissingMappedField
component = Blueprint("toolapi", __name__)
@@ -447,35 +448,38 @@ def queue_dataset():
# source specific
has_confirm = bool(request.form.get("frontend-confirm", False))
- if hasattr(search_worker, "validate_query"):
- # queries are always validated, also if they have been validated before,
- # just in case
- try:
- # first sanitise values
- sanitised_query = UserInput.parse_all(search_worker.get_options(None, g.config), request.form, silently_correct=False)
+ if search_worker.validate_query is BasicProcessor.validate_query:
+ # a datasource is only queryable from here if its search worker
+ # defines its own validate_query; one that does not (e.g. the
+ # import-only/Zeeschuimer datasources) gets its data some other way
+ g.log.warning("Datasource '%s' has no validate_query method, so it cannot be queued via the web interface" % datasource_id)
+ return error(404, message="Datasource '%s' does not support queueing datasets via the web interface" % datasource_id)
- # then validate for this particular datasource
- sanitised_query = {"frontend-confirm": has_confirm, **sanitised_query}
- sanitised_query = search_worker.validate_query(sanitised_query, request, g.config)
+ # queries are always validated, also if they have been validated before,
+ # just in case
+ try:
+ # first sanitise values
+ sanitised_query = UserInput.parse_all(search_worker.get_options(None, g.config), request.form, silently_correct=False)
- except QueryNeedsFurtherInputException as e:
- # ask the user for more input by returning a HTML snippet
- # containing form fields to be added to the form before it is
- # re-submitted
- form = render_template("components/create-dataset-option.html", options=e.config)
- return jsonify({"status": "extra-form", "html": form})
+ # then validate for this particular datasource
+ sanitised_query = {"frontend-confirm": has_confirm, **sanitised_query}
+ sanitised_query = search_worker.get_validated_query(sanitised_query, request, g.config, log=g.log)
- except QueryParametersException as e:
- # parameters need amending
- return jsonify({"status": "error", "message": "Cannot create a dataset with these parameters. %s" % e})
+ except QueryNeedsFurtherInputException as e:
+ # ask the user for more input by returning a HTML snippet
+ # containing form fields to be added to the form before it is
+ # re-submitted
+ form = render_template("components/create-dataset-option.html", options=e.config)
+ return jsonify({"status": "extra-form", "html": form})
- except QueryNeedsExplicitConfirmationException as e:
- # parameters are OK, but we need to be sure the user wants this
- # (because it will e.g. take a long time)
- return jsonify({"status": "confirm", "message": str(e)})
+ except QueryParametersException as e:
+ # parameters need amending
+ return jsonify({"status": "error", "message": "Cannot create a dataset with these parameters. %s" % e})
- else:
- raise NotImplementedError("Data sources MUST sanitise input values with validate_query")
+ except QueryNeedsExplicitConfirmationException as e:
+ # parameters are OK, but we need to be sure the user wants this
+ # (because it will e.g. take a long time)
+ return jsonify({"status": "confirm", "message": str(e)})
# parameters OK, front-end can submit for real
# why not just continue? because the initial validation is done with
@@ -1211,11 +1215,11 @@ def queue_processor(key=None, processor=None):
sanitised_query["frontend-confirm"] = bool(request.form.get("frontend-confirm", False))
- if hasattr(processor_worker, "validate_query"):
- # validate_query is optional for processors
- sanitised_query = processor_worker.validate_query(
- sanitised_query, request, g.config
- )
+ # by default this stores the input as-is; processors with something
+ # to check or confirm define their own validate_query
+ sanitised_query = processor_worker.get_validated_query(
+ sanitised_query, request, g.config, log=g.log
+ )
except QueryParametersException as e:
# parameters need amending