#!/usr/bin/env python3 """Convert selectable NetLine/Crew PDF rosters to UTC ICS, JSON and CSV. Python 3.10+; pip install pdfplumber python roster_pdf.py roster.pdf --out schedule [--own-code DEB] Unknown duty codes become generic events, timed if times are printed, otherwise all-day. Descriptions contain only explicitly listed crew. """ import argparse import calendar import csv import hashlib import json import re import warnings from collections import Counter from datetime import date, datetime, timedelta, timezone from pathlib import Path import pdfplumber MONTHS = {name.lower(): i for i, name in enumerate(calendar.month_abbr) if name} DAY = re.compile(r'^(Mon|Tue|Wed|Thu|Fri|Sat|Sun)(\d{2})\b') FLIGHT = re.compile( r'^(?PDH/)?(?P[A-Z0-9]{2})\s+(?P\d{1,4})\s*R?\s+' r'(?P[A-Z]{3})\s+(?P\d{4})\s+(?P\d{4})\s+' r'(?P[A-Z]{3})(?:\s+(?P[A-Z0-9]{3}))?\s*$' ) DUTY = re.compile( r'^(?P[A-Za-z0-9][A-Za-z0-9_/?+.-]*)\s+(?:R\s+)?' r'(?P[A-Z]{3})' r'(?:\s+(?P\d{4})\s+(?P\d{4}))?' r'(?:\s+(?P[A-Z]{3}))?\s*$' ) # Do not guess the expansion of company-specific codes. LABELS = { 'F': '🏠 Free day off', 'FCT': 'FCT', 'NC': 'NC', 'SF': 'SF', 'CAR': '🚗 Ground transport', 'SIM-737': '📚 Simulator 737', 'LPCB': '📚 Licence proficiency check B737', 'OPCB': '📚 Operator proficiency check B737', 'MED1': 'Medical', 'SFP1': '📚 Safety recurrent pilots Y1', 'SEC1': '📚 Security Y1', 'V31': '🏠 31st day of vacation', 'RES': 'Reserve', } TRAINING = {'SIM-737', 'LPCB', 'OPCB', 'SFP1', 'SEC1'} def pdf_date(value): match = re.fullmatch(r'(\d{2})([A-Za-z]{3})(\d{2})', value) if not match: raise ValueError(f'Unsupported date: {value}') d, m, y = match.groups() return date(2000 + int(y), MONTHS[m.lower()], int(d)) def grouped_words(words): groups = [] for word in sorted(words, key=lambda w: (w['top'], w['x0'])): if not groups or abs(word['top'] - groups[-1][0]['top']) > 1: groups.append([]) groups[-1].append(word) return [sorted(g, key=lambda w: w['x0']) for g in groups] def rows(words): return [' '.join(w['text'] for w in g) for g in grouped_words(words)] def table_columns(page): words = page.extract_words(x_tolerance=1, y_tolerance=1) dates = [w for w in words if w['text'] == 'date'] if not dates: return [] y = min(w['top'] for w in dates) headings = sorted([w for w in dates if abs(w['top'] - y) < 1], key=lambda w: w['x0']) if len(headings) != 3: raise ValueError('Unsupported layout: expected three table columns.') result = [] for i, h in enumerate(headings): left = h['x0'] - 1 right = headings[i + 1]['x0'] - 1 if i < 2 else page.width header = [w for w in words if left <= w['x0'] < right and abs(w['top'] - y) < 1] required = {} for name in ('H', 'R', 'info'): found = [w for w in header if w['text'] == name] if len(found) != 1: raise ValueError(f'Unsupported table header: {name}') required[name] = found[0] # The H field is left of every duty token; R starts the next field. duty_left = required['H']['x1'] + 1 duty_right = required['R']['x0'] - 1 content = [w for w in words if left <= w['x0'] < right and w['top'] > h['bottom']] result.append((grouped_words(content), duty_left, duty_right, required['info']['x0'] - 1)) return result def classify(code): if code.startswith('SB'): return 'standby', '🛰 On standby' if code.startswith('VAC'): return 'leave', '🏠 Leave' if code.startswith('CRM'): return 'training', '📚 Crew Resource Management' if code in TRAINING: return 'training', LABELS[code] if code in LABELS: return 'duty', LABELS[code] return 'generic', code def parse_duty(line): """Parse only text already confirmed to be in the duty field.""" m = FLIGHT.fullmatch(line) if m: data = m.groupdict() return dict(code=('DH/' if data['dh'] else '') + data['airline'], flight=data['airline'] + data['number'], origin=data['origin'], destination=data['destination'], start=data['start'], end=data['end'], aircraft=data['aircraft'] or '', deadhead=bool(data['dh'])) m = DUTY.fullmatch(line) if not m: no_location = re.fullmatch(r'([A-Za-z0-9][A-Za-z0-9_/?+.-]*)\s+(\d{4})\s+(\d{4})', line) if no_location: code, begin, end = no_location.groups() return dict(code=code, flight='', origin='', destination='', start=begin, end=end, aircraft='', deadhead=False) # A malformed flight must never silently become an all-day duty. if re.match(r'^(?:DH/)?[A-Z0-9]{2}\s+\d{1,4}(?:\s|$)', line): raise ValueError(f'Malformed flight: {line}') # Some future all-day duties may have no location at all. if re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_/?+.-]*', line): return dict(code=line, flight='', origin='', destination='', start=None, end=None, aircraft='', deadhead=False) raise ValueError(f'Unparsed duty row (not silently skipped): {line}') d = m.groupdict() return dict(code=d['code'], flight='', origin=d['origin'], destination=d['destination'] or '', start=d['start'], end=d['end'], aircraft='', deadhead=False) def timed_bounds(d, start, end, previous_end=None): def at(value): if not re.fullmatch(r'\d{4}', value): raise ValueError(f'Invalid clock time: {value}') return datetime.combine(d, datetime.strptime(value, '%H%M').time(), timezone.utc) a, b = at(start), at(end) while previous_end is not None and a < previous_end: a += timedelta(days=1) b = datetime.combine(a.date(), b.time(), timezone.utc) if b < a: b += timedelta(days=1) if a == b: raise ValueError('Zero-length timed duty; check the source PDF.') return a, b def attach_crew(pdf, events, own_code): """Align the chronological crew section to exact roster flight legs. Dates in the PDF omit the month. Forward/backward matching must agree, otherwise fail rather than attach crew to an ambiguous month. """ records = [] active = done = False current = None for page in pdf.pages: if done: break for groups, _, _, _ in table_columns(page): for g in groups: text = ' '.join(w['text'] for w in g) if 'Crew Information on Leg' in text: active, current = True, None continue if not active: continue if 'Recurrent Training' in text: done = True break day = DAY.match(text) if day: m = FLIGHT.fullmatch(text[day.end():].strip()) if not m: raise ValueError(f'Unparsed crew leg: {text}') current = dict(weekday=day[1], day=int(day[2]), leg=m.groupdict(), crew_cockpit='', crew_cabin='') records.append(current) elif current is not None: role = re.match(r'^(cockpit|cabin):\s*(.*)$', text) if role: codes = re.findall(r'\b[A-Z]{3}\b', role[2]) current['crew_' + role[1]] = ', '.join(c for c in codes if c != own_code) if done: break flight_events = [e for e in events if e['kind'] == 'flight'] options = [] for record in records: m = record['leg'] matches = [] for i, event in enumerate(flight_events): d = date.fromisoformat(event['date']) if (d.day == record['day'] and calendar.day_abbr[d.weekday()] == record['weekday'] and event['flight'] == m['airline'] + m['number'] and event['origin'] == m['origin'] and event['destination'] == m['destination'] and event['departure_utc'][11:16].replace(':', '') == m['start'] and event['arrival_utc'][11:16].replace(':', '') == m['end']): matches.append(i) if not matches: raise ValueError(f'Crew entry has no matching roster flight: {record}') options.append(matches) earliest, previous = [], -1 for matches in options: choices = [i for i in matches if i > previous] if not choices: raise ValueError('Crew section is not in chronological flight order.') previous = min(choices) earliest.append(previous) latest, following = [], len(flight_events) for matches in reversed(options): choices = [i for i in matches if i < following] if not choices: raise ValueError('Cannot align crew section in chronological order.') following = max(choices) latest.append(following) if earliest != list(reversed(latest)): raise ValueError('Crew dates are ambiguous across months; refusing to guess.') for index, record in zip(earliest, records): flight_events[index].update(crew_cockpit=record['crew_cockpit'], crew_cabin=record['crew_cabin']) def extract_schedule(path, own_code=''): events, seen = [], set() with pdfplumber.open(path) as pdf: first_text = pdf.pages[0].extract_text() or '' period = re.search(r'Period:\s*(\d{2}[A-Za-z]{3}\d{2})\s*-\s*(\d{2}[A-Za-z]{3}\d{2})', first_text) if not period: raise ValueError('No roster period found; unsupported PDF or scanned document.') first, last = map(pdf_date, period.groups()) if last < first: raise ValueError('Invalid roster period.') dates, d = [], first while d <= last: dates.append(d) d += timedelta(days=1) current_date = None previous_end = None roster_done = False for page_no, page in enumerate(pdf.pages, 1): if roster_done: break for groups, duty_left, duty_right, info_left in table_columns(page): for g in groups: text = ' '.join(w['text'] for w in g) if 'Crew Information on Leg' in text: roster_done = True break if re.match(r'^(?:Flight time|Vacation|Sickness|Sum|Fict\.)\b', text): # Totals terminate the detailed roster (before the crew appendix). roster_done = True break day = DAY.match(text) if day: choices = [d for d in dates if d.day == int(day[2]) and calendar.day_abbr[d.weekday()] == day[1] and (current_date is None or d >= current_date)] if not choices: raise ValueError(f'Date outside period or not chronological on page {page_no}: {text}') resolved = min(choices) if resolved != current_date: previous_end = None current_date = resolved duty_words = [w for w in g if duty_left <= w['x0'] < duty_right] if not duty_words: continue # Hotel H?, notes, totals and other non-duty fields. start_x = duty_words[0]['x0'] duty_text = ' '.join(w['text'] for w in g if start_x <= w['x0'] < info_left) data = parse_duty(duty_text) if current_date is None: raise ValueError(f'Duty without date on page {page_no}: {text}') kind, _ = classify(data['code']) if data['flight']: kind = 'flight' has_times = data['start'] is not None a = b = None if has_times: a, b = timed_bounds(current_date, data['start'], data['end'], previous_end) previous_end = b item = dict(flight=data['flight'], origin=data['origin'], destination=data['destination'], departure_utc=a.isoformat() if a else '', arrival_utc=b.isoformat() if b else '', deadhead=data['deadhead'], aircraft=data['aircraft'], source_page=page_no, date=a.date().isoformat() if a else current_date.isoformat(), duty_code=data['code'], kind=kind, all_day=not has_times, crew_cockpit='', crew_cabin='') key = tuple(item[k] for k in ('date', 'duty_code', 'flight', 'origin', 'destination', 'departure_utc', 'arrival_utc')) if key in seen: raise ValueError(f'Duplicate roster event: {key}') seen.add(key) events.append(item) if roster_done: break if not events: raise ValueError('No duties found; no output written.') events.sort(key=lambda f: (f['date'], f['departure_utc'])) attach_crew(pdf, events, own_code) unknown = sorted({e['duty_code'] for e in events if e['kind'] == 'generic'}) if unknown: warnings.warn('New duty codes included as generic events: ' + ', '.join(unknown)) return events def extract_flights(path, own_code=''): return [e for e in extract_schedule(path, own_code) if e['kind'] == 'flight'] def ics_text(value): return value.replace('\\', '\\\\').replace('\n', '\\n').replace(';', '\\;').replace(',', '\\,') def fold(line): result, part = [], '' for char in line: if len((part + char).encode('utf-8')) > 75: result.append(part) part = ' ' part += char return '\r\n'.join(result + [part]) def make_ics(events): lines = ['BEGIN:VCALENDAR', 'VERSION:2.0', 'PRODID:-//NetLine PDF Schedule//EN', 'CALSCALE:GREGORIAN', 'X-WR-CALNAME:Work schedule'] stamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ') identities = Counter() for e in events: # Keep legacy flight UIDs; distinct repeated duties get an occurrence suffix. identity = '|'.join([e['flight'] or e['duty_code'], e['date'], e['origin'], e['destination']]) identities[identity] += 1 if identities[identity] > 1: identity += '|occurrence=' + str(identities[identity]) uid = hashlib.sha256(identity.encode()).hexdigest()[:32] + '@netline-pdf' if e['kind'] == 'flight': summary = ('➡️ (DH)' if e['deadhead'] else '✈️') + f" {e['origin']} - {e['destination']} {e['flight']}" else: _, label = classify(e['duty_code']) summary = label if label == e['duty_code'] else label + ' - ' + e['duty_code'] if e['destination']: summary += f" {e['origin']} - {e['destination']}" description = '\n'.join(label + ': ' + e[field] for label, field in [('Cockpit', 'crew_cockpit'), ('Cabin', 'crew_cabin')] if e.get(field)) def utc(value): parsed = datetime.fromisoformat(value) if parsed.tzinfo is None: raise ValueError('Timed events must have an explicit time zone.') return parsed.astimezone(timezone.utc).strftime('%Y%m%dT%H%M%SZ') lines += ['BEGIN:VEVENT', 'UID:' + uid, 'DTSTAMP:' + stamp] if e['all_day']: start = date.fromisoformat(e['date']) lines += ['DTSTART;VALUE=DATE:' + start.strftime('%Y%m%d'), 'DTEND;VALUE=DATE:' + (start + timedelta(days=1)).strftime('%Y%m%d'), 'X-MICROSOFT-CDO-ALLDAYEVENT:TRUE'] # Leave availability unspecified for company codes whose meaning is unknown. if e['duty_code'] == 'F' or e['kind'] == 'leave' or e['duty_code'] == 'V31': lines.append('TRANSP:TRANSPARENT') else: lines += ['DTSTART:' + utc(e['departure_utc']), 'DTEND:' + utc(e['arrival_utc'])] lines += ['SUMMARY:' + ics_text(summary), 'LOCATION:' + (ics_text(e['origin']) if e['kind'] != 'flight' else '')] if description: lines.append('DESCRIPTION:' + ics_text(description)) lines.append('END:VEVENT') return '\r\n'.join(fold(line) for line in lines + ['END:VCALENDAR']) + '\r\n' def write_outputs(events, prefix): prefix = Path(prefix) prefix.parent.mkdir(parents=True, exist_ok=True) Path(str(prefix) + '.json').write_text(json.dumps(events, indent=2) + '\n', encoding='utf-8') with Path(str(prefix) + '.csv').open('w', encoding='utf-8', newline='') as stream: writer = csv.DictWriter(stream, fieldnames=list(events[0])) writer.writeheader() writer.writerows(events) Path(str(prefix) + '.ics').write_bytes(make_ics(events).encode('utf-8')) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('pdf', type=Path) parser.add_argument('--out', type=Path, default=Path('schedule'), help='Output filename prefix') parser.add_argument('--own-code', default='', help='Optional crew code to omit (none by default)') args = parser.parse_args() events = extract_schedule(args.pdf, args.own_code) write_outputs(events, args.out) print(f'Extracted {len(events)} events: {dict(Counter(e["kind"] for e in events))}') print('Duty codes:', dict(sorted(Counter(e['duty_code'] for e in events).items()))) print(f'Wrote {args.out}.json, {args.out}.csv and {args.out}.ics') if __name__ == '__main__': main()