ETL stages

get_access_token() →
  MSAL ConfidentialClientApplication(
    client_id=AZURE_CLIENT_ID,
    client_credential=AZURE_CLIENT_SECRET,
    authority=f"https://login.microsoftonline.com/{AZURE_TENANT_ID}"
  ).acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])

GET https://graph.microsoft.com/v1.0/users/{OUTLOOK_USER_EMAIL}/messages
  ?$filter=isRead eq false and contains(subject, 'Metaweave Forms')
  &$select=id,subject,body,receivedDateTime
  &$orderby=receivedDateTime asc
  &$top=50

SUBJECT_PATTERN = re.compile(
    r"Metaweave Forms:\s*(.+?)\s*-\s*(.+?)\s*-\s*(\d{2}\.\d{2}\.\d{4})"
)

@dataclass
class FetchedEmail:
    message_id: str
    subject: str
    body_text: str
    received_datetime: datetime
    vessel_name: str
    report_type_raw: str   # "Noon Report", "Arrival Notice", etc.
    report_date: date

re.search(r"Report:\s*(.+?)\s+(v[\d.]+)", body)
# ("Noon Report", "v00.01.02")

---------- BEGIN MW FORM DATA ---------------
<base64 ciphertext>
------------- END MW FORM DATA ----------------

MARKER_BEGIN = "BEGIN MW FORM DATA"
MARKER_END = "END MW FORM DATA"

def decrypt_payload(b64_ciphertext: str, key: str) -> dict:
    raw = base64.b64decode(b64_ciphertext)
    cipher = AES.new(key.encode(), AES.MODE_CBC, key.encode())  # IV = key (16 bytes)
    padded = cipher.decrypt(raw)
    plain = unpad(padded, AES.block_size)
    return json.loads(plain.decode())

@dataclass
class ParseResult:
    form_data: dict        # the decrypted (or fallback) payload
    report_type_raw: str   # from the header line
    form_version: str

{
    "vessel_info":        {"imo": ..., "name": ..., "code": ...},
    "voyage_number":      "V31",
    "report":             Report(...),                # 92 scalar fields
    "events":             [ReportEvent(...), ...],    # at-sea + in-port, with nested fuel breakdown
    "bunker_rob":         [ReportBunkerRob(...), ...],# per fuel type, per context
    "upcoming_ports":     [ReportUpcomingPort(...), ...],
    "fowe_periods":       [ReportFowePeriod(...), ...],
    "scrubber_breakdowns":[ReportScrubberBreakdown(...), ...],
    "bunker_deliveries":  [BunkerDelivery(...), ...],
    "bunker_biofuels":    [BunkerBiofuel(...), ...],
    "sof_activities":     [SofActivity(...), ...],
    "cargo_details":      [ReportCargo(...), ...],
    "month_end_bunker":   [MonthEndBunkerReport(...), ...],
    "berthing":           [BerthingDetails(...), ...],
}

propulsion · maneuver · generator · loaddischarge · deballast · igs ·
boiler · incinerator · cargoheating · tankcleaning · others · flushing

def write_report(session, mapped, email_message_id):
    vessel = upsert_vessel(session, mapped["vessel_info"])             # by imo_number
    voyage = upsert_voyage(session, vessel.vessel_id, mapped["voyage_number"])
    
    # Replacement: same (vessel, type, datetime) → delete old, insert new
    existing = session.query(Report).filter_by(
        vessel_id=vessel.vessel_id,
        report_type=report.report_type,
        report_datetime_utc=report.report_datetime_utc,
    ).one_or_none()
    if existing:
        session.delete(existing)   # CASCADE drops 11 child arrays
        session.flush()
    
    report = mapped["report"]
    report.vessel_id = vessel.vessel_id
    report.voyage_id = voyage.voyage_id
    report.email_message_id = email_message_id
    report.received_at = datetime.utcnow()
    session.add(report)
    session.flush()                # get report_id
    
    # Attach all 11 child arrays with report_id FK
    for events_list_name, items in mapped.items():
        if events_list_name in CHILD_KEYS:
            for item in items:
                item.report_id = report.report_id
                session.add(item)
    
    return report

ForeignKey("metaweave_report.report_id", ondelete="CASCADE")