Skip to content

Commit 9e27338

Browse files
eliperez-develiperez-dev
committed
Integrate approved FSIS acquisition guard
1 parent 2528847 commit 9e27338

8 files changed

Lines changed: 685 additions & 2 deletions

File tree

‎data/manifests/us-fsis-sprint02-20260919.json‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
{
2+
"manifest_version": "us-fsis-sprint02-capture-v1",
3+
"source_id": "us.fsis",
4+
"authority": "USDA Food Safety and Inspection Service",
5+
"page_url": "https://www.fsis.usda.gov/inspection/establishments/meat-poultry-and-egg-product-inspection-directory",
6+
"page_observed_in_normal_browser": true,
7+
"page_observed_at": "2026-09-19",
8+
"page_last_updated_observed": "2026-09-14",
9+
"displayed_file_edition": "2026-09-14",
10+
"current_file_routes": [
11+
"https://www.fsis.usda.gov/sites/default/files/media_file/documents/MPI_Directory_by_Establishment_Name.csv",
12+
"https://www.fsis.usda.gov/sites/default/files/media_file/documents/MPI_Directory_by_Establishment_Number.csv",
13+
"https://www.fsis.usda.gov/sites/default/files/media_file/documents/Dataset_Establishment_Demographic_Data.csv"
14+
],
15+
"api_documentation": "https://www.fsis.usda.gov/science-data/developer-resources/mpi-api",
16+
"official_catalog_check": {
17+
"catalog_url": "https://catalog.data.gov/dataset/fsis-mpi-meat-poultry-and-egg-inspection-directory-by-establishment-number",
18+
"identifier": "USDA-FSIS-02246",
19+
"publisher": "Food Safety and Inspection Service",
20+
"catalog_last_updated": "2025-01-22",
21+
"distribution_access_url_is_same_fsis_landing_page": true,
22+
"alternate_current_artifact": false
23+
},
24+
"authorization": {
25+
"private_acquisition": "authorized by Sprint 02 contract and delegated lane 3 scope",
26+
"source_terms_clearance": "unknown",
27+
"rights_clearance": "unknown",
28+
"publication_authorized": false
29+
},
30+
"current_raw_artifacts": {
31+
"status": "not_captured",
32+
"directory_by_number_http_status": 403,
33+
"directory_by_name_http_status": 403,
34+
"demographics_http_status": 403,
35+
"bounded_ordinary_get": {
36+
"status": "confirmed_blocked",
37+
"max_bytes": 134217728,
38+
"max_time_seconds": 90,
39+
"raw_artifacts_captured": false,
40+
"private_manifest": "private-handoff/bounded-get-20260919.json"
41+
},
42+
"ordinary_browser_link_capture": "no local artifact path exposed by the permitted in-app browser",
43+
"bypass_attempted": false,
44+
"raw_artifacts_captured": false
45+
},
46+
"legacy_comparison": {
47+
"status": "current_not_observed",
48+
"private_report": "private-handoff/legacy-comparison-20260919.json",
49+
"legacy_rows": 7101,
50+
"legacy_source_native_establishments": 7101,
51+
"additions": null,
52+
"not_observed": null,
53+
"missing_current_observation_is_not_closure": true
54+
},
55+
"release_state": "not-created",
56+
"publication_state": "blocked",
57+
"row_payloads_included": false,
58+
"limitations": [
59+
"HTTP 403 was recorded without attempting an access-control bypass.",
60+
"The dashboard and displayed edition are source observations, not a row-level current artifact.",
61+
"No current row-level additions, duplicates, category counts, or identity continuity can be asserted until an authorized artifact capture is available.",
62+
"The private handoff contains only row-free capture evidence and the legacy aggregate comparison."
63+
]
64+
}

‎docs/countries/us/operator-refresh.md‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,24 @@ disabled. A failed lane cannot delete or replace the previous-valid manifest.
8181
The failure record contains an actionable class and fallback without copying
8282
source rows into the aggregate report.
8383

84+
## FSIS legacy comparison
85+
86+
The FSIS adapter manifest includes row-free source metrics for directory rows,
87+
source-native establishments, duplicate identities, and activity-category
88+
coverage. Compare the checked-in historical snapshot without claiming it is
89+
current:
90+
91+
```powershell
92+
python -m pipeline.sources.us.fsis.legacy_compare `
93+
--legacy static_data/us/locations.csv `
94+
--output <private-handoff>/legacy-comparison.json
95+
```
96+
97+
When a current parsed artifact exists, pass it with `--current-records` (and
98+
its adapter manifest with `--current-manifest`) to compute exact-key additions
99+
and not-observed counts. Without a current artifact both values remain
100+
`unknown`; not-observed never means closure.
101+
84102
## Diagnostics and retry behavior
85103

86104
Inspect an existing run without opening raw, parsed, normalized, or quarantine

‎pipeline/sources/us/fsis/adapter.py‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -106,6 +106,64 @@ def _identity_key(row: dict[str, Any]) -> str | None:
106106
return candidates[0] if candidates else None
107107

108108

109+
def _source_metrics(
110+
directory_rows: list[dict[str, Any]],
111+
demographic_rows: list[dict[str, Any]],
112+
parsed_records: Iterable[dict[str, Any]],
113+
duplicate_directory_aliases: set[str],
114+
duplicate_demographic_aliases: set[str],
115+
) -> dict[str, Any]:
116+
"""Return row-free reconciliation facts for the private handoff.
117+
118+
Counts intentionally keep source rows, source-native identities, and
119+
activity categories separate. A source disappearance is not represented
120+
as closure here; comparison code owns that explicit not-observed state.
121+
"""
122+
parsed = list(parsed_records)
123+
124+
def identity_set(rows: Iterable[dict[str, Any]]) -> set[str]:
125+
return {key for row in rows if (key := _identity_key(row))}
126+
127+
def duplicate_row_count(rows: Iterable[dict[str, Any]], aliases: set[str]) -> int:
128+
return sum(1 for row in rows if set(_key_candidates(row)) & aliases)
129+
130+
categories = Counter()
131+
activity_fields = Counter()
132+
for item in parsed:
133+
normalized = item.get("normalized", {})
134+
slaughter = bool(normalized.get("species_slaughtered"))
135+
processing = bool(normalized.get("processing_activities"))
136+
if slaughter:
137+
categories["slaughter"] += 1
138+
if processing:
139+
categories["processing"] += 1
140+
if slaughter and processing:
141+
categories["slaughter_and_processing"] += 1
142+
if not slaughter and not processing:
143+
categories["no_activity_category"] += 1
144+
for group in (normalized.get("species_slaughtered", {}), normalized.get("processing_activities", {})):
145+
for field in group:
146+
activity_fields[field] += 1
147+
148+
return {
149+
"directory_source_rows": len(directory_rows),
150+
"demographic_source_rows": len(demographic_rows),
151+
"source_native_establishments": len(identity_set(directory_rows)),
152+
"missing_directory_identity_rows": sum(1 for row in directory_rows if not _identity_key(row)),
153+
"duplicate_directory_aliases": len(duplicate_directory_aliases),
154+
"duplicate_directory_rows": duplicate_row_count(directory_rows, duplicate_directory_aliases),
155+
"duplicate_demographic_aliases": len(duplicate_demographic_aliases),
156+
"duplicate_demographic_rows": duplicate_row_count(demographic_rows, duplicate_demographic_aliases),
157+
"category_coverage": {
158+
"slaughter_rows": categories["slaughter"],
159+
"processing_rows": categories["processing"],
160+
"slaughter_and_processing_rows": categories["slaughter_and_processing"],
161+
"no_activity_category_rows": categories["no_activity_category"],
162+
},
163+
"activity_field_row_counts": dict(sorted(activity_fields.items())),
164+
}
165+
166+
109167
def _coordinate(row: dict[str, Any]) -> tuple[dict[str, Any] | None, str, str | None]:
110168
latitude = _field(row, "latitude", "lat", "y")
111169
longitude = _field(row, "longitude", "lon", "lng", "long", "x")
@@ -370,6 +428,13 @@ def parse_sources(self, directory: bytes, demographics: bytes | None = None) ->
370428
"matched_demographic_rows": len(matched_demographics),
371429
"orphan_demographic_rows": orphan_demographics,
372430
"identity_conflicts": identity_conflicts,
431+
"source_metrics": _source_metrics(
432+
directory_rows,
433+
demographic_rows,
434+
accepted + [item["record"] for item in quarantined],
435+
duplicate_directory_aliases,
436+
duplicate_demographic_aliases,
437+
),
373438
}
374439

375440
def run(self, raw_path: str | Path, run_dir: str | Path, artifact: SourceArtifact) -> dict[str, Any]:
@@ -437,6 +502,7 @@ def run_sources(self, raw_paths: dict[str, bytes | str | Path], run_dir: str | P
437502
"orphan_demographic_rows": result["orphan_demographic_rows"],
438503
"identity_conflicts": result["identity_conflicts"], "unmatched_demographic_is_not_closure": True,
439504
},
505+
"source_metrics": result["source_metrics"],
440506
"geocoding": "disabled",
441507
"coverage": "FSIS-regulated meat, poultry, and egg establishments in the captured edition; state-inspection programs and non-FSIS populations excluded",
442508
"publication_state": "private-candidate",

0 commit comments

Comments
 (0)