"""Rule-based itinerary route parser for the map tab.""" from __future__ import annotations import math import re import unicodedata try: from .map_builder import ( DB, extract_ordered_locations, normalize_location_text, ) except ImportError: # pragma: no cover - supports direct module imports. from map_builder import DB, extract_ordered_locations, normalize_location_text DAY_PREFIX_PATTERN = r"(?:Day|Jour|Giorno|Ngay|Ngày)" ROUTE_SEPARATOR_PATTERN = re.compile(r"\s*(?:->|>|/|\||;|\s+-\s+|,\s+|\s*&\s+)\s*") META_CLAUSE_SEPARATOR_PATTERN = re.compile(r"\s*(?:/|\||;|\s*&\s*)\s*") MODE_KEYWORDS = { "plane": { "vol", "flight", "flights", "avion", "plane", "fly", "flying", "volo", "air", "aerien", "aerienne", "aereo", }, "train": { "train", "rail", "railway", "tgv", "treno", "night train", "sleeper train", }, "boat": { "boat", "bateau", "cruise", "ferry", "junk", "croisiere", "barca", "speedboat", "speed boat", "hydrofoil", "du thuyen", }, "car": { "car", "road", "route", "drive", "driving", "driver", "transfer", "transfert", "trasferimento", "vehicle", "van", "bus", "coach", "xe", }, } ARRIVAL_WORDS = {"arrival", "arrivee", "arrive", "arrivo"} DEPARTURE_WORDS = {"departure", "depart", "partenza"} METADATA_WORDS = ARRIVAL_WORDS | DEPARTURE_WORDS PENDING_SKIP_WORDS = { "airport", "aeroport", "station", "pier", "port", "harbor", "hotel", "resort", "homestay", "guide", "chauffeur", "driver", "transfer", "transfert", "arrival", "arrivee", "arrive", "departure", "depart", "day", "jour", } def _strip_day_prefix(text: str) -> str: cleaned = unicodedata.normalize("NFKC", text or "").strip() cleaned = re.sub(r"[–—−‐‑‒―]", "-", cleaned) return re.sub(rf"^{DAY_PREFIX_PATTERN}\s*\d+[:\s\-–—]*", "", cleaned, flags=re.I).strip() def _phrase_in_normalized(phrase: str, normalized_text: str) -> bool: phrase_norm = normalize_location_text(phrase) if not phrase_norm: return False return bool(re.search(rf"(? str | None: normalized = normalize_location_text(text) if not normalized: return None matched = [] for mode, phrases in MODE_KEYWORDS.items(): if any(_phrase_in_normalized(phrase, normalized) for phrase in phrases): matched.append(mode) if not matched: return None for mode in ("plane", "train", "boat", "car"): if mode in matched: return mode return None def _metadata_kind(text: str) -> str | None: normalized = normalize_location_text(text) if not normalized: return None first_token = normalized.split()[0] if first_token in ARRIVAL_WORDS: return "arrival" if first_token in DEPARTURE_WORDS: return "departure" return None def _trailing_metadata_kind(text: str) -> str | None: normalized = normalize_location_text(text) if not normalized: return None tokens = normalized.split() if not tokens: return None if tokens[-1] in ARRIVAL_WORDS: return "arrival" if tokens[-1] in DEPARTURE_WORDS: return "departure" return None def _remove_metadata_prefix(text: str) -> str: return re.sub( r"^\s*(?:arrival|arrivee|arriv[eé]e?|arrivo|departure|depart|d[eé]part|partenza)" r"\s*(?::|\b)?\s*(?:at|a|à|to|vers|from|de|da)?\s*", "", text, flags=re.I, ).strip(" -") def _resolved_city_list(text: str, include_unknown: bool = True) -> tuple[list[str], list[str]]: cities = [] pending = [] seen = set() for candidate in extract_ordered_locations(text, include_unknown=include_unknown): matched = DB.find_city(candidate) if matched: key = normalize_location_text(matched.name) if key not in seen: cities.append(matched.name) seen.add(key) elif include_unknown and _keep_pending_candidate(candidate): key = normalize_location_text(candidate) if key not in seen: pending.append(candidate) seen.add(key) return cities, pending def _keep_pending_candidate(candidate: str) -> bool: normalized = normalize_location_text(candidate) if not normalized or len(normalized) < 3: return False tokens = normalized.split() if not tokens or len(tokens) > 6: return False if any(token in PENDING_SKIP_WORDS for token in tokens): return False if any(token in METADATA_WORDS for token in tokens): return False if _mode_from_text(candidate): return False return True def _haversine_km(city_a: str, city_b: str) -> float | None: a = DB.find_city(city_a) b = DB.find_city(city_b) if not a or not b: return None lat1, lon1 = math.radians(a.lat), math.radians(a.lon) lat2, lon2 = math.radians(b.lat), math.radians(b.lon) dlat, dlon = lat2 - lat1, lon2 - lon1 h = math.sin(dlat / 2) ** 2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon / 2) ** 2 return 6371 * 2 * math.asin(math.sqrt(h)) def _default_mode(city_a: str, city_b: str) -> str: distance = _haversine_km(city_a, city_b) if distance is not None and distance > 300: return "plane" return "car" def _extract_metadata(line: str) -> tuple[str | None, str | None, str | None]: arrival = None departure = None route_parts = [] any_metadata = False for clause in META_CLAUSE_SEPARATOR_PATTERN.split(line): clause = clause.strip() if not clause: continue kind = _metadata_kind(clause) or _trailing_metadata_kind(clause) if not kind: route_parts.append(clause) continue cities, _ = _resolved_city_list(clause, include_unknown=False) if not cities: continue any_metadata = True if kind == "arrival": arrival = arrival or cities[0] else: departure = cities[-1] if len(cities) > 1: route_parts.append(_remove_metadata_prefix(clause)) if not any_metadata: return None, None, line route_text = " - ".join(part for part in route_parts if part.strip()) return arrival, departure, route_text or None def _route_fragments(line: str) -> list[str]: line = re.sub(r"\s*&\s*(?=(?:vol|flight|volo)\b)", " - ", line, flags=re.I) return [fragment.strip() for fragment in ROUTE_SEPARATOR_PATTERN.split(line) if fragment.strip()] def _append_leg(legs: list[dict], city_a: str, city_b: str, mode: str | None) -> None: if not city_a or not city_b or city_a == city_b: return legs.append({"a": city_a, "b": city_b, "mode": mode or _default_mode(city_a, city_b)}) def _parse_route_line(line: str) -> tuple[list[dict], list[str]]: fragments = _route_fragments(line) if not fragments: return [], [] fragment_modes = [_mode_from_text(fragment) for fragment in fragments] whole_line_mode = None if any(fragment_modes) else _mode_from_text(line) legs = [] pending = [] previous_city = None for fragment, fragment_mode in zip(fragments, fragment_modes): cities, unknowns = _resolved_city_list(fragment, include_unknown=True) pending.extend(unknowns) if not cities: continue incoming_mode = fragment_mode or whole_line_mode if previous_city and cities[0] != previous_city: _append_leg(legs, previous_city, cities[0], incoming_mode) for index in range(len(cities) - 1): _append_leg(legs, cities[index], cities[index + 1], incoming_mode) previous_city = cities[-1] return legs, pending def parse_route_text(text: str) -> dict: """Parse free-form itinerary text into map legs and route metadata.""" legs = [] pending = [] arrival = None departure = None for raw_line in (text or "").splitlines(): line = _strip_day_prefix(raw_line) if not line: continue parsed_arrival, parsed_departure, route_text = _extract_metadata(line) arrival = arrival or parsed_arrival departure = parsed_departure or departure if not route_text: continue line_legs, line_pending = _parse_route_line(route_text) legs.extend(line_legs) pending.extend(line_pending) return { "legs": legs, "pending": sorted(set(pending)), "arrival": arrival, "departure": departure, }