#!/usr/bin/env python3 """Deterministic extraction of domestic aviation route and fleet data. Sources: - NCAA PPTX summaries (2022 annual, Q1 2023) - THISDAY 2023 domestic airline passengers article - BASL MMA2 busiest routes article - BusinessDay lucrative routes / plane shortage article - TBI domestic aircraft grounded / fleet types article """ import csv import re from datetime import date from pathlib import Path from pptx import Presentation BASE = Path("C:/Users/HP/palmgrove/NEFDB") OUT_DIR = BASE / "datasets/transport_aviation/derived" OUT_DIR.mkdir(parents=True, exist_ok=True) TODAY = date.today().isoformat() COLUMNS = [ "route", "aircraft_type", "value", "unit", "data_year", "source_file", "source_publisher", "source_year", "source_url_or_path", "page_or_slide_or_article", "extraction_method", "extraction_date", "methodology_note", "confidence_flag", ] CITY_IATA = { "Lagos": "LOS", "Abuja": "ABV", "Port Harcourt": "PHC", "Owerri": "QOW", "Kano": "KAN", "Ilorin": "ILR", "Akure": "AKR", "Anambra": "ANA", "Asaba": "ABB", "Ibadan": "IBA", "Calabar": "CBQ", "Enugu": "ENU", } DOMESTIC_AIRLINES = { "aero contractors", "arik air", "azman air", "dana air", "overland", "air peace", "max air", "ibom", "ibom air", "united airlines", "united airline", "united nigeria airlines", "green africa", "greenafrica", "value jet", "valuejet", "rano air", "ng eagle", "overland airways", } def clean_num(raw): if raw is None: return None s = str(raw).strip().replace(",", "").replace(" ", "") if s in ("-", "", "NOOPERATION", "NO OPERATION"): return None try: return int(s) except ValueError: try: return float(s) except ValueError: return None def norm_route(text): """Convert city-pair strings like 'Lagos–Abuja' to LOS-ABV, or None.""" if not text: return None text = text.strip() parts = re.split(r"\s*(?:to|–|—|-)\s*", text, flags=re.IGNORECASE) if len(parts) != 2: return None a = parts[0].strip().rstrip(".,;()").strip() b = parts[1].strip().rstrip(".,;()").strip() ia = CITY_IATA.get(a) ib = CITY_IATA.get(b) if ia and ib: return f"{ia}-{ib}" return None def route_pairs_from_origin_list(origin, destinations): """Generate IATA route pairs where the first token is an origin city and remaining tokens are destinations.""" routes = [] first_pair = norm_route(origin) if first_pair: routes.append(first_pair) iata_origin = first_pair.split("-")[0] else: iata_origin = CITY_IATA.get(origin.strip().rstrip(".,;")) if iata_origin and destinations: first_dst = destinations[0].strip().rstrip(".,;") iata_dst = CITY_IATA.get(first_dst) if iata_dst: routes.append(f"{iata_origin}-{iata_dst}") if not iata_origin: return routes for dst in destinations: dst_clean = dst.strip().rstrip(".,;") route = norm_route(dst_clean) if route: if route not in routes: routes.append(route) continue iata_dst = CITY_IATA.get(dst_clean) if iata_dst: candidate = f"{iata_origin}-{iata_dst}" if candidate not in routes: routes.append(candidate) return routes def add_row(rows, **kwargs): rows.append({col: kwargs.get(col, "") for col in COLUMNS}) # --------------------------------------------------------------------------- # PPTX extraction # --------------------------------------------------------------------------- def is_domestic_airline(name): """Return True if the supplied airline name matches a known Nigerian domestic carrier.""" key = name.strip().lower().replace(" ", " ") return key in DOMESTIC_AIRLINES def table_has_domestic_airline(cells): """Return True if any airline cell matches a known domestic carrier.""" for r in cells: for cell in r: if is_domestic_airline(cell): return True return False def extract_pptx(path, document, publisher, source_year): rows = [] failures = [] prs = Presentation(str(path)) for slide_idx, slide in enumerate(prs.slides, start=1): tbl_idx = 0 for shape in slide.shapes: if not shape.has_table: continue tbl_idx += 1 table = shape.table cells = [] for row in table.rows: cells.append([c.text.replace("\n", " ").strip() for c in row.cells]) if not cells: continue header = [c.upper() for c in cells[0]] slide_ref = f"slide {slide_idx}, table {tbl_idx}" has_months = any(name.upper() in " ".join(header) for name in ("JANUARY", "FEBRUARY", "MARCH")) # --- DOMESTIC aggregate row (summary slides) --- for r in cells: if not r: continue first = r[0].strip().upper() if first == "DOMESTIC AIRLINES" or first.startswith("DOMESTIC"): for direction, needle in (("in-bound", "IN-BOU"), ("out-bound", "OUT-BOU")): for h, v in zip(header, r): if needle in h and clean_num(v) is not None: add_row( rows, value=clean_num(v), unit="passengers", data_year=source_year, source_file=path.name, source_document=document, source_publisher=publisher, source_year=source_year, source_url_or_path=str(path), page_or_slide_or_article=slide_ref, extraction_method="python-pptx table cell extraction", extraction_date=TODAY, methodology_note=f"Domestic-airline aggregate {direction} passengers from summary table.", confidence_flag="high_aggregate", ) break # Skip tables with no domestic carriers and skip monthly break-down tables. if not table_has_domestic_airline(cells) or has_months: continue # --- Airline-level summary rows (domestic carriers only) --- has_airlines = any("AIRLINES" in h for h in header) if has_airlines: airline_col = next((i for i, h in enumerate(header) if "AIRLINES" in h), None) in_col = next( (i for i, h in enumerate(header) if "IN-BOU" in h and "TOTAL" in h), None ) out_col = next( (i for i, h in enumerate(header) if "OUT-BOU" in h and "TOTAL" in h), None ) if airline_col is not None and (in_col is not None or out_col is not None): for r in cells[2:]: if len(r) <= airline_col: continue airline = r[airline_col].strip() if ( not airline or airline.upper().startswith("TOTAL") or airline.upper() in ("SN", "S/N") or not is_domestic_airline(airline) ): continue for direction, col in (("in-bound", in_col), ("out-bound", out_col)): if col is None or col >= len(r): continue n = clean_num(r[col]) if n is not None: add_row( rows, value=n, unit="passengers", data_year=source_year, source_file=path.name, source_document=document, source_publisher=publisher, source_year=source_year, source_url_or_path=str(path), page_or_slide_or_article=slide_ref, extraction_method="python-pptx table cell extraction", extraction_date=TODAY, methodology_note=f"Airline-level total {direction} passengers for {airline}; not route-disaggregated.", confidence_flag="high_airline_aggregate", ) if any("LOAD" in h and "FACTOR" in h for h in header): failures.append( f"{path.name} {slide_ref}: load-factor column present but no route rows matched." ) return rows, failures # --------------------------------------------------------------------------- # Markdown article extraction helpers # --------------------------------------------------------------------------- def parse_article_header(text): lines = text.splitlines() title = lines[0].lstrip("# ").strip() if lines else "" source_url, article_date = "", "" for line in lines: m = re.search(r"Source:\s*(https?://\S+)", line, re.IGNORECASE) if m: source_url = m.group(1) m = re.search(r"Date:\s*(.+)", line, re.IGNORECASE) if m: article_date = m.group(1).strip() year = None if article_date: m = re.search(r"20\d{2}", article_date) if m: year = int(m.group()) return title, source_url, article_date, year def extract_thisday(path): rows, failures = [], [] text = path.read_text(encoding="utf-8") title, url, article_date, year = parse_article_header(text) in_domestic_section = False for line in text.splitlines(): if "Domestic market" in line: in_domestic_section = True continue if in_domestic_section and re.match(r"^#+\s", line.strip()): break if not in_domestic_section: continue m = re.match( r"-\s+(.+?):\s+([\d,]+)\s+flights?,\s+([\d,]+)\s+passengers?", line.strip(), ) if m: airline, flights, pax = m.group(1).strip(), clean_num(m.group(2)), clean_num(m.group(3)) if pax is not None: add_row( rows, value=pax, unit="passengers", data_year=2023, source_file=path.name, source_document=title, source_publisher="THISDAYLIVE", source_year=year or 2024, source_url_or_path=url, page_or_slide_or_article=f"article dated {article_date}", extraction_method="regex from markdown", extraction_date=TODAY, methodology_note=f"Domestic airline total passengers for {airline} in 2023 (flights={flights}). Not route-disaggregated.", confidence_flag="high_airline_aggregate", ) return rows, failures def extract_basl(path): rows, failures = [], [] text = path.read_text(encoding="utf-8") title, url, article_date, year = parse_article_header(text) ranking_line = None for line in text.splitlines(): if "Busiest domestic routes" in line: ranking_line = line break if ranking_line: m = re.search(r":\s*(.+)", ranking_line) if m: rank = 0 for item in re.split(r"[,;]", m.group(1)): route = norm_route(item.split("(")[0].strip()) if route: rank += 1 add_row( rows, route=route, value=rank, unit="ranking", data_year=2026, source_file=path.name, source_document=title, source_publisher="Nairametrics / Bi-Courtney", source_year=year or 2026, source_url_or_path=url, page_or_slide_or_article=f"article dated {article_date}", extraction_method="regex from markdown", extraction_date=TODAY, methodology_note="Busiest route ranking from MMA2; no route-level passenger count provided.", confidence_flag="low_rank_only", ) else: failures.append(f"{path.name}: could not locate busiest-routes line.") footfall_patterns = [ (r"Peak season.*?≥?([\d,]+)\s+passenger footfalls? per day", "peak_season_daily_passengers"), (r"Off-peak.*?~?([\d,]+)\s+passengers? per day", "off_peak_daily_passengers"), (r"MMA2 handles ~?([\d]+)% of Lagos domestic traffic", "lagos_domestic_traffic_share_pct"), ] for pat, metric in footfall_patterns: m = re.search(pat, text, re.IGNORECASE) if m: n = clean_num(m.group(1)) if n is not None: add_row( rows, value=n, unit="passengers" if "passenger" in metric else "load_factor_pct", data_year=2026, source_file=path.name, source_document=title, source_publisher="Nairametrics / Bi-Courtney", source_year=year or 2026, source_url_or_path=url, page_or_slide_or_article=f"article dated {article_date}", extraction_method="regex from markdown", extraction_date=TODAY, methodology_note=f"MMA2 aggregate metric: {metric}. Not route-disaggregated.", confidence_flag="medium_aggregate", ) return rows, failures def extract_businessday(path): rows, failures = [], [] text = path.read_text(encoding="utf-8") title, url, article_date, year = parse_article_header(text) hf_match = re.search( r"High-frequency routes:\s*(.+?)(?:\n|Airports seeing)", text, re.IGNORECASE | re.DOTALL ) if hf_match: pieces = [p.strip() for p in re.split(r"[,;]", hf_match.group(1)) if p.strip()] routes = route_pairs_from_origin_list(pieces[0], pieces[1:] if len(pieces) > 1 else []) for route in routes: add_row( rows, route=route, value="", unit="passengers", data_year=2024, source_file=path.name, source_document=title, source_publisher="BusinessDay Nigeria", source_year=year or 2024, source_url_or_path=url, page_or_slide_or_article=f"article dated {article_date}", extraction_method="regex from markdown", extraction_date=TODAY, methodology_note="Route pair listed as high-frequency; no passenger number provided in source.", confidence_flag="low_no_value", ) else: failures.append(f"{path.name}: high-frequency routes not found.") # Aggregate fleet counts quoted in the article patterns = [ (r"13 domestic airlines.*?had ([\d,]+) aircraft", 1, 2024, "Aggregate domestic fleet size (includes aircraft on maintenance)."), (r"10 airlines had ([<>\d,]+) aircraft", 1, None, "Historical aggregate fleet size (>120 aircraft pre-2024)."), (r"number of aircraft on domestic routes was (\d+).*?by 2024 it had fallen to ~?(\d+)", 1, 2022, "Domestic-route aircraft count in 2022."), (r"number of aircraft on domestic routes was (\d+).*?by 2024 it had fallen to ~?(\d+)", 2, 2024, "Domestic-route aircraft count in 2024."), (r"Dana Air grounding removed ~?([\d,]+) aircraft", 1, 2024, "Aircraft removed due to Dana Air grounding."), ] for pat, group, data_year, note in patterns: m = re.search(pat, text) if m: n = clean_num(m.group(group)) if n is not None: add_row( rows, value=n, unit="aircraft_count", data_year=data_year if data_year else 2024, source_file=path.name, source_document=title, source_publisher="BusinessDay Nigeria", source_year=year or 2024, source_url_or_path=url, page_or_slide_or_article=f"article dated {article_date}", extraction_method="regex from markdown", extraction_date=TODAY, methodology_note=note, confidence_flag="medium_aggregate", ) return rows, failures def extract_tbi(path): rows, failures = [], [] text = path.read_text(encoding="utf-8") title, url, article_date, year = parse_article_header(text) # General aircraft types operating in Nigeria (no per-airline counts) m = re.search(r"Aircraft operating in Nigeria include:\s*(.+?)(?:\n\n|\nFleet)", text) if m: for raw_type in re.split(r"[,;]", m.group(1)): raw_type = raw_type.strip() if raw_type: add_row( rows, aircraft_type=raw_type.replace(".", ""), value="", unit="aircraft_count", data_year=2024, source_file=path.name, source_document=title, source_publisher="TBI Africa", source_year=year or 2024, source_url_or_path=url, page_or_slide_or_article=f"article dated {article_date}", extraction_method="regex from markdown", extraction_date=TODAY, methodology_note="Aircraft type reported as operating in Nigeria; no per-route or per-airline count provided.", confidence_flag="medium_fleet_mention", ) else: failures.append(f"{path.name}: aircraft types list not found.") # Fleet snapshots by airline (currently/operational count) fleet_pat = re.compile(r"-\s+(.+?):\s*(.+)") for line in text.splitlines(): m = fleet_pat.match(line.strip()) if not m: continue airline, body = m.group(1).strip(), m.group(2).strip() cm = re.search(r"currently\s+([\d,]+)", body, re.IGNORECASE) or re.search( r"operational\s+([\d,]+)", body, re.IGNORECASE ) if cm: n = clean_num(cm.group(1)) atype = "" type_m = re.search(r"\d+\s+(.+?)(?:;|$)", body) if type_m: atype = type_m.group(1).strip() if n is not None: add_row( rows, aircraft_type=atype, value=n, unit="aircraft_count", data_year=2024, source_file=path.name, source_document=title, source_publisher="TBI Africa", source_year=year or 2024, source_url_or_path=url, page_or_slide_or_article=f"article dated {article_date}", extraction_method="regex from markdown", extraction_date=TODAY, methodology_note=f"Fleet snapshot for {airline}; 'currently/operational' count extracted.", confidence_flag="medium_fleet_snapshot", ) # Route/fare data fare_pat = re.compile(r"([A-Za-z\s]+)(?:–|-)\s*([A-Za-z\s]+):\s*~?₦?\s*([\d,]+)") for line in text.splitlines(): if not line.strip().startswith("-"): continue body = line.lstrip("- ").strip() m = fare_pat.match(body) if m: route = norm_route(f"{m.group(1).strip()}-{m.group(2).strip()}") fare = clean_num(m.group(3)) if route and fare is not None: add_row( rows, route=route, value=fare, unit="fare_ngn", data_year=2024, source_file=path.name, source_document=title, source_publisher="TBI Africa", source_year=year or 2024, source_url_or_path=url, page_or_slide_or_article=f"article dated {article_date}", extraction_method="regex from markdown", extraction_date=TODAY, methodology_note="Average one-way fare for route in NGN.", confidence_flag="medium_fare", ) return rows, failures # --------------------------------------------------------------------------- # Main orchestration # --------------------------------------------------------------------------- def main(): rows = [] failures = [] sources = [ ( BASE / "datasets/transport_aviation/ncaa_summaries/NCAA_2022_annual_executive_summary.pptx", "NCAA 2022 Annual Executive Summary", "Nigerian Civil Aviation Authority (NCAA)", 2022, ), ( BASE / "datasets/transport_aviation/ncaa_summaries/NCAA_Q1_2023_executive_summary.pptx", "NCAA Q1 2023 Executive Summary", "Nigerian Civil Aviation Authority (NCAA)", 2023, ), ] for path, document, publisher, src_year in sources: r, f = extract_pptx(path, document, publisher, src_year) rows.extend(r) failures.extend(f) md_sources = [ (BASE / "docs/transport_aviation/route_data/THISDAY_2023_domestic_airline_passengers.md", extract_thisday), (BASE / "docs/transport_aviation/route_data/BASL_MMA2_busiest_routes_2026.md", extract_basl), (BASE / "docs/transport_aviation/route_data/BusinessDay_lucrative_routes_plane_shortage_2024.md", extract_businessday), (BASE / "docs/transport_aviation/route_data/TBI_domestic_aircraft_grounded_fleet_types_2024.md", extract_tbi), ] for path, extractor in md_sources: r, f = extractor(path) rows.extend(r) failures.extend(f) csv_path = OUT_DIR / "domestic_route_passengers.csv" with csv_path.open("w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=COLUMNS) writer.writeheader() writer.writerows(rows) method_path = OUT_DIR / "domestic_route_passengers_methodology.md" method_path.write_text(build_methodology(rows, failures), encoding="utf-8") print(f"Rows written: {len(rows)}") if failures: print(f"Failures / notes ({len(failures)}):") for note in failures: print(f" - {note}") else: print("No failures reported.") def build_methodology(rows, failures): route_rows = [r for r in rows if r["route"]] pax_rows = [r for r in rows if r["unit"] == "passengers"] fleet_rows = [r for r in rows if r["unit"] == "aircraft_count"] fare_rows = [r for r in rows if r["unit"] == "fare_ngn"] rank_rows = [r for r in rows if r["unit"] == "ranking"] lines = [ "# Domestic Route Passengers – Extraction Methodology", "", f"**Extraction date:** {TODAY}", "", "## Output file", f"`datasets/transport_aviation/derived/domestic_route_passengers.csv` ({len(rows)} rows)", "", "## Sources processed", "1. `NCAA_2022_annual_executive_summary.pptx` — NCAA 2022 full-year domestic airline aggregates and airline totals.", "2. `NCAA_Q1_2023_executive_summary.pptx` — NCAA Q1 2023 domestic airline aggregates and airline totals.", "3. `THISDAY_2023_domestic_airline_passengers.md` — 2023 domestic airline passenger totals by airline.", "4. `BASL_MMA2_busiest_routes_2026.md` — MMA2 busiest route ranking and aggregate footfall numbers.", "5. `BusinessDay_lucrative_routes_plane_shortage_2024.md` — high-frequency route mentions and fleet totals.", "6. `TBI_domestic_aircraft_grounded_fleet_types_2024.md` — fleet snapshots by airline, aircraft types, and route fares.", "", "## Methodology", "* All PPTX tables were read with `python-pptx`. Slide number and table index are recorded in `page_or_slide_or_article`.", "* Markdown articles were parsed with regular expressions; the exact article source URL and date are preserved.", "* City pairs were normalised to uppercase three-letter IATA codes separated by a hyphen (e.g. `LOS-ABV`) using a Nigeria domestic-city lookup table.", "* Numeric values were stripped of commas and spaces before storage.", "* PPTX airline-level totals were restricted to tables that contain known domestic Nigerian carriers; monthly breakdown tables were skipped to avoid duplicate totals.", "* Rows that represent aggregates (national or airline-level) have an empty `route` field because the source does not provide route-disaggregated data.", "", "## Row composition", f"* Total rows: {len(rows)}", f"* Rows with a route pair: {len(route_rows)}", f"* Passenger-unit rows (aggregate + airline): {len(pax_rows)}", f"* Fleet-unit rows (`aircraft_count`): {len(fleet_rows)}", f"* Fare rows (`fare_ngn`): {len(fare_rows)}", f"* Route ranking rows (`ranking`): {len(rank_rows)}", "", "## Key limitations", "* **No route-level passenger volumes** were found in any source. NCAA summaries and THISDAY provide airline-level or national totals only.", "* MMA2 and BusinessDay name high-frequency routes but do not quote per-route passenger numbers.", "* TBI provides per-route airfares, not passenger counts, so those rows use `unit=fare_ngn`.", "* Fleet snapshots are qualitative or point-in-time counts; some aircraft counts may be approximate (e.g. 'about 6').", "", "## Failures / extraction notes", ] if failures: for note in failures: lines.append(f"* {note}") else: lines.append("* None.") lines.append("") return "\n".join(lines) if __name__ == "__main__": main()