bsncubed
408fde440d
Photograph/screenshot a boarding pass, decode its BCBP barcode (PDF417/ Aztec/QR/DataMatrix), fetch the historical flight track from FlightAware the day after the flight, stage it for review, and write it into AirTrail via its REST API on approval. ntfy notifications carry signed Approve/Deny/Investigate actions.
105 lines
3.5 KiB
Python
105 lines
3.5 KiB
Python
"""GET/POST /upload - photograph or pick a boarding pass image, decode it,
|
|
and queue it for the daily FlightAware job."""
|
|
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from flask import Blueprint, redirect, render_template, request, url_for
|
|
|
|
import bcbp
|
|
import config
|
|
import decode
|
|
import models
|
|
import reference_data
|
|
|
|
bp = Blueprint("upload", __name__)
|
|
|
|
|
|
@bp.route("/upload", methods=["GET", "POST"])
|
|
def upload():
|
|
if request.method == "GET":
|
|
return render_template("upload.html")
|
|
|
|
file = request.files.get("photo")
|
|
if not file or not file.filename:
|
|
return render_template("upload.html", error="No photo selected."), 400
|
|
|
|
image_bytes = file.read()
|
|
upload_dt = datetime.now(timezone.utc)
|
|
|
|
image_name = f"{uuid.uuid4().hex}.jpg"
|
|
image_path = config.UPLOADS_DIR / image_name
|
|
config.ensure_dirs()
|
|
with open(image_path, "wb") as f:
|
|
f.write(image_bytes)
|
|
|
|
raw_bcbp = decode.decode_boarding_pass(image_bytes)
|
|
|
|
if not raw_bcbp:
|
|
flight_id = models.create_pending_flight(
|
|
image_path=image_name,
|
|
status="decode_failed",
|
|
error_message="Could not detect a PDF417 barcode in the photo.",
|
|
)
|
|
return redirect(url_for("review.detail", flight_id=flight_id))
|
|
|
|
try:
|
|
parsed = bcbp.parse_bcbp(raw_bcbp, upload_dt=upload_dt)
|
|
except bcbp.BCBPParseError as e:
|
|
flight_id = models.create_pending_flight(
|
|
image_path=image_name,
|
|
raw_bcbp=raw_bcbp,
|
|
status="decode_failed",
|
|
error_message=str(e),
|
|
)
|
|
return redirect(url_for("review.detail", flight_id=flight_id))
|
|
|
|
if parsed.julian_day is None:
|
|
flight_id = models.create_pending_flight(
|
|
image_path=image_name,
|
|
raw_bcbp=raw_bcbp,
|
|
passenger_name=parsed.passenger_name,
|
|
pnr=parsed.pnr,
|
|
from_iata=parsed.from_iata,
|
|
to_iata=parsed.to_iata,
|
|
operating_carrier_iata=parsed.operating_carrier,
|
|
flight_number=parsed.flight_number,
|
|
compartment=parsed.compartment,
|
|
seat=parsed.seat,
|
|
parse_warnings=parsed.warnings,
|
|
status="decode_failed",
|
|
error_message="Could not determine the flight date from the barcode.",
|
|
)
|
|
return redirect(url_for("review.detail", flight_id=flight_id))
|
|
|
|
flight_date = bcbp.resolve_flight_date(parsed.julian_day, upload_dt)
|
|
process_after = datetime.combine(
|
|
flight_date + timedelta(days=1), datetime.min.time(), tzinfo=timezone.utc
|
|
).isoformat()
|
|
|
|
carrier_icao = reference_data.iata_to_icao_airline(parsed.operating_carrier)
|
|
from_icao = reference_data.iata_to_icao_airport(parsed.from_iata)
|
|
to_icao = reference_data.iata_to_icao_airport(parsed.to_iata)
|
|
|
|
flight_id = models.create_pending_flight(
|
|
image_path=image_name,
|
|
raw_bcbp=raw_bcbp,
|
|
passenger_name=parsed.passenger_name,
|
|
pnr=parsed.pnr,
|
|
from_iata=parsed.from_iata,
|
|
to_iata=parsed.to_iata,
|
|
from_icao=from_icao,
|
|
to_icao=to_icao,
|
|
operating_carrier_iata=parsed.operating_carrier,
|
|
operating_carrier_icao=carrier_icao,
|
|
flight_number=parsed.flight_number,
|
|
julian_day=parsed.julian_day,
|
|
flight_date=flight_date.isoformat(),
|
|
compartment=parsed.compartment,
|
|
seat=parsed.seat,
|
|
process_after=process_after,
|
|
parse_warnings=parsed.warnings,
|
|
status="queued",
|
|
)
|
|
return redirect(url_for("review.detail", flight_id=flight_id))
|