""" 空运 TMS 线路报价:从 airOptions 拆出可选线路。 本文件职责:判断一条还是多条、拼航线(含中转)、抽出该线路费用。 选定后把起运港写回 facts。群摘要现场投影同一套分项。 中转只用于展示,不写进报价单模板。 禁止:猜没回的机场名;把贸易条款当线路;改六态。 """ from __future__ import annotations import json import logging from typing import Any from agent.schema.tms_air_query import _AIRPORTS, resolve_airport_code logger = logging.getLogger(__name__) # 三字码 → 销售侧地名;同码多个中文时取较短的常用名。 _CODE_NAME: dict[str, str] = {} for name, code in _AIRPORTS.items(): prev = _CODE_NAME.get(code) if prev is None or len(name) < len(prev): _CODE_NAME[code] = name def parse_air_options(quote: dict[str, Any] | None) -> list[dict[str, Any]]: """把 quote.airOptions(list 或 JSON 串)收成字典列表。""" src = dict(quote or {}) raw = src.get("airOptions") or src.get("air_options") if isinstance(raw, str) and raw.strip(): try: raw = json.loads(raw) except json.JSONDecodeError: logger.warning("air.routes airOptions 不是 JSON") return [] if not isinstance(raw, list): return [] out: list[dict[str, Any]] = [] for item in raw: if isinstance(item, dict): out.append(dict(item)) return out def needs_route_pick(quote: dict[str, Any] | None) -> bool: """没写起运地、且多于一条线路,才进 H5 选。有起运地时由调用方先按起运地收口。""" return len(parse_air_options(quote)) > 1 def options_matching_origin( options: list[dict[str, Any]], origin_text: str, ) -> list[dict[str, Any]]: """ 销售已经写了起运地时,只留起运机场对得上的线路。 杭州对 HGH,珠海对 ZUH。回包里的南京/深圳/郑州不能拿来给这张单选。 对不上返回空列表。不改入参。 """ text = (origin_text or "").strip() if not text: return list(options) code = resolve_airport_code(text).upper() raw = text.upper() matched: list[dict[str, Any]] = [] for opt in options: origin = str(opt.get("originAirportCode") or opt.get("origin") or "").strip().upper() if not origin: continue if code and origin == code: matched.append(opt) elif origin == raw: matched.append(opt) return matched def choose_stated_origin_option( options: list[dict[str, Any]], quote: dict[str, Any] | None = None, ) -> dict[str, Any]: """ 起运地已指定时直接用一条,不再让销售选线。 只有一条就用它。多条时优先和报价总价相同的那条,否则用第一条。 """ if len(options) == 1: return options[0] total = (quote or {}).get("total") for opt in options: if _same_amount(opt.get("salePrice"), total): return opt return options[0] def _place_name(code: str, fallback: str = "") -> str: """三字码能对上就用中文,否则用原文或 fallback。""" raw = (code or "").strip() if not raw: return (fallback or "").strip() mapped = _CODE_NAME.get(raw.upper()) if mapped: return mapped if resolve_airport_code(raw): return _CODE_NAME.get(resolve_airport_code(raw), raw) return raw def route_line(option: dict[str, Any] | None, facts: dict[str, str] | None = None) -> str: """ 销售侧航线:有中转写成 起运-中转-目的。 对不上字典就展示三字码,禁止编城市。 """ src = dict(option or {}) facts = facts or {} origin = _place_name( str(src.get("originAirportCode") or src.get("origin") or ""), str(facts.get("起运港") or ""), ) dest = _place_name( str(src.get("destinationAirportCode") or src.get("destination") or ""), str(facts.get("目的港") or ""), ) transfer = _place_name(str(src.get("transferAirportCode") or src.get("transfer") or "")) bits = [x for x in (origin, transfer, dest) if x] return "-".join(bits) def _iata_code(raw: str, fallback: str = "") -> str: """只要三字码。已是代码就用;中文能投影就投影。禁止编假码。""" text = (raw or "").strip() if len(text) == 3 and text.isalpha(): return text.upper() mapped = resolve_airport_code(text) or resolve_airport_code(fallback) return (mapped or "").upper() def airport_code_for_display(raw: str, fallback: str = "") -> str: """ 锁舱成功回执上的起运地/目的地。 能收成三字码就用三字码(NKG、KUL)。对不上就留原文,不编城市。 """ code = _iata_code(raw, fallback) if code: return code return (raw or fallback or "").strip() def route_codes( option: dict[str, Any] | None, facts: dict[str, str] | None = None, quote: dict[str, Any] | None = None, ) -> str: """ 价格明细/方案卡用的航线三字码:有中转写成 ZUH-CAN-CRK。 对不上就不写假中转。 selected_air_option 可能是字典、JSON 串,或只写了舱位编号; 不能直接 dict(字符串),否则成交卡拼航线会整单炸掉(WO001)。 """ src = _parse_option_dict(option) if not src: q = dict(quote or {}) if isinstance(quote, dict) else {} # 先认 camelCase,和主账快照字段对齐。 src = _parse_option_dict( q.get("selected_air_option") or q.get("selectedAirOption") ) if not src: opts = parse_air_options(q) src = dict(opts[0]) if opts else {} facts = facts or {} origin = _iata_code( str(src.get("originAirportCode") or src.get("origin") or facts.get("originCode") or ""), str(facts.get("起运港") or ""), ) dest = _iata_code( str(src.get("destinationAirportCode") or src.get("destination") or facts.get("destinationCode") or ""), str(facts.get("目的港") or ""), ) transfer = _iata_code( str( src.get("transferAirportCode") or src.get("transfer") or (quote or {}).get("transferAirportCode") or "" ) ) bits = [x for x in (origin, transfer, dest) if x] return "-".join(bits) def scheme_route_codes( quote: dict[str, Any] | None, facts: dict[str, str] | None = None, ) -> str: """方案卡航线:空运ZUH→中转→CRK。没有代码则退回地名。""" codes = route_codes(None, facts, quote) if codes: return "空运" + codes.replace("-", "→") facts = facts or {} origin = (facts.get("起运港") or "-").strip() or "-" dest = (facts.get("目的港") or "-").strip() or "-" return f"空运{origin}→{dest}" def _fee_row(item: dict[str, Any]) -> tuple[str, str] | None: """一条 TMS 费用。有名称才算;金额空则用单价。禁止编费目。""" name = str( item.get("chargeName") or item.get("feeName") or item.get("name") or "" ).strip() if not name: return None ccy = str(item.get("currency") or "").strip() amount = item.get("amount") if amount is None or str(amount).strip() == "": amount = item.get("unitPrice") or item.get("unit_price") text = str(amount or "").strip() if ccy and text and ccy not in text: text = f"{ccy} {text}" return (name, text or "-") def _rows_from_list(raw: Any) -> list[tuple[str, str]]: """把 TMS 费用数组收成展示行。不是列表就当没有。""" if isinstance(raw, str) and raw.strip(): try: raw = json.loads(raw) except json.JSONDecodeError: return [] if not isinstance(raw, list): return [] out: list[tuple[str, str]] = [] for item in raw: if isinstance(item, dict): row = _fee_row(item) if row: out.append(row) return out def _fee_item_dicts(raw: Any) -> list[dict[str, Any]]: """TMS 费用数组收成字典;字符串 JSON 也认。无名称的跳过。""" if isinstance(raw, str) and raw.strip(): try: raw = json.loads(raw) except json.JSONDecodeError: return [] if not isinstance(raw, list): return [] out: list[dict[str, Any]] = [] for item in raw: if not isinstance(item, dict): continue name = str( item.get("chargeName") or item.get("feeName") or item.get("name") or item.get("label") or "" ).strip() if not name: continue row = dict(item) row["name"] = name out.append(row) return out def _top_fee_items_for_option( quote: dict[str, Any] | None, option: dict[str, Any] | None, ) -> list[dict[str, Any]]: """ 舱位缺 feeItems 时补整单顶层(OMS 空运表价常把空运费放在 data.feeItems)。 顶层带 airOptionNo 时优先本舱位;对不上编号时不能退成空列表(否则只剩增点/拆板)。 与主账 InquiryTmsQueryService.extractAirDepartureCharges 口径一致。 """ items = _fee_item_dicts((quote or {}).get("feeItems") or (quote or {}).get("fee_items")) if not items: return [] opt_no = str( (option or {}).get("optionNo") or (option or {}).get("option_no") or "" ).strip() tagged = [ row for row in items if str(row.get("airOptionNo") or row.get("air_option_no") or "").strip() ] if opt_no and tagged: matched = [ row for row in items if str(row.get("airOptionNo") or row.get("air_option_no") or "").strip() == opt_no ] if matched: return matched return items def _has_air_freight_name(rows: list[dict[str, Any]]) -> bool: """明细里是否已有空运段运费名。""" for row in rows: name = str(row.get("name") or "").strip() if name == "空运费" or name == "航空运费" or (name.endswith("运费") and "空运" in name): return True return False def _only_origin_surcharges(rows: list[dict[str, Any]]) -> bool: """ 是否只剩起运附加费(增点/拆板等),没有运输费类分项。 有运输费/提送货费时不注入总价空运费,避免和 ZUH 分项叠两套。 """ if not rows: return False for row in rows: name = str(row.get("name") or "").strip() if not name: continue if name in {"运输费", "提送货费", "人工费", "空运费", "航空运费"}: return False if "运费" in name and "增点" not in name and "拆板" not in name: return False return True def _air_freight_from_quote_fee_lines( quote: dict[str, Any] | None, ) -> list[dict[str, Any]]: """主账 fee_lines 里已有的空运费行(带单价/单位更好)。""" rows = _fee_item_dicts( (quote or {}).get("fee_lines") or (quote or {}).get("fee_rows") ) return [row for row in rows if str(row.get("name") or "").strip() in {"空运费", "航空运费"}] def _inject_air_freight_from_option( option: dict[str, Any] | None, quote: dict[str, Any] | None, ) -> dict[str, Any] | None: """ 舱位/顶层都没有空运费时,用 salePrice 或 unitPrice×计费重补一行。 对齐主账 injectAirFreightFeeLineIfMissing。无可用金额则不补。 """ src = dict(option or {}) q = dict(quote or {}) unit_price = src.get("unitPrice") if unit_price in (None, "", "-"): unit_price = src.get("unit_price") or src.get("pricePerKg") or src.get("freightRate") sale = src.get("salePrice") if sale in (None, "", "-"): sale = src.get("sale_price") or q.get("total") or q.get("salePrice") cw = src.get("chargeableWeightKg") if cw in (None, "", "-"): cw = src.get("chargeable_weight_kg") or q.get("chargeableWeightKg") amount = "" price = str(unit_price or "").strip() cw_text = str(cw or "").strip() if price and cw_text: try: amount = str( (float(str(price).replace(",", "")) * float(cw_text.replace(",", ""))) ) # 去掉多余尾零:3334.0 amount = f"{float(amount):.10g}" except ValueError: amount = "" if not amount: amount = str(sale or "").strip() if not amount or amount == "-": return None if (not price or price == "-") and cw_text: try: price = f"{(float(str(amount).replace(',', '')) / float(cw_text.replace(',', ''))):.10g}" except ValueError: price = amount ccy = str(src.get("currency") or q.get("currency") or "").strip() row: dict[str, Any] = { "name": "空运费", "amount": amount, "project": "空运段", } if ccy: row["currency"] = ccy if price and price != "-": row["unit_price"] = price row["unitPrice"] = price if cw_text: row["quantity"] = cw_text row["unit"] = "KG" return row def option_fee_dicts( option: dict[str, Any] | None, quote: dict[str, Any] | None = None, ) -> list[dict[str, Any]]: """ 该线路费用明细(保留单价/单位/数量),供群摘要与投影写回 fee_lines。 顺序与主账一致:空运段 feeItems → 起运港 departureCharges → 目的地。 舱位 feeItems 为空时补整单顶层,避免只剩增点费/拆板费丢掉空运费(WO202610150047)。 """ src = dict(option or {}) q = dict(quote or {}) cabin = _fee_item_dicts(src.get("feeItems") or src.get("fee_items")) if not cabin: cabin = _top_fee_items_for_option(q, src) dep = _fee_item_dicts( src.get("departureCharges") or src.get("departure_charges") or src.get("charges") ) dest = _fee_item_dicts( src.get("destinationCharges") or src.get("destination_charges") or src.get("arrivalCharges") ) rows: list[dict[str, Any]] = [] seen: set[str] = set() for bucket in (cabin, dep, dest): for row in bucket: key = str(row.get("name") or "").strip().lower() if not key or key in seen: continue seen.add(key) rows.append(row) # 仍无空运费:先取主账 fee_lines,再按舱位总价注入(OMS 仅增点/拆板时) if rows and not _has_air_freight_name(rows): for row in _air_freight_from_quote_fee_lines(q): key = str(row.get("name") or "").strip().lower() if key and key not in seen: seen.add(key) rows.insert(0, row) if not _has_air_freight_name(rows) and _only_origin_surcharges(rows): extra = _inject_air_freight_from_option(src, q) if extra: rows.insert(0, extra) if rows: return rows options = parse_air_options(q) if len(options) > 1: return [] return _fee_item_dicts(q.get("fee_lines") or q.get("fee_rows") or q.get("feeItems")) def option_fee_lines(option: dict[str, Any] | None, quote: dict[str, Any] | None = None) -> list[tuple[str, str]]: """ 该线路 TMS 回了哪些费用就列哪些(名称 + 展示金额)。 明细字典见 option_fee_dicts:舱位缺空运段时会补整单顶层 feeItems。 """ out: list[tuple[str, str]] = [] for row in option_fee_dicts(option, quote): pair = _fee_row(row) if pair: out.append(pair) return out def option_total(option: dict[str, Any] | None, quote: dict[str, Any] | None = None) -> str: """该线路总销售价。""" src = dict(option or {}) q = dict(quote or {}) raw = src.get("salePrice") if raw is None or str(raw).strip() == "": raw = q.get("total") or q.get("salePrice") ccy = str(src.get("currency") or q.get("currency") or "").strip() text = str(raw or "").strip() or "-" if ccy and text != "-" and ccy not in text: return f"{ccy} {text}" return text # TMS 舱位「班期」可能用的键。页面只展示第一个非空值,例如 D234567,不翻译。 _SCHEDULE_KEYS = ( "班期", "schedule", "flightSchedule", "flight_schedule", "frequency", "weekFrequency", "scheduleDays", "operatingDays", "flightFrequency", "departureSchedule", "scheduleCode", "flightPeriod", "frequencyCode", ) def option_flight_date(option: dict[str, Any] | None) -> str: """ 舱位上的航班日期。优先 flightDate,其次从起飞时间取日期。 没有则空串。不编当天。 """ src = dict(option or {}) for key in ("flightDate", "flight_date", "departureDate", "departure_date"): text = str(src.get(key) or "").strip() if text: return _date_head(text) return _date_head(str(src.get("departureTime") or src.get("departure_time") or "")) def _date_head(raw: str) -> str: """yyyy-MM-dd 开头就取日期,否则原样。空串保持空。""" text = (raw or "").strip() if len(text) >= 10 and text[4] == "-" and text[7] == "-": return text[:10] return text def option_schedule(option: dict[str, Any] | None) -> str: """班期按 TMS 原样,例如 D234567。没有就空串,不编。""" src = dict(option or {}) for key in _SCHEDULE_KEYS: text = str(src.get(key) or "").strip() if text: return text return "" def option_airline(option: dict[str, Any] | None) -> str: """航司:有中文名用中文名,否则用代码。没有就空串。""" src = dict(option or {}) for key in ("airlineName", "airline", "airlineCode", "carrier"): text = str(src.get(key) or "").strip() if text: return text return "" def option_endpoint_name( option: dict[str, Any] | None, side: str, facts: dict[str, str] | None = None, ) -> str: """ 舱位起运地或目的地。三字码能对上就用中文,对不上展示代码。 side 取 origin / destination。不编城市。 """ src = dict(option or {}) facts = facts or {} if side == "destination": code = str(src.get("destinationAirportCode") or src.get("destination") or "") fallback = str(facts.get("目的港") or facts.get("目的地") or "") else: code = str(src.get("originAirportCode") or src.get("origin") or "") fallback = str(facts.get("起运港") or facts.get("起运地") or "") return _place_name(code, fallback) def _price_sort_value(option: dict[str, Any]) -> float: """总价排序用。解析不出的排到同一天最后。""" raw = option.get("salePrice") if raw is None or str(raw).strip() == "": return float("inf") num = "" for ch in str(raw).replace(",", ""): if ch.isdigit() or ch == ".": num += ch elif num: break try: return float(num) if num else float("inf") except ValueError: return float("inf") def sort_cabin_options(options: list[dict[str, Any]] | None) -> list[dict[str, Any]]: """ 没填报价日期时的舱位顺序:航班日期从早到晚,同一天总价从低到高。 没有日期的排在最后。不改入参。 """ rows = [dict(item) for item in list(options or [])] def _key(item: dict[str, Any]) -> tuple[str, float]: day = option_flight_date(item) or "9999-99-99" return (day, _price_sort_value(item)) return sorted(rows, key=_key) def apply_selected_option(facts: dict[str, str] | None, option: dict[str, Any] | None) -> dict[str, str]: """ 选定舱位后写回起运港(及能对上的三字码)。改选覆盖。 工单还没有报价日期时,把这条的航班日期写回去,供报价单和锁舱使用。 销售已经说了日期则不覆盖。 """ from agent.schema.field_validate import is_blank_quote_date merged = dict(facts or {}) src = dict(option or {}) flight_date = option_flight_date(src) # 占位文案「(非必填):」当没填,必须让航班日期盖上去。 if flight_date and is_blank_quote_date(merged.get("报价日期")): merged["报价日期"] = flight_date code = str(src.get("originAirportCode") or src.get("origin") or "").strip().upper() name = _place_name(code) if name: merged["起运港"] = name if code and len(code) == 3 and code.isalpha(): merged["originCode"] = code dest_code = str(src.get("destinationAirportCode") or "").strip().upper() dest_name = _place_name(dest_code, str(merged.get("目的港") or "")) if dest_name: merged["目的港"] = dest_name if dest_code and len(dest_code) == 3 and dest_code.isalpha(): merged["destinationCode"] = dest_code return merged def project_selected_quote(quote: dict[str, Any] | None, option: dict[str, Any] | None) -> dict[str, Any]: """把选中线路投影成价格卡用的 quote。""" base = dict(quote or {}) src = dict(option or {}) # 保留单价/单位/数量,避免后面把「空运费=总价」误判成重复兜底删掉。 lines: list[dict[str, Any]] = [] for row in option_fee_dicts(src, base): pair = _fee_row(row) if not pair: continue name, amount = pair line = { "name": name, "amount": amount, "currency": str(row.get("currency") or row.get("currencyCode") or "").strip(), } unit = str(row.get("unit") or row.get("unitName") or row.get("chargeUnit") or "").strip() if unit: line["unit"] = unit unit_price = row.get("unitPrice") if unit_price in (None, ""): unit_price = row.get("unit_price") if unit_price not in (None, ""): line["unit_price"] = unit_price line["unitPrice"] = unit_price qty = row.get("quantity") if qty in (None, ""): qty = row.get("chargeQuantity") or row.get("charge_quantity") if qty not in (None, ""): line["quantity"] = qty lines.append(line) # 必须覆盖整单 fee_lines:空列表也要写回去,否则会残留别的线路空运费。 base["fee_lines"] = lines if not lines: base["freight"] = "-" base["other_fee"] = "-" total = src.get("salePrice") if total is not None and str(total).strip() != "": base["total"] = str(total).strip() if src.get("currency"): base["currency"] = src.get("currency") if src.get("validUntil"): base["valid_until"] = src.get("validUntil") base["selected_air_option"] = src base["route_line"] = route_line(src, {}) # 选定后不再是待选舱快照。 base.pop("pending_cabin_pick", None) return base def _parse_option_dict(raw: Any) -> dict[str, Any]: """selected_air_option:字典或 JSON 串。解不开当没有。""" if isinstance(raw, str) and raw.strip(): try: raw = json.loads(raw) except json.JSONDecodeError: return {} return dict(raw) if isinstance(raw, dict) else {} def resolve_display_air_option( quote: dict[str, Any] | None, facts: dict[str, str] | None = None, ) -> dict[str, Any] | None: """ 群摘要用哪一条线路:已选定 > 仅 1 条 > 用工单起运/目的对上的那条。 多条对不上就不猜。禁止把整单最低价的空运费套到没选的线路。 """ src = dict(quote or {}) selected = _parse_option_dict(src.get("selected_air_option")) if selected: return selected options = parse_air_options(src) if not options: return None if len(options) == 1: return options[0] facts = facts or {} origin = str(facts.get("originCode") or "").strip().upper() if not origin: origin = resolve_airport_code( str(facts.get("起运港") or facts.get("起运地") or "") ) dest = str(facts.get("destinationCode") or "").strip().upper() if not dest: dest = resolve_airport_code( str(facts.get("目的港") or facts.get("目的地") or "") ) matched: list[dict[str, Any]] = [] for opt in options: oc = str(opt.get("originAirportCode") or opt.get("origin") or "").strip().upper() dc = str( opt.get("destinationAirportCode") or opt.get("destination") or "" ).strip().upper() if origin and oc and oc != origin: continue if dest and dc and dc != dest: continue if origin or dest: matched.append(opt) if len(matched) == 1: return matched[0] return None def _amount_digits(raw: Any) -> str: """只留数字和小数点,便于判断空运费是否就是总价。""" text = str(raw or "").strip().replace(",", "") for prefix in ("CNY", "USD", "EUR", "HKD", "RMB"): if text.upper().startswith(prefix): text = text[len(prefix):].strip() break return text def _same_amount(left: Any, right: Any) -> bool: """金额是否同一笔。币种前缀、逗号、小数末零忽略。""" a = _amount_digits(left) b = _amount_digits(right) if not a or not b or a == "-" or b == "-": return False try: return float(a) == float(b) except ValueError: return a == b def _rated_air_freight_line(item: dict[str, Any] | None) -> bool: """ 是否为带计价单位的真实空运费行(单价×计费重),而非「空运费=总价」快照兜底。 有 KG 单位、或有单价/数量,都按真实费率行保留。 """ src = dict(item or {}) unit = str(src.get("unit") or src.get("unitName") or src.get("chargeUnit") or "").strip().upper() unit = unit.replace("/", "").replace("每", "").replace(" ", "") if unit in {"KG", "KGS", "公斤", "千克"}: return True if src.get("unitPrice") not in (None, "", "-") or src.get("unit_price") not in (None, "", "-"): return True if src.get("quantity") not in (None, "", "-") or src.get("chargeQuantity") not in (None, "", "-"): return True return False def drop_air_freight_total_alias(quote: dict[str, Any] | None) -> dict[str, Any]: """ 已经有其它费用分项时,去掉「空运费=总价」这一行兜底。 舱位上有运输费、提送货费时,主账快照常再塞一行空运费,金额就是总价,不能和分项叠在一起。 只有空运费这一项时必须留下:TMS 经常只回这一项,金额自然等于总价,群里仍要写出费用名。 不写库。群摘要投影之后调用。 """ src = dict(quote or {}) total = src.get("total") raw = src.get("fee_lines") or src.get("fee_rows") parsed: Any = raw if isinstance(raw, str) and raw.strip(): try: parsed = json.loads(raw) except json.JSONDecodeError: parsed = None if not isinstance(parsed, list): return src others: list[Any] = [] alias: list[Any] = [] for item in parsed: if not isinstance(item, dict): others.append(item) continue name = str(item.get("name") or item.get("label") or item.get("feeName") or "").strip() amount = item.get("amount") or item.get("value") # OMS 空运费金额常等于顶层 salePrice,但带 KG 单价/数量,不是「总价兜底」假行。 if name == "空运费" and _same_amount(amount, total) and not _rated_air_freight_line(item): alias.append(item) continue others.append(item) if others: # 分项已经在,空运费=总价是重复兜底。运费字段同值也不再补一行。 src["fee_lines"] = others if _same_amount(src.get("freight"), total): src["freight"] = "-" return src # 明细就是这一条空运费,群摘要要显示费用名,不能只剩费用合计。 src["fee_lines"] = alias return src def project_group_air_quote( quote: dict[str, Any] | None, facts: dict[str, str] | None = None, ) -> dict[str, Any]: """ 群摘要 TMS 段:有线路就投影该线路费用,与单聊价格明细同一套分项。 主账查价快照常只有「空运费=总价」;选定后的明细没写回主账,必须现场投影。 副作用:无。不写库、不改六态。 """ src = dict(quote or {}) option = resolve_display_air_option(src, facts) if option: src = project_selected_quote(src, option) return drop_air_freight_total_alias(src)