shawn
457e7794a0
Modules: - parser.py: Full Excel parser (1,848 machines, Disney mapping, Pattern A+B parsing) - db_writer.py: SQLite writer with per-field update policy (seed + update modes) - backup.py: GPS/photo backup, restore, and compare - reporter.py: Comprehensive validation report generator - main.py: CLI entry point Database: assets.db with 1,848 machines across 36 columns Handbook: reference_handbook.md v2.0 — all sections finalized Report: validation_report.md — 184 OCR corrections, 143 unknown machines
294 lines
12 KiB
Python
294 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
📊 Canteen Seed Import — Validation Report Generator
|
|
|
|
Parses the Machine_List.xlsx (1,848 machines) and generates a comprehensive
|
|
markdown validation report at seed-data/validation_report.md.
|
|
|
|
Usage:
|
|
python3 reporter.py seed-data/Machine_List.xlsx
|
|
"""
|
|
|
|
import sys
|
|
import re
|
|
import os
|
|
from collections import Counter, defaultdict
|
|
|
|
import openpyxl
|
|
from parser import parse_excel, OCR_CORRECTIONS
|
|
|
|
|
|
# ─── OCR Correction Detection ────────────────────────────────────────────────
|
|
|
|
def _clean_raw(raw):
|
|
"""Strip non-alphanumeric characters (same as parse.valid_serial)."""
|
|
if raw is None:
|
|
return ""
|
|
return re.sub(r'[^a-zA-Z0-9]', '', str(raw).strip())
|
|
|
|
|
|
def _ocr_correct(cleaned):
|
|
"""Apply OCR character corrections."""
|
|
return ''.join(OCR_CORRECTIONS.get(ch, ch) for ch in cleaned)
|
|
|
|
|
|
# ─── Report Generation ───────────────────────────────────────────────────────
|
|
|
|
def generate_report(filepath):
|
|
"""Parse the Excel and produce the full markdown validation report."""
|
|
# ── Parse ────────────────────────────────────────────────────────────
|
|
machines = parse_excel(filepath)
|
|
total = len(machines)
|
|
|
|
# ── Also read raw serials from xlsx for OCR correction detection ─────
|
|
wb = openpyxl.load_workbook(filepath, data_only=True)
|
|
ws = wb.active
|
|
raw_serials = {}
|
|
# Column 38 = "Serial Number" (1-based)
|
|
serial_col = 38
|
|
for row_idx in range(2, ws.max_row + 1):
|
|
asset_id = str(ws.cell(row_idx, 3).value or "").strip()
|
|
raw_ser = ws.cell(row_idx, serial_col).value
|
|
raw_serials[asset_id] = raw_ser
|
|
wb.close()
|
|
|
|
# ── Compute all stats ────────────────────────────────────────────────
|
|
|
|
# By class
|
|
class_counts = Counter(m.get("class", "Unknown") for m in machines)
|
|
|
|
# By make
|
|
make_counts = Counter(m.get("make", "Unknown") for m in machines)
|
|
|
|
# By Disney property
|
|
disney_machines = [m for m in machines if m.get("disney_park")]
|
|
disney_prop_counts = Counter(m["disney_park"] for m in disney_machines)
|
|
|
|
# Invalid / placeholder serials
|
|
invalid_serials = []
|
|
placeholder_serials = []
|
|
for m in machines:
|
|
mid = m.get("machine_id", "???")
|
|
if m.get("serial_is_placeholder"):
|
|
placeholder_serials.append(mid)
|
|
elif not m.get("serial_is_valid"):
|
|
invalid_serials.append(mid)
|
|
|
|
# OCR corrections applied
|
|
ocr_fixed = []
|
|
for m in machines:
|
|
mid = m.get("machine_id", "???")
|
|
raw_ser = raw_serials.get(mid)
|
|
cleaned = _clean_raw(raw_ser)
|
|
if cleaned:
|
|
corrected = _ocr_correct(cleaned)
|
|
if cleaned != corrected:
|
|
ocr_fixed.append((mid, cleaned, corrected))
|
|
|
|
# Unknown Type/Class/Make/Model
|
|
unknown_type = []
|
|
for m in machines:
|
|
mid = m.get("machine_id", "???")
|
|
cls = m.get("class", "")
|
|
make = m.get("make", "")
|
|
model = m.get("model", "")
|
|
if cls == "Unknown" or make == "Unknown" or model == "Unknown":
|
|
unknown_type.append((mid, cls, make, model))
|
|
|
|
# Machines lacking any GPS (no address, city, state, postal code)
|
|
no_gps = []
|
|
for m in machines:
|
|
mid = m.get("machine_id", "???")
|
|
addr = (m.get("address") or "").strip()
|
|
city = (m.get("location_area") or "").strip()
|
|
state = (m.get("state") or "").strip()
|
|
zipc = (m.get("postal_code") or "").strip()
|
|
if not addr and not city and not state and not zipc:
|
|
no_gps.append(mid)
|
|
|
|
# Priority distribution
|
|
priority_counts = Counter(m.get("priority", "Unknown") for m in machines)
|
|
|
|
# Alert severity breakdown
|
|
severity_counts = Counter()
|
|
for m in machines:
|
|
alerts = m.get("alerts", {})
|
|
if isinstance(alerts, dict):
|
|
sev = alerts.get("severity", "none")
|
|
else:
|
|
sev = "none"
|
|
severity_counts[sev] += 1
|
|
|
|
# Telemetry provider breakdown
|
|
telem_counts = Counter(m.get("telemetry_provider") or "None" for m in machines)
|
|
|
|
# ── Build Report ─────────────────────────────────────────────────────
|
|
lines = []
|
|
lines.append("# 🛠️ Validation Report — Canteen Seed Import")
|
|
lines.append("")
|
|
lines.append(f"**Generated:** {__import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
lines.append(f"**Source:** `{os.path.basename(filepath)}`")
|
|
lines.append(f"**Total Machines:** {total}")
|
|
lines.append("")
|
|
lines.append("---")
|
|
lines.append("")
|
|
|
|
# ── 1. Total by Class ────────────────────────────────────────────────
|
|
lines.append("## 1. Machine Count by Class")
|
|
lines.append("")
|
|
lines.append(f"| Class | Count |")
|
|
lines.append(f"|-------|-------|")
|
|
for cls in ["Snack", "Bev", "Food"]:
|
|
c = class_counts.get(cls, 0)
|
|
lines.append(f"| {cls} | {c} |")
|
|
for cls, c in class_counts.most_common():
|
|
if cls not in ("Snack", "Bev", "Food"):
|
|
lines.append(f"| {cls} | {c} |")
|
|
lines.append("")
|
|
lines.append(f"**Subtotal (Snack/Bev/Food):** {sum(class_counts.get(c, 0) for c in ['Snack','Bev','Food'])}")
|
|
lines.append("")
|
|
|
|
# ── 2. By Make ───────────────────────────────────────────────────────
|
|
lines.append("## 2. Machine Count by Make")
|
|
lines.append("")
|
|
lines.append(f"| Make | Count |")
|
|
lines.append(f"|------|-------|")
|
|
for make, c in make_counts.most_common():
|
|
lines.append(f"| {make} | {c} |")
|
|
lines.append("")
|
|
|
|
# ── 3. By Disney Property ────────────────────────────────────────────
|
|
lines.append("## 3. Machine Count by Disney Property")
|
|
lines.append("")
|
|
lines.append(f"**Total Disney machines:** {len(disney_machines)}")
|
|
lines.append("")
|
|
lines.append(f"| Disney Property | Count |")
|
|
lines.append(f"|-----------------|-------|")
|
|
for prop, c in disney_prop_counts.most_common():
|
|
lines.append(f"| {prop} | {c} |")
|
|
lines.append("")
|
|
|
|
# ── 4. Invalid / Placeholder Serials ─────────────────────────────────
|
|
lines.append("## 4. Invalid / Placeholder Serials")
|
|
lines.append("")
|
|
lines.append(f"**Placeholder serials:** {len(placeholder_serials)}")
|
|
if placeholder_serials:
|
|
lines.append("")
|
|
lines.append("Machine IDs with placeholder serials:")
|
|
for mid in sorted(placeholder_serials):
|
|
lines.append(f"- `{mid}`")
|
|
lines.append("")
|
|
lines.append(f"**Invalid serials:** {len(invalid_serials)}")
|
|
if invalid_serials:
|
|
lines.append("")
|
|
lines.append("Machine IDs with invalid serials:")
|
|
for mid in sorted(invalid_serials):
|
|
lines.append(f"- `{mid}`")
|
|
lines.append("")
|
|
|
|
# ── 5. OCR Corrections Applied ───────────────────────────────────────
|
|
lines.append("## 5. OCR Corrections Applied")
|
|
lines.append("")
|
|
lines.append(f"**Serials corrected:** {len(ocr_fixed)}")
|
|
if ocr_fixed:
|
|
lines.append("")
|
|
lines.append("| Machine ID | Raw (cleaned) | Corrected |")
|
|
lines.append("|------------|---------------|-----------|")
|
|
for mid, raw_c, corr in sorted(ocr_fixed, key=lambda x: x[0]):
|
|
lines.append(f"| `{mid}` | `{raw_c}` | `{corr}` |")
|
|
lines.append("")
|
|
|
|
# ── 6. Unknown Type/Class/Make/Model ─────────────────────────────────
|
|
lines.append("## 6. Unknown Type / Class / Make / Model")
|
|
lines.append("")
|
|
lines.append(f"**Machines with unknowns:** {len(unknown_type)}")
|
|
if unknown_type:
|
|
lines.append("")
|
|
lines.append("| Machine ID | Class | Make | Model |")
|
|
lines.append("|------------|-------|------|-------|")
|
|
for mid, cls, make, model in sorted(unknown_type, key=lambda x: x[0]):
|
|
lines.append(f"| `{mid}` | {cls} | {make} | {model} |")
|
|
lines.append("")
|
|
|
|
# ── 7. Machines Lacking Any GPS ──────────────────────────────────────
|
|
lines.append("## 7. Machines Lacking Any GPS / Location Data")
|
|
lines.append("")
|
|
lines.append(f"**Machines with no address data:** {len(no_gps)}")
|
|
if no_gps:
|
|
lines.append("")
|
|
lines.append("Machine IDs:")
|
|
for mid in sorted(no_gps):
|
|
lines.append(f"- `{mid}`")
|
|
lines.append("")
|
|
|
|
# ── 8. Priority Distribution ─────────────────────────────────────────
|
|
lines.append("## 8. Priority Distribution")
|
|
lines.append("")
|
|
lines.append(f"| Priority | Count |")
|
|
lines.append(f"|----------|-------|")
|
|
for pri in ["Low", "Mid", "High", "Unknown"]:
|
|
c = priority_counts.get(pri, 0)
|
|
lines.append(f"| {pri} | {c} |")
|
|
lines.append("")
|
|
|
|
# ── 9. Alert Severity Breakdown ──────────────────────────────────────
|
|
lines.append("## 9. Alert Severity Breakdown")
|
|
lines.append("")
|
|
lines.append(f"| Severity | Count |")
|
|
lines.append(f"|----------|-------|")
|
|
for sev in ["critical", "info", "none"]:
|
|
c = severity_counts.get(sev, 0)
|
|
lines.append(f"| {sev} | {c} |")
|
|
lines.append("")
|
|
|
|
# ── 10. Telemetry Provider Breakdown ─────────────────────────────────
|
|
lines.append("## 10. Telemetry Provider Breakdown")
|
|
lines.append("")
|
|
lines.append(f"| Provider | Count |")
|
|
lines.append(f"|----------|-------|")
|
|
for prov, c in telem_counts.most_common():
|
|
lines.append(f"| {prov} | {c} |")
|
|
lines.append("")
|
|
|
|
lines.append("---")
|
|
lines.append("")
|
|
lines.append("*Report generated automatically by `reporter.py`*")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ─── CLI Entry Point ────────────────────────────────────────────────────────
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print("Usage: python3 reporter.py seed-data/Machine_List.xlsx")
|
|
sys.exit(1)
|
|
|
|
filepath = sys.argv[1]
|
|
|
|
if not os.path.exists(filepath):
|
|
print(f"❌ File not found: {filepath}")
|
|
sys.exit(1)
|
|
|
|
print(f"📄 Loading {filepath}...")
|
|
report = generate_report(filepath)
|
|
|
|
out_dir = os.path.dirname(os.path.abspath(filepath))
|
|
out_path = os.path.join(out_dir, "validation_report.md")
|
|
with open(out_path, "w") as f:
|
|
f.write(report)
|
|
|
|
print(f"✅ Report written to {out_path}")
|
|
print(f" {len(report)} bytes")
|
|
|
|
# Print a quick summary to stdout
|
|
import re as _re
|
|
total_match = _re.search(r'\*\*Total Machines:\*\* (\d+)', report)
|
|
if total_match:
|
|
print(f"\n📊 Summary: {total_match.group(1)} machines analyzed.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|