diff --git a/README.md b/README.md index d91abf8..e4cb2e7 100644 --- a/README.md +++ b/README.md @@ -369,7 +369,7 @@ Run once per `(field, disease)` job by `pipeline/job.py::run_job`: 5. **Phenology** — carry-forward of latest observation ≤ day; future days null; if today's phase is missing, use yesterday's 6. **Applied treatments** — one query loads crop-protection sprays over `as_of-25 .. as_of` from `AI_agrosupport_agro_ril_operations*` with `rilop_operation = 10`. Days in `as_of-5 .. as_of` fill meteorological `applied_treatment`. If the latest spray in the full 25-day window still has a non-NULL `product_uses.treatment_interval_min_days` strictly greater than days elapsed (any tank-mix partner), query synthesis, prefilter, and vector search are skipped and `allowed_products` stays empty. `last_applied_treatment` still uses only `as_of-25 .. as_of-6`, and only when the recent window is empty. 7. **LLM** — synthesise 1–2 anonymised search queries from the subset -8. **Product prefilter** — organic / crop / disease gates, then one prefilter rule against the batch-wide, in-memory `products` index (`pipeline/stages/disease.py::select_prefilter_rule`, first match wins): +8. **Product prefilter** — legal catalog state, then organic / crop / disease gates, then one prefilter rule against the batch-wide, in-memory `products` index (`pipeline/stages/disease.py::select_prefilter_rule`, first match wins). A product stays eligible only when `registration_number` matches `AI_agrosupport_products_catalog.prodgov_registr` and `prodgov_state` is a non-blank value other than `Revocato`; revoked rows remain in the index for interval checks and applied-treatment labels: | Rule | Predicates | |---|---| @@ -379,7 +379,7 @@ Run once per `(field, disease)` job by `pipeline/job.py::run_job`: | `FASE5` both past/today and future | `systemicity == mixed` and `preventive_action = 1` and `curative_action = 1` and `eradicant_action = 0` | | `FASE5` in the future only | same predicates as the `FASE2`-only rule | | `FASE5` today or in the past only | `systemicity` in `{contact, mixed}` and `preventive_action = 1` and `curative_action = 0` and `eradicant_action = 0` | - | none of the above | no FASE-specific cut; organic/crop/disease gates only | + | none of the above | no FASE-specific cut; legal-state / organic / crop / disease gates only | This stage is skipped (along with query synthesis and vector search) when the latest spray's min treatment interval is still open; see stage 6. diff --git a/pipeline/stages/products.py b/pipeline/stages/products.py index ab8366f..5697c89 100644 --- a/pipeline/stages/products.py +++ b/pipeline/stages/products.py @@ -5,6 +5,7 @@ from __future__ import annotations import logging from dataclasses import dataclass from datetime import date +from typing import Any import pyodbc @@ -38,6 +39,7 @@ class ProductRow: target_diseases: list[str] active_ingredients: list[str] frac_groups: list[str] + permitted: bool = True # catalog prodgov_state is a non-blank value other than Revocato def as_candidate(self) -> ProductCandidate: return ProductCandidate( @@ -72,6 +74,73 @@ WHERE crop = ? AND product_id IN ({placeholders}) """ +_CATALOG_STATE_SQL = """ +SELECT prodgov_registr, prodgov_state +FROM AI_agrosupport_products_catalog +WHERE prodgov_registr IS NOT NULL +""" + +_REVOKED_STATE = "revocato" + + +def _normalize_registration(value: Any) -> str | None: + """Ministry registration key: stripped, digit-only values zero-padded to 6.""" + if value is None: + return None + cleaned = str(value).strip() + if not cleaned: + return None + if cleaned.isdigit(): + return cleaned.zfill(6) + return cleaned + + +def _state_permits(state: str | None) -> bool: + """True when prodgov_state is a non-blank value other than Revocato.""" + if state is None: + return False + folded = str(state).strip().casefold() + return bool(folded) and folded != _REVOKED_STATE + + +def _permitted_by_registration(catalog_rows: list[Any]) -> dict[str, bool]: + """ + Map a normalized prodgov_registr to whether any catalog row permits it. + + One permitted row is enough when several rows share a registration. Distinct + states for the same registration are logged. + """ + states_by_registration: dict[str, set[str]] = {} + for row in catalog_rows: + key = _normalize_registration(row.prodgov_registr) + if key is None: + continue + raw_state = row.prodgov_state + folded = "" if raw_state is None else str(raw_state).strip().casefold() + states_by_registration.setdefault(key, set()).add(folded) + + permitted: dict[str, bool] = {} + for key, states in states_by_registration.items(): + permitted[key] = any(_state_permits(state) for state in states) + if len(states) > 1: + logger.warning( + "Catalog registration %s has disagreeing prodgov_state values: %s", + key, + sorted(state or "(blank)" for state in states), + ) + return permitted + + +def _registration_permitted( + registration_number: str | None, + permitted_by_registration: dict[str, bool], +) -> tuple[bool, bool]: + """Return (permitted, unmatched). A known revoked registration is not unmatched.""" + key = _normalize_registration(registration_number) + if key is None or key not in permitted_by_registration: + return False, True + return permitted_by_registration[key], False + class ProductIndex: """ @@ -107,25 +176,48 @@ class ProductIndex: @classmethod def load(cls, conn: pyodbc.Connection) -> "ProductIndex": - rows = [ - ProductRow( - product_id=str(r.product_id), - product_name=str(r.product_name).strip(), - registration_number=( - str(r.registration_number) if r.registration_number is not None else None - ), - organic_certified=bool(r.organic_certified), - systemicity=(r.systemicity or "").strip(), - preventive_action=bool(r.preventive_action), - curative_action=bool(r.curative_action), - eradicant_action=bool(r.eradicant_action), - target_crops=parse_json_list(r.target_crops), - target_diseases=parse_json_list(r.target_diseases), - active_ingredients=parse_json_list(r.active_ingredients), - frac_groups=parse_json_list(r.frac_groups), + product_rows = fetch_all(conn, _PRODUCTS_SQL) + catalog_rows = fetch_all(conn, _CATALOG_STATE_SQL) + permitted_by_registration = _permitted_by_registration(catalog_rows) + + rows: list[ProductRow] = [] + not_permitted = 0 + unmatched = 0 + for record in product_rows: + registration_number = ( + str(record.registration_number) if record.registration_number is not None else None ) - for r in fetch_all(conn, _PRODUCTS_SQL) - ] + permitted, missing_catalog = _registration_permitted( + registration_number, permitted_by_registration + ) + if not permitted: + not_permitted += 1 + if missing_catalog: + unmatched += 1 + rows.append( + ProductRow( + product_id=str(record.product_id), + product_name=str(record.product_name).strip(), + registration_number=registration_number, + organic_certified=bool(record.organic_certified), + systemicity=(record.systemicity or "").strip(), + preventive_action=bool(record.preventive_action), + curative_action=bool(record.curative_action), + eradicant_action=bool(record.eradicant_action), + target_crops=parse_json_list(record.target_crops), + target_diseases=parse_json_list(record.target_diseases), + active_ingredients=parse_json_list(record.active_ingredients), + frac_groups=parse_json_list(record.frac_groups), + permitted=permitted, + ) + ) + logger.info( + "Catalog legal state: %d registration(s); %d product(s) not permitted, " + "%d unmatched (no registration or no catalog row)", + len(permitted_by_registration), + not_permitted, + unmatched, + ) return cls(rows) def candidates_for(self, crop_english: str, disease_english: str) -> list[ProductRow]: @@ -222,8 +314,12 @@ def prefilter_products( rule: str | None, ) -> list[ProductCandidate]: """ - Filter the (crop, disease) bucket by organic flag and the selected - prefilter rule (see pipeline.stages.disease.select_prefilter_rule). + Filter the (crop, disease) bucket by legal catalog state, organic flag, + and the selected prefilter rule (see pipeline.stages.disease.select_prefilter_rule). + + A product is kept only when ProductRow.permitted is true: its registration + matches the government catalog and prodgov_state is a non-blank value other + than Revocato. Revoked rows stay in the index for by_name. observation -> systemicity in {systemic, mixed} AND eradicant_action = 1 incubazprimaria -> systemicity in {systemic, mixed} AND curative_action = 1 @@ -235,11 +331,13 @@ def prefilter_products( fase5_future -> same predicates as fase2_only fase5_past -> systemicity in {contact, mixed} AND preventive_action = 1 AND curative_action = 0 AND eradicant_action = 0 - None -> no FASE-specific cut; crop/disease/organic gates only + None -> no FASE-specific cut; legal-state / crop / disease / organic gates only """ candidates: list[ProductCandidate] = [] for row in index.candidates_for(crop_english, disease_english): + if not row.permitted: + continue if organic and not row.organic_certified: continue diff --git a/tests/test_products.py b/tests/test_products.py index 0e90476..fdc09c0 100644 --- a/tests/test_products.py +++ b/tests/test_products.py @@ -1,4 +1,4 @@ -"""Unit tests for the treatment-interval skip helper.""" +"""Unit tests for the product index, legal-state gate, and treatment-interval skip.""" from __future__ import annotations @@ -7,7 +7,15 @@ from datetime import date from types import SimpleNamespace from unittest.mock import MagicMock, patch -from pipeline.stages.products import ProductIndex, ProductRow, last_treatment_interval_open +from pipeline.stages.products import ( + ProductIndex, + ProductRow, + _normalize_registration, + _permitted_by_registration, + _state_permits, + last_treatment_interval_open, + prefilter_products, +) _AS_OF = date(2026, 4, 16) _SPRAY_DATE = date(2026, 4, 6) # 10 days before as_of @@ -19,6 +27,7 @@ def _product( *, product_id: str = "p1", product_name: str = "LIETO", + permitted: bool = True, ) -> ProductRow: return ProductRow( product_id=product_id, @@ -33,6 +42,7 @@ def _product( target_diseases=[_DISEASE], active_ingredients=["mancozeb"], frac_groups=[], + permitted=permitted, ) @@ -163,5 +173,106 @@ class LastTreatmentIntervalOpenTests(unittest.TestCase): self.assertFalse(result) +class LegalStateTests(unittest.TestCase): + def test_state_permits_only_a_non_blank_value_other_than_revocato(self) -> None: + self.assertFalse(_state_permits("Revocato")) + self.assertFalse(_state_permits("revocato")) + self.assertFalse(_state_permits(" Revocato ")) + self.assertFalse(_state_permits("")) + self.assertFalse(_state_permits(" ")) + self.assertFalse(_state_permits(None)) + self.assertTrue(_state_permits("Autorizzato")) + self.assertTrue(_state_permits("Ri-registrato")) + + def test_digit_registration_is_zero_padded_to_six(self) -> None: + self.assertEqual(_normalize_registration("13817"), "013817") + self.assertEqual(_normalize_registration("013817"), "013817") + self.assertEqual(_normalize_registration(" 842 "), "000842") + self.assertIsNone(_normalize_registration(None)) + self.assertIsNone(_normalize_registration(" ")) + + def test_disagreeing_states_permit_when_any_row_is_legal(self) -> None: + rows = [ + SimpleNamespace(prodgov_registr="000013", prodgov_state="Revocato"), + SimpleNamespace(prodgov_registr="13", prodgov_state="Autorizzato"), + ] + with self.assertLogs("pipeline.stages.products", level="WARNING") as logs: + permitted = _permitted_by_registration(rows) + self.assertTrue(permitted["000013"]) + self.assertTrue(any("disagreeing" in message for message in logs.output)) + + def test_prefilter_drops_revoked_and_keeps_four_legal_candidates(self) -> None: + legal = [ + _product(product_id=f"p{i}", product_name=f"LEGAL{i}") + for i in range(4) + ] + revoked = _product(product_id="rev", product_name="REVOKED", permitted=False) + index = ProductIndex([legal[0], legal[1], revoked, legal[2], legal[3]]) + + candidates = prefilter_products( + index, + crop_english=_CROP, + disease_english=_DISEASE, + organic=False, + rule=None, + ) + + self.assertEqual([item.product_id for item in candidates], ["p0", "p1", "p2", "p3"]) + self.assertIs(index.by_name("REVOKED"), revoked) + self.assertIn(revoked, index.candidates_for(_CROP, _DISEASE)) + + +def _sql_product(**overrides: object) -> SimpleNamespace: + fields: dict[str, object] = { + "product_id": "p1", + "product_name": "LIETO", + "registration_number": "000001", + "organic_certified": 0, + "systemicity": "contact", + "preventive_action": 1, + "curative_action": 0, + "eradicant_action": 0, + "target_crops": '["grapevine"]', + "target_diseases": '["downy mildew"]', + "active_ingredients": '["mancozeb"]', + "frac_groups": "[]", + } + fields.update(overrides) + return SimpleNamespace(**fields) + + +class ProductIndexLoadTests(unittest.TestCase): + def test_load_sets_permitted_from_catalog_and_keeps_revoked_by_name(self) -> None: + products = [ + _sql_product(product_id="alial", product_name="ALIAL", registration_number="13817"), + _sql_product( + product_id="microsulf", + product_name="MICROSULF", + registration_number="000842", + ), + _sql_product(product_id="ghost", product_name="GHOST", registration_number=None), + ] + catalog = [ + SimpleNamespace(prodgov_registr="013817", prodgov_state="Autorizzato"), + SimpleNamespace(prodgov_registr="000842", prodgov_state="Revocato"), + ] + with patch( + "pipeline.stages.products.fetch_all", + side_effect=[products, catalog], + ) as fetch: + index = ProductIndex.load(MagicMock()) + + self.assertEqual(fetch.call_count, 2) + self.assertIn("FROM products", fetch.call_args_list[0].args[1]) + self.assertIn("AI_agrosupport_products_catalog", fetch.call_args_list[1].args[1]) + self.assertTrue(index.by_name("ALIAL").permitted) + revoked = index.by_name("MICROSULF") + self.assertIsNotNone(revoked) + self.assertFalse(revoked.permitted) + unmatched = index.by_name("GHOST") + self.assertIsNotNone(unmatched) + self.assertFalse(unmatched.permitted) + + if __name__ == "__main__": unittest.main()