diff --git a/automated_ingestion/scripts/report_duplicate_observations.py b/automated_ingestion/scripts/report_duplicate_observations.py new file mode 100644 index 000000000..96ffbba94 --- /dev/null +++ b/automated_ingestion/scripts/report_duplicate_observations.py @@ -0,0 +1,153 @@ +# =============================================================================== +# Copyright 2026 ross +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +Report duplicate transducer observations before the unique constraint migration. + +Does what ``sql/find_duplicate_observations.sql`` does, without needing ``psql`` +or a password: it connects through the Cloud SQL connector with IAM +authentication, so the credential is your own gcloud login and nothing is +stored. + + gcloud auth application-default login + uv run --group ingestion python -m \\ + automated_ingestion.scripts.report_duplicate_observations \\ + --instance waterdatainitiative-271000:us-west4:dataservices \\ + --database ocotillo-staging + +You need a database login. Being a project owner is not enough -- Cloud SQL +requires the principal to exist as a database user: + + gcloud sql users create YOUR_EMAIL --instance=dataservices \\ + --type=cloud_iam_user --project=waterdatainitiative-271000 + +Read-only. It counts and reports; deciding what to do about duplicates is a +judgement about the data, not something a script should make. +""" + +import argparse +import sys + +DUPLICATE_GROUPS = """ +SELECT deployment_id, parameter_id, observation_datetime, + count(*) AS copies, count(DISTINCT value) AS distinct_values +FROM transducer_observation +GROUP BY deployment_id, parameter_id, observation_datetime +HAVING count(*) > 1 +ORDER BY count(*) DESC, observation_datetime +LIMIT 20 +""" + +TOTALS = """ +SELECT count(*) AS duplicate_groups, + coalesce(sum(copies) - count(*), 0) AS rows_above_the_first, + count(*) FILTER (WHERE distinct_values > 1) AS groups_that_disagree +FROM ( + SELECT count(*) AS copies, count(DISTINCT value) AS distinct_values + FROM transducer_observation + GROUP BY deployment_id, parameter_id, observation_datetime + HAVING count(*) > 1 +) g +""" + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--instance", required=True, help="PROJECT:REGION:INSTANCE") + parser.add_argument("--database", required=True, help="e.g. ocotillo-staging") + parser.add_argument("--user", help="IAM principal; defaults to your gcloud account") + args = parser.parse_args() + + user = args.user or _current_account() + if not user: + print("Could not determine your gcloud account; pass --user.", file=sys.stderr) + return 2 + + from google.cloud.sql.connector import Connector + + connector = Connector() + try: + conn = connector.connect( + args.instance, + "pg8000", + user=user, + db=args.database, + enable_iam_auth=True, + ) + except Exception as exc: # noqa: BLE001 - the message is the useful part + print(f"Could not connect as {user}: {exc}", file=sys.stderr) + print( + "\nIf this is a permissions error, the principal probably has no " + "database user:\n" + f" gcloud sql users create {user} --instance=" + f"{args.instance.split(':')[-1]} --type=cloud_iam_user", + file=sys.stderr, + ) + return 1 + + try: + cursor = conn.cursor() + cursor.execute(TOTALS) + groups, extra_rows, disagreeing = cursor.fetchone() + + print(f"Database: {args.database}") + print(f" duplicate groups : {groups}") + print(f" rows above the first : {extra_rows}") + print(f" groups that disagree : {disagreeing}") + + if not groups: + print("\nNo duplicates. The unique constraint migration is safe to run.") + return 0 + + print( + "\nThe migration will FAIL until these are resolved.\n" + "Groups that disagree are the ones to look at first: those rows hold " + "different values for the same instant, so they are conflicting " + "measurements rather than redundant copies, and collapsing them " + "discards a reading somebody recorded." + ) + cursor.execute(DUPLICATE_GROUPS) + print("\n deployment parameter observed copies values") + for dep, param, observed, copies, values in cursor.fetchall(): + print( + f" {dep:>10} {param:>9} {str(observed):<24} {copies:>6} {values:>6}" + ) + return 1 + finally: + conn.close() + connector.close() + + +def _current_account() -> str | None: + import subprocess + + try: + result = subprocess.run( + ["gcloud", "config", "get-value", "account"], + capture_output=True, + text=True, + timeout=30, + ) + except Exception: # noqa: BLE001 + return None + account = result.stdout.strip() + return account or None + + +if __name__ == "__main__": + raise SystemExit(main()) + + +# ============= EOF =============================================