diff --git a/CHANGELOG.md b/CHANGELOG.md index 7fdf48f3..975ccda7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -52,6 +52,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 now rejected up front, in the form and in `FileConfig`, with an instruction to save the file as `.xlsx`; README, User Guide and Architecture no longer list `.xls` (`src/datasure/connectors/local.py`) +- **Correction replay**: Replay silently overwrote a value fixed at source when + the corrected value was originally missing. `_current_value_mismatch` no + longer skips a missing recorded value, so the correction applies only while + the cell is still missing and otherwise fails with a reason; the same check + runs when a correction is first applied + (`src/datasure/processing/corrections.py`) — #337 +- **Acceptance status**: An accept row's Status in the Correction Log was + judged against the data at that point in replay, so an accept followed by a + modify of the same cell showed `Successful` while the check pages treated it + as inactive. Accept statuses are now judged against the final corrected data, + as `get_active_acceptances` does, on replay and on every log save — closes #337 +- **Correction KEY lookup**: Replay and acceptance statuses found a record by + the first column holding its KEY value, so a value repeated in another column + could point them at the wrong record and mark an active acceptance + `Inactive`. They now use the KEY column the check config sets for the + dataset (the survey or backcheck KEY), falling back to the old lookup only + when none is set (`src/datasure/processing/corrections.py`) — #337 - **Remove Credentials**: Reported success but left the SurveyCTO password in the OS keyring. The view passed the credential type as the server, and `delete_stored_credentials` read the username from the wrong level of the diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 5d52ab02..408c731d 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -113,6 +113,11 @@ also reads. entries all or nothing. GPS coordinate changes (one entry per coordinate column) and duplicate card saves (one entry per changed record) use it. - **`status` / `status_reason`**: the outcome of the last replay of each entry. + An accept row is `Inactive` when `get_active_acceptances` would drop it; + its status is judged against the final corrected data on replay and on + every log append. Replay finds each entry's record by the KEY column the + check config sets for the dataset, or, if none is set, by the first column + holding the entry's KEY. The shared correction form lives in `utils/correction_form.py` (`render_correction_form`, `render_correction_inputs`, diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index 0ece9164..c8b24a9f 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -603,7 +603,8 @@ refresh: reviewed** to see corrected values marked Corrected. 3. In the Correction Log, check that the entry's Status is `Successful`. If new data or a prep change means a correction can no longer be applied, its - Status shows why. + Status shows why. This includes a correction of a blank value that was later + filled in at source: the source value is kept and the correction fails. **Demo Example**: Correcting duplicate household ID: diff --git a/src/datasure/processing/corrections.py b/src/datasure/processing/corrections.py index 149f7003..d7830055 100644 --- a/src/datasure/processing/corrections.py +++ b/src/datasure/processing/corrections.py @@ -8,6 +8,7 @@ import polars as pl import streamlit as st +from datasure.checks.backchecks.models import backcheck_id_columns from datasure.processing.correction_log import ( ACCEPT_CHECK_TYPES, CORRECTION_ACTIONS, @@ -25,6 +26,7 @@ ) from datasure.utils.reapply_utils import ReapplyFailure from datasure.utils.reviewer import get_reviewer_name +from datasure.utils.settings_utils import page_settings_file def _describe_correction_row(row: dict[str, Any]) -> str: @@ -426,6 +428,7 @@ def add_correction_entry( reason: str, source: str = CORRECTIONS_PAGE_SOURCE, check_type: str | None = None, + key_col: str | None = None, ) -> None: """Add a new correction entry to the log. @@ -453,6 +456,9 @@ def add_correction_entry( check page such as "outliers" check_type : str | None For "accept" entries, the check whose flag was accepted + key_col : str | None + The key column name. If None, the KEY column the check config + sets for `alias` is used. """ self._append_log_rows( alias, @@ -470,9 +476,12 @@ def add_correction_entry( user=get_reviewer_name(), ) ], + key_col, ) - def _append_log_rows(self, alias: str, rows: list[dict[str, Any]]) -> None: + def _append_log_rows( + self, alias: str, rows: list[dict[str, Any]], key_col: str | None = None + ) -> None: """Append rows to the correction log in a single save. Parameters @@ -481,6 +490,9 @@ def _append_log_rows(self, alias: str, rows: list[dict[str, Any]]) -> None: The data alias/table name rows : list[dict[str, Any]] Log rows built by `_build_log_row` + key_col : str | None + The key column name. If None, the KEY column the check config + sets for `alias` is used. """ current_log = self.get_correction_log(alias) new_rows = pl.DataFrame(rows, schema=CORRECTION_LOG_SCHEMA) @@ -495,6 +507,14 @@ def _append_log_rows(self, alias: str, rows: list[dict[str, Any]]) -> None: ) updated_log = pl.concat([aligned_current_log, new_rows]) + # The new rows may change a value an earlier acceptance recorded. + # Callers save the corrected data first, so it is current here. + if (updated_log["action"] == Action.ACCEPT).any(): + data = self.get_corrected_data(alias) + updated_log = self._with_acceptance_statuses( + updated_log, data, key_col or self._configured_key_column(alias, data) + ) + duckdb_save_table( project_id=self.project_id, table_data=updated_log, @@ -571,6 +591,7 @@ def accept_value( reason=reason, source=source or check_type, check_type=check_type, + key_col=key_col, ) def get_active_acceptances( @@ -731,6 +752,7 @@ def apply_correction( current_value=current_value, new_value=new_value, reason=reason, + key_col=key_col, ) return corrected_data @@ -795,7 +817,7 @@ def apply_corrections( self.save_corrected_data(alias, corrected_data) try: - self._append_log_rows(alias, log_rows) + self._append_log_rows(alias, log_rows, key_col) except Exception: # Keep data and log in step: undo the data change, then re-raise. self.save_corrected_data(alias, original_data) @@ -835,10 +857,8 @@ def _apply_entry( if entry.action not in CORRECTION_ACTIONS: raise ValueError(f"Unknown correction action '{entry.action}'") - if ( - entry.action in (Action.MODIFY_VALUE, Action.REMOVE_VALUE) - and entry.current_value is not None - ): + # A missing current value is checked too: the cell must still be missing. + if entry.action in (Action.MODIFY_VALUE, Action.REMOVE_VALUE) and entry.column: _check_acceptance_against_data( data, key_col, entry.key_value, entry.column, entry.current_value ) @@ -1187,18 +1207,16 @@ def _reapply_all_corrections(self, alias: str) -> list[ReapplyFailure]: return [] correction_log = self.get_correction_log(alias) + key_col = self._configured_key_column(alias, fresh_data) corrected_data = fresh_data failures: list[ReapplyFailure] = [] statuses: list[str] = [] status_reasons: list[str | None] = [] for row in correction_log.iter_rows(named=True): - if row["action"] == Action.ACCEPT: - inactive_reason = self._inactive_acceptance_reason(corrected_data, row) - statuses.append("Inactive" if inactive_reason else "Successful") - status_reasons.append(inactive_reason) - continue - corrected_data, error = self._apply_correction_row(corrected_data, row) + corrected_data, error = self._apply_correction_row( + corrected_data, row, key_col + ) if error: statuses.append("Failed") status_reasons.append(error) @@ -1209,9 +1227,15 @@ def _reapply_all_corrections(self, alias: str) -> list[ReapplyFailure]: statuses.append("Successful") status_reasons.append(None) - refreshed_log = correction_log.with_columns( - pl.Series("status", statuses, dtype=pl.String), - pl.Series("status_reason", status_reasons, dtype=pl.String), + # Accept rows don't change the data; their status depends on the final + # data, as for `get_active_acceptances`. + refreshed_log = self._with_acceptance_statuses( + correction_log.with_columns( + pl.Series("status", statuses, dtype=pl.String), + pl.Series("status_reason", status_reasons, dtype=pl.String), + ), + corrected_data, + key_col, ) duckdb_save_table( project_id=self.project_id, @@ -1226,7 +1250,7 @@ def _reapply_all_corrections(self, alias: str) -> list[ReapplyFailure]: return failures def _apply_correction_row( - self, data: pl.DataFrame, row: dict[str, Any] + self, data: pl.DataFrame, row: dict[str, Any], key_col: str | None = None ) -> tuple[pl.DataFrame, str | None]: """Apply one correction-log row to data. @@ -1241,6 +1265,9 @@ def _apply_correction_row( The data to apply the correction to row : dict[str, Any] A single row from the correction log + key_col : str | None + The key column name. If None, the first column holding the row's + KEY is used. Returns ------- @@ -1255,7 +1282,7 @@ def _apply_correction_row( return data, None key_value = row["KEY"] - key_col = self._find_key_column(data, key_value) + key_col = self._find_key_column(data, key_value, key_col) if not key_col: return data, f"Key '{key_value}' not found in current data" @@ -1291,11 +1318,63 @@ def _apply_correction_row( return data, None + def _configured_key_column(self, alias: str, data: pl.DataFrame) -> str | None: + """Return the KEY column the check config sets for `alias`. + + That is the `survey_key` of a page whose survey data is `alias`, or + the backcheck KEY of a page whose backcheck data is `alias` (see + `backcheck_id_columns`). None if no page sets one that is in `data`. + """ + pages = duckdb_get_table( + project_id=self.project_id, alias="check_config", db_name="logs" + ) + for page in pages.iter_rows(named=True): + if page.get("survey_data_name") == alias: + key_col = page.get("survey_key") + elif page.get("backcheck_data_name") == alias: + key_col, _ = backcheck_id_columns( + str(page_settings_file(self.project_id, page["page_name"])), + page.get("survey_key"), + page.get("survey_id"), + ) + else: + continue + if key_col in data.columns: + return key_col + return None + + def _with_acceptance_statuses( + self, log: pl.DataFrame, data: pl.DataFrame, key_col: str | None + ) -> pl.DataFrame: + """Return `log` with each accept row's status judged against `data`. + + An accept row is `Successful` while `data` holds its recorded value + and `Inactive`, with the reason, once it doesn't. Other rows are + returned unchanged. Records are found by `key_col`, or, if None, by + the first column holding the row's KEY. + """ + if log.is_empty(): + return log + statuses: list[str | None] = [] + status_reasons: list[str | None] = [] + for row in log.iter_rows(named=True): + if row["action"] == Action.ACCEPT: + inactive_reason = self._inactive_acceptance_reason(data, row, key_col) + statuses.append("Inactive" if inactive_reason else "Successful") + status_reasons.append(inactive_reason) + else: + statuses.append(row["status"]) + status_reasons.append(row["status_reason"]) + return log.with_columns( + pl.Series("status", statuses, dtype=pl.String), + pl.Series("status_reason", status_reasons, dtype=pl.String), + ) + def _inactive_acceptance_reason( - self, data: pl.DataFrame, row: dict[str, Any] + self, data: pl.DataFrame, row: dict[str, Any], key_col: str | None ) -> str | None: """Explain why an accept row no longer holds for `data`, or None if it does.""" - key_col = self._find_key_column(data, row["KEY"]) + key_col = key_col or self._find_key_column(data, row["KEY"]) if not key_col: return f"Key '{row['KEY']}' not found in current data" return _acceptance_mismatch(data, key_col, row["KEY"], _accepted_values(row)) @@ -1323,17 +1402,14 @@ def _current_value_mismatch( The column the correction targets recorded_value : str | None The value recorded in the log at the time the correction was - made, or None if no value was recorded + made. None means the value was missing, so the correction still + applies only while the value is missing. Returns ------- str | None - A description of the mismatch, or None if the value still - matches (or there is nothing recorded to compare against) + A description of the mismatch, or None if the value still matches """ - if recorded_value is None: - return None - actual_value = data.filter(pl.col(key_col) == key_value)[0, column] if not _values_match(actual_value, recorded_value): @@ -1345,8 +1421,10 @@ def _current_value_mismatch( return None @staticmethod - def _find_key_column(data: pl.DataFrame, key_value: str) -> str | None: - """Find the first column containing the given key value. + def _find_key_column( + data: pl.DataFrame, key_value: str, key_col: str | None = None + ) -> str | None: + """Find the column containing the given key value. Parameters ---------- @@ -1354,13 +1432,17 @@ def _find_key_column(data: pl.DataFrame, key_value: str) -> str | None: The data to search key_value : str The key value to look for + key_col : str | None + The configured key column. If given, only it is searched; + otherwise the first column holding `key_value` is used. Returns ------- str | None The matching column name, or None if not found """ - for col in data.columns: + candidates = data.columns if key_col is None else [key_col] + for col in candidates: try: if data[col].is_in([key_value]).any(): return col diff --git a/tests/checks/gpschecks/test_review.py b/tests/checks/gpschecks/test_review.py index 47e6b687..d75c62e1 100644 --- a/tests/checks/gpschecks/test_review.py +++ b/tests/checks/gpschecks/test_review.py @@ -395,7 +395,7 @@ def processor(monkeypatch) -> CorrectionProcessor: ) monkeypatch.setattr(proc, "get_correction_log", lambda alias: store["log"]) - def append(alias, rows): + def append(alias, rows, key_col=None): store["log"] = pl.concat( [store["log"], pl.DataFrame(rows)], how="diagonal_relaxed" ) diff --git a/tests/processing/test_corrections.py b/tests/processing/test_corrections.py index abde430e..c338383f 100644 --- a/tests/processing/test_corrections.py +++ b/tests/processing/test_corrections.py @@ -77,6 +77,9 @@ def correction_processor(mock_streamlit): with ( patch("datasure.processing.corrections.duckdb_get_table") as mock_get, patch("datasure.processing.corrections.duckdb_save_table") as mock_save, + # No check config: keep the mocked `duckdb_get_table` sequences to data + # and logs, and find records by the first column holding the KEY. + patch.object(CorrectionProcessor, "_configured_key_column", return_value=None), ): processor = CorrectionProcessor("test_project") # Clear all caches before each test @@ -2385,3 +2388,176 @@ def test_reapply_starts_from_raw(self, correction_processor, sample_data): assert failures == [] corrected = mock_save.call_args_list[-1].kwargs["table_data"] assert corrected["survey_key"].to_list() == ["key1", "key3"] + + +class TestReplayOfCorrectionsOfMissingValues: + """A correction of a missing value is rechecked on replay like any other: + it reapplies while the source value is still missing, and fails with a + reason if the source value was filled in since (#337). + """ + + @pytest.fixture + def processor(self, store): + _seed_prep(store, pl.DataFrame({"KEY": ["k1", "k2"], "age": [None, 30]})) + processor = CorrectionProcessor("p1") + processor.apply_corrections( + "survey", + "KEY", + [CorrectionEntry("k1", "modify value", "r", "age", None, "5")], + ) + return processor + + def test_reapplies_while_the_source_value_is_still_missing(self, processor): + failures = processor.refresh_corrected_data("survey") + + assert failures == [] + assert processor.get_corrected_data("survey")["age"].to_list() == [5, 30] + assert processor.get_correction_log("survey")["status"].to_list() == [ + "Successful" + ] + + def test_fails_when_the_source_value_was_filled_in(self, store, processor): + _seed_prep(store, pl.DataFrame({"KEY": ["k1", "k2"], "age": [8, 30]})) + + failures = processor.refresh_corrected_data("survey") + + assert len(failures) == 1 + assert "changed since this correction was recorded" in failures[0].reason + assert processor.get_corrected_data("survey")["age"].to_list() == [8, 30] + log = processor.get_correction_log("survey") + assert log["status"].to_list() == ["Failed"] + assert "found '8'" in log["status_reason"][0] + + def test_a_correction_of_a_filled_in_value_is_rejected_when_applied(self, store): + _seed_prep(store, pl.DataFrame({"KEY": ["k1"], "age": [8]})) + processor = CorrectionProcessor("p1") + + with pytest.raises(ValueError, match="has changed since it was flagged"): + processor.apply_corrections( + "survey", + "KEY", + [CorrectionEntry("k1", "modify value", "r", "age", None, "5")], + ) + + assert processor.get_correction_log("survey").is_empty() + + +class TestAcceptanceStatusMatchesActiveAcceptances: + """An accept row's Status agrees with `get_active_acceptances`, which + judges it against the final corrected data, not the data at the accept + row's position in the log (#337). + """ + + def _accept_then_modify(self, store): + _seed_prep(store, pl.DataFrame({"KEY": ["k1"], "age": [25]})) + processor = CorrectionProcessor("p1") + processor.accept_value("survey", "KEY", "k1", "outliers", "age", 25, "ok") + processor.apply_corrections( + "survey", + "KEY", + [CorrectionEntry("k1", "modify value", "r", "age", 25, "30")], + ) + return processor + + def test_replay_marks_an_acceptance_of_a_later_corrected_value_inactive( + self, store + ): + processor = self._accept_then_modify(store) + + processor.refresh_corrected_data("survey") + + assert processor.get_active_acceptances("survey", "outliers", "KEY").is_empty() + log = processor.get_correction_log("survey") + assert log["status"].to_list() == ["Inactive", "Successful"] + assert "has changed" in log["status_reason"][0] + + def test_saving_a_correction_marks_the_acceptance_it_overrides_inactive( + self, store + ): + processor = self._accept_then_modify(store) + + assert processor.get_active_acceptances("survey", "outliers", "KEY").is_empty() + log = processor.get_correction_log("survey") + assert log["status"].to_list() == ["Inactive", "Successful"] + + def test_an_acceptance_of_the_corrected_value_stays_successful(self, store): + processor = self._accept_then_modify(store) + processor.accept_value("survey", "KEY", "k1", "outliers", "age", 30, "ok") + + processor.refresh_corrected_data("survey") + + active = processor.get_active_acceptances("survey", "outliers", "KEY") + assert active["current_value"].to_list() == ["30"] + log = processor.get_correction_log("survey") + assert log["status"].to_list() == ["Inactive", "Successful", "Successful"] + + +class TestKeyColumnWhenAnotherColumnHoldsTheKey: + """Records are found by the configured KEY column, not the first column + that happens to hold the KEY value, when saving and when replaying. + """ + + DATA = pl.DataFrame( + {"other_id": ["k2", "k1"], "KEY": ["k1", "k2"], "age": [25, 30]} + ) + + @staticmethod + def _seed_config(store, **page): + store[("p1", "logs", "check_config")] = pl.DataFrame( + [{"page_name": "Page", "survey_id": None, **page}] + ) + + def test_saving_an_acceptance_keeps_it_successful(self, store): + _seed_prep(store, self.DATA) + processor = CorrectionProcessor("p1") + + processor.accept_value("survey", "KEY", "k1", "outliers", "age", 25, "ok") + + assert processor.get_correction_log("survey")["status"].to_list() == [ + "Successful" + ] + assert processor.get_active_acceptances("survey", "outliers", "KEY").height + + def test_replay_uses_the_configured_survey_key(self, store): + _seed_prep(store, self.DATA) + self._seed_config(store, survey_data_name="survey", survey_key="KEY") + processor = CorrectionProcessor("p1") + processor.accept_value("survey", "KEY", "k1", "outliers", "age", 25, "ok") + processor.apply_corrections( + "survey", + "KEY", + [CorrectionEntry("k2", "modify value", "r", "age", 30, "31")], + ) + + failures = processor.refresh_corrected_data("survey") + + assert failures == [] + assert processor.get_corrected_data("survey")["age"].to_list() == [25, 31] + assert processor.get_correction_log("survey")["status"].to_list() == [ + "Successful", + "Successful", + ] + + def test_replay_uses_the_configured_backcheck_key(self, store): + _seed_prep(store, self.DATA, alias="backcheck") + self._seed_config( + store, + survey_data_name="survey", + survey_key="other_id", + backcheck_data_name="backcheck", + ) + processor = CorrectionProcessor("p1") + processor.apply_corrections( + "backcheck", + "KEY", + [CorrectionEntry("k1", "modify value", "r", "age", 25, "26")], + ) + + with patch( + "datasure.processing.corrections.backcheck_id_columns", + return_value=("KEY", None), + ): + failures = processor.refresh_corrected_data("backcheck") + + assert failures == [] + assert processor.get_corrected_data("backcheck")["age"].to_list() == [26, 30]