#!/usr/bin/env python3 """Validate a TPNote portable trip plan using the app's schema invariants.""" from __future__ import annotations import argparse import json import math import sys import uuid from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo, ZoneInfoNotFoundError MAX_FILE_SIZE = 20 * 1024 * 1024 MAX_RECORD_COUNT = 100_000 CURRENT_SCHEMA_VERSION = 6 MINIMUM_SCHEMA_VERSION = 1 ENUMS = { "themeID": {"sakura", "ocean", "forest", "sunset", "graphite"}, "itemType": {"Activity", "Hotel", "Flight", "Train", "Food", "Note", "Payment", "Reminder", "Custom"}, "status": {"Idea", "Planned", "Booked", "Done", "Skipped"}, "timelineRole": {"Scheduled Event", "Background Context"}, "category": {"Hotel", "Food", "Activity", "Transit", "Shopping"}, "placeProvider": {"Apple Maps", "Google Places", "Manual"}, "provider": {"Apple Maps", "Google Places", "Manual"}, "kind": {"Note", "Booking", "Payment", "Reminder"}, "paymentStatus": {"Planned", "Paid", "Refunded", "Disputed"}, "detailType": {"Note", "Checklist", "Link", "Confirmation", "Payment", "Weather", "Attachment", "Custom Text"}, "deliveryStyle": {"Standard", "Time Sensitive"}, "transportMode": {"Public Transit", "Walking", "Driving", "Cycling", "Flight"}, } ROOT_ARRAYS = ("days", "items", "notes", "places", "bookings", "moneyEntries", "details") OPTIONAL_ROOT_ARRAYS = ("reminders", "legs") RECORD_ARRAYS = ROOT_ARRAYS + OPTIONAL_ROOT_ARRAYS REQUIRED = { "plan": {"id", "title", "destinationSummary", "ownerName", "createdAt", "updatedAt"}, "days": {"id", "date", "title", "timeZoneIdentifier", "sortIndex", "createdAt", "updatedAt"}, "items": { "id", "dayID", "parentItemID", "placeID", "bookingID", "moneyEntryID", "name", "subtitle", "address", "startTime", "endTime", "allDay", "latitude", "longitude", "itemType", "status", "category", "placeProvider", "providerPlaceID", "phoneNumber", "websiteURL", "details", "confirmation", "amount", "sortIndex", "orderRank", "createdAt", "updatedAt", }, "notes": {"id", "dayID", "parentItemID", "time", "title", "body", "kind", "sortIndex", "orderRank", "createdAt", "updatedAt"}, "places": {"id", "name", "address", "latitude", "longitude", "provider", "providerPlaceID", "phoneNumber", "websiteURL", "category", "notes", "createdAt", "updatedAt"}, "bookings": {"id", "itemID", "title", "provider", "confirmationNumber", "contactName", "phoneNumber", "email", "websiteURL", "startsAt", "endsAt", "notes", "createdAt", "updatedAt"}, "moneyEntries": {"id", "itemID", "title", "amount", "currencyCode", "paymentStatus", "transactionDate", "notes", "createdAt", "updatedAt"}, "details": {"id", "itemID", "detailType", "title", "body", "value", "orderRank", "createdAt", "updatedAt"}, "reminders": {"id", "itemID", "minutesBefore", "isEnabled", "deliveryStyle", "createdAt", "updatedAt"}, "legs": {"id", "dayID", "fromItemID", "toItemID", "transportMode", "createdAt", "updatedAt"}, } # Swift's synthesized Codable encoder omits nil optionals. Accept both an # omitted key and an explicit JSON null for portable files from any TPNote build. OMITTABLE_NULLABLE_FIELDS = { "items": {"dayID", "parentItemID", "placeID", "bookingID", "moneyEntryID"}, "notes": {"parentItemID"}, "bookings": {"itemID", "startsAt", "endsAt"}, "moneyEntries": {"itemID"}, "details": {"itemID"}, } DATE_FIELDS = { "plan": ("createdAt", "updatedAt"), "days": ("date", "createdAt", "updatedAt"), "items": ("startTime", "endTime", "createdAt", "updatedAt"), "notes": ("time", "createdAt", "updatedAt"), "places": ("createdAt", "updatedAt"), "bookings": ("startsAt", "endsAt", "createdAt", "updatedAt"), "moneyEntries": ("transactionDate", "createdAt", "updatedAt"), "details": ("createdAt", "updatedAt"), "reminders": ("createdAt", "updatedAt"), "legs": ("createdAt", "updatedAt"), } class Validation: def __init__(self) -> None: self.errors: list[str] = [] self.warnings: list[str] = [] def error(self, path: str, message: str) -> None: self.errors.append(f"{path}: {message}") def warning(self, path: str, message: str) -> None: self.warnings.append(f"{path}: {message}") def parse_date(value: object, path: str, result: Validation, *, nullable: bool = False) -> datetime | None: if value is None and nullable: return None if not isinstance(value, str): result.error(path, "must be an ISO 8601 string") return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: result.error(path, f"invalid ISO 8601 timestamp {value!r}") return None def parse_uuid(value: object, path: str, result: Validation, *, nullable: bool = False) -> str | None: if value is None and nullable: return None if not isinstance(value, str): result.error(path, "must be a UUID string") return None try: return str(uuid.UUID(value)).upper() except ValueError: result.error(path, f"invalid UUID {value!r}") return None def finite_number(value: object) -> bool: return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value) def require_fields(record: object, kind: str, path: str, result: Validation) -> dict | None: if not isinstance(record, dict): result.error(path, "must be an object") return None missing = sorted( REQUIRED[kind] - OMITTABLE_NULLABLE_FIELDS.get(kind, set()) - record.keys() ) if missing: result.error(path, "missing required fields: " + ", ".join(missing)) return record def validate_enum(record: dict, field: str, path: str, result: Validation, *, optional: bool = False) -> None: if field not in record and optional: return value = record.get(field) if value is None and optional: return if value not in ENUMS[field]: result.error(f"{path}.{field}", f"unsupported value {value!r}; expected one of {sorted(ENUMS[field])}") def check_reference(record: dict, field: str, targets: set[str], path: str, result: Validation, *, nullable: bool = False) -> str | None: value = parse_uuid(record.get(field), f"{path}.{field}", result, nullable=nullable) if value and value not in targets: result.error(f"{path}.{field}", f"references missing record {value}") return value def validate_coordinates(record: dict, path: str, result: Validation) -> None: latitude = record.get("latitude") longitude = record.get("longitude") if not finite_number(latitude) or not -90 <= latitude <= 90: result.error(f"{path}.latitude", "must be a finite number from -90 through 90") if not finite_number(longitude) or not -180 <= longitude <= 180: result.error(f"{path}.longitude", "must be a finite number from -180 through 180") def validate(path: Path, *, app_ready: bool = False) -> Validation: result = Validation() try: size = path.stat().st_size except OSError as error: result.error(str(path), str(error)) return result if size > MAX_FILE_SIZE: result.error("$", f"file is {size} bytes; maximum is {MAX_FILE_SIZE}") return result try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as error: result.error("$", f"cannot read UTF-8 JSON: {error}") return result if not isinstance(data, dict): result.error("$", "root must be an object") return result if data.get("format") != "tpnote-plan": result.error("$.format", "must equal 'tpnote-plan'") version = data.get("schemaVersion") if not isinstance(version, int) or isinstance(version, bool) or not MINIMUM_SCHEMA_VERSION <= version <= CURRENT_SCHEMA_VERSION: result.error("$.schemaVersion", f"must be an integer from {MINIMUM_SCHEMA_VERSION} through {CURRENT_SCHEMA_VERSION}") elif version < CURRENT_SCHEMA_VERSION: result.warning("$.schemaVersion", f"version {version} is importable, but new files should use {CURRENT_SCHEMA_VERSION}") parse_date(data.get("exportedAt"), "$.exportedAt", result) plan = require_fields(data.get("plan"), "plan", "$.plan", result) if plan: if not isinstance(plan.get("title"), str) or not plan["title"].strip(): result.error("$.plan.title", "must be a nonblank string") if "themeID" in plan and plan["themeID"] is not None: validate_enum(plan, "themeID", "$.plan", result, optional=True) for field in DATE_FIELDS["plan"]: parse_date(plan.get(field), f"$.plan.{field}", result) arrays: dict[str, list] = {} for name in ROOT_ARRAYS: value = data.get(name) if not isinstance(value, list): result.error(f"$.{name}", "must be an array") arrays[name] = [] else: arrays[name] = value for name in OPTIONAL_ROOT_ARRAYS: value = data.get(name, []) if value is None: value = [] if not isinstance(value, list): result.error(f"$.{name}", "must be an array when present") arrays[name] = [] else: arrays[name] = value record_count = 1 + sum(len(arrays[name]) for name in RECORD_ARRAYS) if record_count > MAX_RECORD_COUNT: result.error("$", f"contains {record_count} records; maximum is {MAX_RECORD_COUNT}") ids: dict[str, set[str]] = {name: set() for name in RECORD_ARRAYS} all_ids: dict[str, str] = {} if plan: plan_id = parse_uuid(plan.get("id"), "$.plan.id", result) if plan_id: all_ids[plan_id] = "$.plan.id" for kind in RECORD_ARRAYS: for index, raw_record in enumerate(arrays[kind]): record_path = f"$.{kind}[{index}]" record = require_fields(raw_record, kind, record_path, result) if not record: continue record_id = parse_uuid(record.get("id"), f"{record_path}.id", result) if record_id: if record_id in all_ids: result.error(f"{record_path}.id", f"duplicates {all_ids[record_id]}") else: all_ids[record_id] = f"{record_path}.id" ids[kind].add(record_id) for field in DATE_FIELDS[kind]: parse_date( record.get(field), f"{record_path}.{field}", result, nullable=kind == "bookings" and field in {"startsAt", "endsAt"}, ) for index, day in enumerate(arrays["days"]): if not isinstance(day, dict): continue timezone = day.get("timeZoneIdentifier") if not isinstance(timezone, str): result.error(f"$.days[{index}].timeZoneIdentifier", "must be an IANA time-zone string") continue try: ZoneInfo(timezone) except ZoneInfoNotFoundError: result.error(f"$.days[{index}].timeZoneIdentifier", f"invalid IANA time zone {timezone!r}") parent_by_item: dict[str, str | None] = {} item_day: dict[str, str | None] = {} places_by_id = { str(place.get("id", "")).lower(): place for place in arrays["places"] if isinstance(place, dict) } for index, item in enumerate(arrays["items"]): if not isinstance(item, dict): continue path_prefix = f"$.items[{index}]" item_id = parse_uuid(item.get("id"), f"{path_prefix}.id", Validation()) day_id = check_reference(item, "dayID", ids["days"], path_prefix, result, nullable=True) parent_id = check_reference(item, "parentItemID", ids["items"], path_prefix, result, nullable=True) check_reference(item, "placeID", ids["places"], path_prefix, result, nullable=True) place = places_by_id.get(str(item.get("placeID", "")).lower()) if place: for item_key, place_key in (("latitude", "latitude"), ("longitude", "longitude"), ("address", "address"), ("phoneNumber", "phoneNumber"), ("websiteURL", "websiteURL"), ("placeProvider", "provider"), ("providerPlaceID", "providerPlaceID"), ("googlePlaceID", "googlePlaceID")): if (item.get(item_key) or "") != (place.get(place_key) or ""): result.warning(f"{path_prefix}.{item_key}", "differs from the linked place; copy its canonical location/contact fields. " "Event category and title may differ, but location identity must agree.") subtitle = item.get("subtitle") if place and isinstance(subtitle, str) and subtitle.strip(): place_name = place.get("name") if isinstance(place_name, str) and place_name.strip() != subtitle.strip(): result.warning(f"{path_prefix}.subtitle", "does not match the linked place name; subtitle is a place-name fallback, " "not event or booking context. Preserve descriptive text in details when repairing.") check_reference(item, "bookingID", ids["bookings"], path_prefix, result, nullable=True) check_reference(item, "moneyEntryID", ids["moneyEntries"], path_prefix, result, nullable=True) validate_enum(item, "itemType", path_prefix, result) validate_enum(item, "status", path_prefix, result) validate_enum(item, "category", path_prefix, result) validate_enum(item, "placeProvider", path_prefix, result) validate_enum(item, "timelineRole", path_prefix, result, optional=True) validate_coordinates(item, path_prefix, result) if not finite_number(item.get("orderRank")): result.error(f"{path_prefix}.orderRank", "must be a finite number") start = parse_date(item.get("startTime"), f"{path_prefix}.startTime", Validation()) end = parse_date(item.get("endTime"), f"{path_prefix}.endTime", Validation()) if start and end and end < start: result.error(path_prefix, "endTime cannot be earlier than startTime") if item_id: parent_by_item[item_id] = parent_id item_day[item_id] = day_id for index, place in enumerate(arrays["places"]): if not isinstance(place, dict): continue path_prefix = f"$.places[{index}]" validate_enum(place, "provider", path_prefix, result) validate_enum(place, "category", path_prefix, result) validate_coordinates(place, path_prefix, result) for index, note in enumerate(arrays["notes"]): if not isinstance(note, dict): continue path_prefix = f"$.notes[{index}]" check_reference(note, "dayID", ids["days"], path_prefix, result) check_reference(note, "parentItemID", ids["items"], path_prefix, result, nullable=True) validate_enum(note, "kind", path_prefix, result) if not finite_number(note.get("orderRank")): result.error(f"{path_prefix}.orderRank", "must be a finite number") for kind in ("bookings", "moneyEntries", "details"): for index, record in enumerate(arrays[kind]): if not isinstance(record, dict): continue path_prefix = f"$.{kind}[{index}]" check_reference(record, "itemID", ids["items"], path_prefix, result, nullable=True) if kind == "moneyEntries": validate_enum(record, "paymentStatus", path_prefix, result) if kind == "details": validate_enum(record, "detailType", path_prefix, result) if not finite_number(record.get("orderRank")): result.error(f"{path_prefix}.orderRank", "must be a finite number") for index, reminder in enumerate(arrays["reminders"]): if not isinstance(reminder, dict): continue path_prefix = f"$.reminders[{index}]" check_reference(reminder, "itemID", ids["items"], path_prefix, result) validate_enum(reminder, "deliveryStyle", path_prefix, result) minutes = reminder.get("minutesBefore") if not isinstance(minutes, int) or isinstance(minutes, bool) or minutes < 0: result.error(f"{path_prefix}.minutesBefore", "must be a nonnegative integer") for index, leg in enumerate(arrays["legs"]): if not isinstance(leg, dict): continue path_prefix = f"$.legs[{index}]" day_id = check_reference(leg, "dayID", ids["days"], path_prefix, result) from_id = check_reference(leg, "fromItemID", ids["items"], path_prefix, result) to_id = check_reference(leg, "toItemID", ids["items"], path_prefix, result) validate_enum(leg, "transportMode", path_prefix, result) if from_id and to_id and from_id == to_id: result.warning(path_prefix, "fromItemID and toItemID are the same") for endpoint, item_id in (("fromItemID", from_id), ("toItemID", to_id)): if day_id and item_id and item_day.get(item_id) not in {None, day_id}: result.warning(f"{path_prefix}.{endpoint}", "item belongs to a different day") for item_id in parent_by_item: visited: set[str] = set() current: str | None = item_id while current: if current in visited: result.error(all_ids.get(item_id, "$.items"), "parentItemID chain contains a cycle") break visited.add(current) current = parent_by_item.get(current) if app_ready: audit_app_ready(data, arrays, result) return result def audit_app_ready(data: dict, arrays: dict[str, list], result: Validation) -> None: """Apply the stronger quality contract used for finished plugin output.""" places = { str(place.get("id", "")).upper(): place for place in arrays["places"] if isinstance(place, dict) } items = { str(item.get("id", "")).upper(): item for item in arrays["items"] if isinstance(item, dict) } bookings = { str(booking.get("id", "")).upper(): booking for booking in arrays["bookings"] if isinstance(booking, dict) } for index, place in enumerate(arrays["places"]): if not isinstance(place, dict): continue path_prefix = f"$.places[{index}]" if not isinstance(place.get("name"), str) or not place["name"].strip(): result.error(f"{path_prefix}.name", "app-ready places need a recognizable nonblank name") if not isinstance(place.get("address"), str) or not place["address"].strip(): result.error(f"{path_prefix}.address", "app-ready places need a complete readable address") if not str(place.get("providerPlaceID") or "").strip() and not str(place.get("googlePlaceID") or "").strip(): result.warning( path_prefix, "has no trusted provider identity; keep verified name/address/coordinates and disclose that " "TPNote will attempt Google Places enrichment after import", ) for index, item in enumerate(arrays["items"]): if not isinstance(item, dict): continue path_prefix = f"$.items[{index}]" is_scheduled = item.get("timelineRole", "Scheduled Event") == "Scheduled Event" needs_place = item.get("itemType") in {"Activity", "Hotel", "Flight", "Train", "Food"} place_id = str(item.get("placeID") or "").upper() if is_scheduled and needs_place and not place_id: result.error( f"{path_prefix}.placeID", "app-ready in-person events must link to a canonical place", ) if place_id and place_id in places: place = places[place_id] if not str(item.get("address") or "").strip(): result.error(f"{path_prefix}.address", "linked event needs the canonical place address snapshot") if not str(item.get("subtitle") or "").strip(): result.error(f"{path_prefix}.subtitle", "linked event needs the canonical place name as its subtitle") booking_id = str(item.get("bookingID") or "").upper() if booking_id and booking_id in bookings: linked_item_id = str(bookings[booking_id].get("itemID") or "").upper() if linked_item_id != str(item.get("id") or "").upper(): result.error( f"{path_prefix}.bookingID", "booking relationship must be reciprocal through bookings[].itemID", ) if item.get("status") == "Booked" and not booking_id: result.warning( f"{path_prefix}.bookingID", "booked event has no booking record; store user-provided reservation facts in bookings, not event prose", ) for index, booking in enumerate(arrays["bookings"]): if not isinstance(booking, dict): continue path_prefix = f"$.bookings[{index}]" item_id = str(booking.get("itemID") or "").upper() if item_id and item_id in items: linked_booking_id = str(items[item_id].get("bookingID") or "").upper() if linked_booking_id != str(booking.get("id") or "").upper(): result.error( f"{path_prefix}.itemID", "booking relationship must be reciprocal through items[].bookingID", ) leg_pairs = { (str(leg.get("fromItemID") or "").upper(), str(leg.get("toItemID") or "").upper()) for leg in arrays["legs"] if isinstance(leg, dict) } for day in arrays["days"]: if not isinstance(day, dict): continue day_id = str(day.get("id") or "").upper() scheduled = [ item for item in arrays["items"] if isinstance(item, dict) and str(item.get("dayID") or "").upper() == day_id and item.get("timelineRole", "Scheduled Event") == "Scheduled Event" ] scheduled.sort(key=lambda item: ( str(item.get("startTime") or ""), float(item.get("orderRank") or 0), int(item.get("sortIndex") or 0), )) for earlier, later in zip(scheduled, scheduled[1:]): earlier_place = str(earlier.get("placeID") or "").upper() later_place = str(later.get("placeID") or "").upper() if not earlier_place or not later_place or earlier_place == later_place: continue pair = ( str(earlier.get("id") or "").upper(), str(later.get("id") or "").upper(), ) if pair not in leg_pairs: result.error( "$.legs", f"missing commute leg from {earlier.get('name')!r} to {later.get('name')!r}", ) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("files", nargs="+", type=Path, help=".tpnote files to validate") parser.add_argument( "--app-ready", action="store_true", help="also require polished places, reciprocal bookings, and commute coverage", ) args = parser.parse_args() failed = False for path in args.files: result = validate(path, app_ready=args.app_ready) for warning in result.warnings: print(f"WARNING {warning}") for error in result.errors: print(f"ERROR {error}") if result.errors: failed = True label = "APP-READY INVALID" if args.app_ready else "INVALID" print(f"{label} {path} ({len(result.errors)} error(s), {len(result.warnings)} warning(s))") else: label = "APP-READY" if args.app_ready else "VALID" print(f"{label} {path} ({len(result.warnings)} warning(s))") return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())