654 lines
23 KiB
Python
654 lines
23 KiB
Python
"""
|
||
调价规则：抽项结构、费用对齐、比例/改到、币种合计、利润点。
|
||
|
||
本文件职责：只算新价，不发企微、不改六态、不写主账。
|
||
模型只负责「是不是调价、调哪几项」；金额必须走这里，禁止模型改价。
|
||
调用：私聊/群 Worker 槽；禁止回调线程同步跑。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_CLOSED = frozenset({"已成交", "未成交", "已关闭"})
|
||
_QUOTED = frozenset({"已报价", "协商中"})
|
||
|
||
_PERCENT_UP = re.compile(
|
||
r"(上调|上涨|上升|上浮|提高|加|涨|升)\s*(\d+(?:\.\d+)?)\s*%"
|
||
)
|
||
_PERCENT_DOWN = re.compile(r"(下调|下降|降低|减|降)\s*(\d+(?:\.\d+)?)\s*%")
|
||
_TARGET = re.compile(
|
||
r"(提升到|减少到|上调到|下调到|加到|减到|调到|改为|改成)\s*(\d+(?:\.\d+)?)"
|
||
)
|
||
_SPLIT = re.compile(r"[，,；;。！!？?\n]|和(?=\S)")
|
||
_FEE_TOKEN = re.compile(r"([\u4e00-\u9fffA-Za-z0-9]{1,16}费)")
|
||
_TOTAL_WORDS = ("合计", "总价", "汇总价", "总销售价", "总成本价")
|
||
_QUOTE_WORD = "报价"
|
||
|
||
|
||
@dataclass
|
||
class AdjustItem:
|
||
"""一条抽出来的调价项。kind=percent 时 value 是比例；kind=target 时 value 是金额。"""
|
||
|
||
fee_name: str
|
||
kind: str
|
||
direction: str
|
||
value: float
|
||
|
||
|
||
@dataclass
|
||
class AdjustExtract:
|
||
"""一句话抽完的调价意图。wants_total=只动合计数字，不改明细。"""
|
||
|
||
is_adjust: bool
|
||
items: list[AdjustItem] = field(default_factory=list)
|
||
wants_total: bool = False
|
||
|
||
|
||
@dataclass
|
||
class ApplyResult:
|
||
"""规则算完的结果。quote 是新副本，入参原报价不被改。"""
|
||
|
||
ok: bool
|
||
code: str
|
||
quote: dict[str, Any]
|
||
applied: list[dict[str, Any]] = field(default_factory=list)
|
||
unmatched: list[str] = field(default_factory=list)
|
||
ambiguous: list[str] = field(default_factory=list)
|
||
currency_hint: bool = False
|
||
lines_hint: bool = False
|
||
profit_point: Optional[float] = None
|
||
|
||
|
||
def private_session_can_adjust(status: str, has_quote: bool, collab_chat_id: str) -> str:
|
||
"""
|
||
私聊能不能按未协同套走。空串=可以。
|
||
|
||
already_collab / closed / no_quote 给话术用。
|
||
"""
|
||
if (collab_chat_id or "").strip():
|
||
return "already_collab"
|
||
st = (status or "").strip()
|
||
if st in _CLOSED:
|
||
return "closed"
|
||
if st not in _QUOTED or not has_quote:
|
||
return "no_quote"
|
||
return ""
|
||
|
||
|
||
def _fee_key(name: str) -> str:
|
||
raw = (name or "").replace(" ", "").replace("（", "(").replace("）", ")")
|
||
for sep in ("/", "／"):
|
||
if sep in raw:
|
||
raw = raw.split(sep, 1)[0]
|
||
return raw.strip()
|
||
|
||
|
||
def fee_name_same(left: str, right: str) -> bool:
|
||
"""费用名对齐：海运费=海运费用；操作费 ≠ 码头操作费。"""
|
||
a = _fee_key(left)
|
||
b = _fee_key(right)
|
||
if not a or not b:
|
||
return False
|
||
if a == b:
|
||
return True
|
||
if a == b + "用" or a == b + "费用" or b == a + "用" or b == a + "费用":
|
||
return True
|
||
if a + "费" == b or b + "费" == a:
|
||
return True
|
||
if len(a) >= 3 and len(a) == len(b) and a[:-1] == b[:-1] and (a.endswith("费") or b.endswith("费")):
|
||
return True
|
||
return False
|
||
|
||
|
||
def match_fee_names(spoken: str, rows: list[dict[str, Any]]) -> list[str]:
|
||
"""对上当前明细里的费用名；0/1/多条都原样返回。"""
|
||
hits: list[str] = []
|
||
for row in rows:
|
||
name = str(row.get("name") or "").strip()
|
||
if name and fee_name_same(spoken, name) and name not in hits:
|
||
hits.append(name)
|
||
if hits:
|
||
return hits
|
||
spoken_key = _fee_key(spoken)
|
||
if spoken_key:
|
||
for row in rows:
|
||
name = str(row.get("name") or "").strip()
|
||
key = _fee_key(name)
|
||
if key and (spoken_key in key or key in spoken_key) and name not in hits:
|
||
hits.append(name)
|
||
return hits
|
||
|
||
|
||
def quote_fee_rows(quote: dict[str, Any] | None) -> list[dict[str, Any]]:
|
||
"""抽出费用行。TMS 常把 fee_lines 存成 JSON 字符串。"""
|
||
src = dict(quote or {})
|
||
raw = src.get("fee_rows") or src.get("fee_lines") or src.get("feeItems") or []
|
||
if isinstance(raw, str) and raw.strip():
|
||
try:
|
||
raw = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
raw = []
|
||
if not isinstance(raw, list):
|
||
return []
|
||
rows: list[dict[str, Any]] = []
|
||
for item in raw:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
name = str(item.get("name") or item.get("label") or "").strip()
|
||
amount = str(item.get("amount") or item.get("value") or item.get("salePrice") or "").strip()
|
||
if name and amount:
|
||
row = dict(item)
|
||
row["name"] = name
|
||
row["amount"] = amount
|
||
rows.append(row)
|
||
return rows
|
||
|
||
|
||
def parse_money(raw: Any) -> tuple[Optional[float], str]:
|
||
"""抽出金额和币种前缀。读不了返回 (None, '')。"""
|
||
text = str(raw if raw not in (None, "") else "").strip()
|
||
if not text or text == "-":
|
||
return None, ""
|
||
ccy = ""
|
||
upper = text.upper().replace(",", "")
|
||
for prefix in ("USD", "CNY", "EUR", "HKD", "JPY", "GBP", "RMB"):
|
||
if upper.startswith(prefix):
|
||
ccy = "CNY" if prefix == "RMB" else prefix
|
||
text = text[len(prefix) :].strip()
|
||
break
|
||
cleaned = text.replace(",", "")
|
||
try:
|
||
return float(cleaned), ccy
|
||
except ValueError:
|
||
return None, ccy
|
||
|
||
|
||
def format_money(value: float, *, sample: str = "", currency: str = "") -> str:
|
||
"""金额写法跟原行对齐：有小数留两位，整数也写两位方便对账。"""
|
||
body = f"{value:.2f}"
|
||
code = (currency or "").strip().upper()
|
||
if code == "RMB":
|
||
code = "CNY"
|
||
sample_text = str(sample or "")
|
||
if code and (sample_text.upper().startswith(code) or sample_text.upper().startswith("RMB")):
|
||
return f"{code} {body}"
|
||
return body
|
||
|
||
|
||
def row_currency(row: dict[str, Any] | None, fallback: str = "") -> str:
|
||
src = dict(row or {})
|
||
code = str(src.get("currency") or src.get("currencyCode") or fallback or "").strip().upper()
|
||
if code == "RMB":
|
||
return "CNY"
|
||
if code:
|
||
return code
|
||
_, prefix = parse_money(src.get("amount"))
|
||
return prefix or (fallback or "").strip().upper().replace("RMB", "CNY")
|
||
|
||
|
||
def _apply_amount(row: dict[str, Any], amount: str) -> None:
|
||
row["amount"] = amount
|
||
row["unit_price"] = amount
|
||
for key in ("salePrice", "price", "value"):
|
||
if row.get(key) not in (None, ""):
|
||
row[key] = amount
|
||
|
||
|
||
def effective_quote(tms: dict[str, Any] | None, overlay: dict[str, Any] | None) -> dict[str, Any]:
|
||
"""
|
||
人工盖 TMS：同名用人工，没有的项保留 TMS，人工新增的补上。
|
||
|
||
合计先沿用 TMS，再按同币种差额加减。不做汇率。
|
||
"""
|
||
base = dict(tms or {})
|
||
rows = [dict(x) for x in quote_fee_rows(base)]
|
||
incoming = dict(overlay or {})
|
||
for item in quote_fee_rows(incoming):
|
||
name = str(item.get("name") or "").strip()
|
||
if not name:
|
||
continue
|
||
idx = next((i for i, r in enumerate(rows) if fee_name_same(name, str(r.get("name") or ""))), -1)
|
||
if idx < 0:
|
||
rows.append(dict(item))
|
||
continue
|
||
old = dict(rows[idx])
|
||
_apply_amount(old, str(item.get("amount") or ""))
|
||
if item.get("currency"):
|
||
old["currency"] = item.get("currency")
|
||
rows[idx] = old
|
||
out = dict(base)
|
||
out["fee_rows"] = rows
|
||
out["fee_lines"] = rows
|
||
header = str(out.get("currency") or incoming.get("currency") or "").strip()
|
||
base_total, _ = parse_money(base.get("total"))
|
||
if base_total is not None:
|
||
total_ccy = row_currency({"currency": header, "amount": base.get("total")}, header)
|
||
old_rows = quote_fee_rows(base)
|
||
delta = 0.0
|
||
for new_row in rows:
|
||
new_amt, _ = parse_money(new_row.get("amount"))
|
||
if new_amt is None:
|
||
continue
|
||
line_ccy = row_currency(new_row, total_ccy)
|
||
if line_ccy and total_ccy and line_ccy != total_ccy:
|
||
continue
|
||
idx = next(
|
||
(i for i, r in enumerate(old_rows) if fee_name_same(str(new_row.get("name") or ""), str(r.get("name") or ""))),
|
||
-1,
|
||
)
|
||
if idx < 0:
|
||
delta += new_amt
|
||
continue
|
||
old_amt, _ = parse_money(old_rows[idx].get("amount"))
|
||
if old_amt is None:
|
||
continue
|
||
delta += new_amt - old_amt
|
||
out["total"] = format_money(base_total + delta, sample=str(base.get("total") or ""), currency=header)
|
||
return out
|
||
|
||
|
||
def extract_adjust(
|
||
text: str,
|
||
*,
|
||
fee_names: list[str] | None = None,
|
||
injected: AdjustExtract | dict[str, Any] | None = None,
|
||
allow_model: Optional[bool] = None,
|
||
chat_fn: Any = None,
|
||
) -> AdjustExtract:
|
||
"""
|
||
从一句话抽出调价项。单测可注入；正式优先 DeepSeek B。
|
||
|
||
本地回退只给无网/模型失败用，不是正式识别。
|
||
allow_model=False 时禁止打模型。
|
||
"""
|
||
if injected is not None:
|
||
if isinstance(injected, AdjustExtract):
|
||
return injected
|
||
items = []
|
||
for raw in injected.get("items") or []:
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
items.append(
|
||
AdjustItem(
|
||
fee_name=str(raw.get("fee_name") or "").strip(),
|
||
kind=str(raw.get("kind") or "percent"),
|
||
direction=str(raw.get("direction") or "up"),
|
||
value=float(raw.get("value") or 0),
|
||
)
|
||
)
|
||
return AdjustExtract(
|
||
is_adjust=bool(injected.get("is_adjust")),
|
||
items=items,
|
||
wants_total=bool(injected.get("wants_total")),
|
||
)
|
||
raw = (text or "").strip()
|
||
if not raw:
|
||
return AdjustExtract(is_adjust=False)
|
||
names = [str(x).strip() for x in (fee_names or []) if str(x).strip()]
|
||
use_model = allow_model
|
||
if use_model is None:
|
||
try:
|
||
from agent.config import get_settings
|
||
|
||
settings = get_settings()
|
||
use_model = bool(
|
||
getattr(settings, "llm_allow_network", False)
|
||
and getattr(settings, "llm_data_usage_confirmed", False)
|
||
)
|
||
except Exception:
|
||
use_model = False
|
||
if use_model or chat_fn is not None:
|
||
try:
|
||
from agent.llm.mode_extract_adjust import invoke_extract_adjust
|
||
|
||
modeled = invoke_extract_adjust(
|
||
raw,
|
||
fee_names=names,
|
||
allow_network=bool(use_model),
|
||
chat_fn=chat_fn,
|
||
)
|
||
if modeled is not None:
|
||
if modeled.wants_total and not modeled.items:
|
||
op = _item_from_chunk(raw, "")
|
||
if op:
|
||
modeled.items = [op]
|
||
return modeled
|
||
except Exception:
|
||
logger.exception("调价模型抽取失败，回退本地")
|
||
names_sorted = sorted(names, key=len, reverse=True)
|
||
blob = raw.replace(" ", "")
|
||
total_hit = any(w in raw for w in _TOTAL_WORDS)
|
||
has_op = bool(_PERCENT_UP.search(raw) or _PERCENT_DOWN.search(raw) or _TARGET.search(raw))
|
||
fee_in_text = [n for n in names_sorted if n and n in raw]
|
||
if total_hit and has_op and not fee_in_text:
|
||
op = _item_from_chunk(raw, "")
|
||
return AdjustExtract(is_adjust=True, wants_total=True, items=[op] if op else [])
|
||
if _QUOTE_WORD in raw and has_op and not fee_in_text and not total_hit:
|
||
return AdjustExtract(is_adjust=True, wants_total=False)
|
||
items: list[AdjustItem] = []
|
||
for part in _SPLIT.split(raw) or [raw]:
|
||
chunk = (part or "").strip()
|
||
if not chunk:
|
||
continue
|
||
spoken = ""
|
||
for name in names_sorted:
|
||
if name and name in chunk:
|
||
spoken = name
|
||
break
|
||
if not spoken:
|
||
found = _FEE_TOKEN.findall(chunk)
|
||
spoken = found[-1] if found else ""
|
||
if not spoken:
|
||
continue
|
||
item = _item_from_chunk(chunk, spoken)
|
||
if item:
|
||
items.append(item)
|
||
if not items and has_op:
|
||
# 说了费用口语但和名单不完全一致：整句再扫一次
|
||
item = _item_from_chunk(raw, "")
|
||
if item:
|
||
items.append(item)
|
||
return AdjustExtract(is_adjust=bool(items or has_op), items=items)
|
||
|
||
|
||
def _item_from_chunk(chunk: str, fee_name: str) -> Optional[AdjustItem]:
|
||
up = _PERCENT_UP.search(chunk)
|
||
down = _PERCENT_DOWN.search(chunk)
|
||
target = _TARGET.search(chunk)
|
||
if target:
|
||
return AdjustItem(fee_name=fee_name, kind="target", direction="set", value=float(target.group(2)))
|
||
if up:
|
||
return AdjustItem(fee_name=fee_name, kind="percent", direction="up", value=float(up.group(2)))
|
||
if down:
|
||
return AdjustItem(fee_name=fee_name, kind="percent", direction="down", value=float(down.group(2)))
|
||
return None
|
||
|
||
|
||
def total_price_label(
|
||
business_line: str = "",
|
||
transport_mode: str = "",
|
||
quote: dict[str, Any] | None = None,
|
||
) -> str:
|
||
"""
|
||
销售可见总价名称。
|
||
|
||
空运一律总销售价。海运/陆运：TMS、产品报价是总成本价，销售调价后才是总销售价。
|
||
禁止写「合计」。Excel/PDF 栏目名不走这里。
|
||
"""
|
||
src = dict(quote or {})
|
||
raw = f"{business_line or ''} {transport_mode or ''} {src.get('business_line') or ''}"
|
||
key = raw.upper()
|
||
if key.strip() in {"AIR"} or "AIR" in key or "空运" in raw:
|
||
return "总销售价"
|
||
source = str(src.get("source") or "").strip().upper()
|
||
label = str(src.get("source_label") or src.get("sourceLabel") or "").replace(" ", "")
|
||
if source == "SALES_ADJUST" or "销售调价" in label:
|
||
return "总销售价"
|
||
return "总成本价"
|
||
|
||
|
||
def _mark_sales_adjust(quote: dict[str, Any]) -> None:
|
||
"""算价成功后这版就是销售调价，卡片和后台都按销售价展示。"""
|
||
quote["source"] = "SALES_ADJUST"
|
||
quote["source_label"] = "销售调价"
|
||
|
||
|
||
def compute_profit(*, x: Optional[float], baseline: str) -> Optional[float]:
|
||
"""
|
||
只在销售调价后算一个利润点。
|
||
|
||
基准=调价前生效价：有手工报价用手工，没有才用 TMS。
|
||
"""
|
||
if x is None:
|
||
return None
|
||
base, _ = parse_money(baseline)
|
||
if base is None or abs(base) < 1e-9:
|
||
return None
|
||
return (x - base) / base * 100.0
|
||
|
||
|
||
def _profit_baseline(
|
||
src: dict[str, Any],
|
||
*,
|
||
tms_cost: str,
|
||
tms_sale: str,
|
||
business_line: str,
|
||
) -> str:
|
||
"""第一次调价锁住基准；之后连续调仍对这一档。"""
|
||
stored = str(src.get("profit_baseline") or "").strip()
|
||
if stored:
|
||
return stored
|
||
current = str(src.get("total") or "").strip()
|
||
if current:
|
||
return current
|
||
line = (business_line or "").upper()
|
||
if line == "AIR":
|
||
return str(tms_sale or "").strip()
|
||
return str(tms_cost or tms_sale or "").strip()
|
||
|
||
|
||
def _attach_profit(
|
||
out: dict[str, Any],
|
||
src: dict[str, Any],
|
||
*,
|
||
x: Optional[float],
|
||
tms_cost: str,
|
||
tms_sale: str,
|
||
business_line: str,
|
||
) -> Optional[float]:
|
||
"""写入利润点和锁定基准。原报价没有利润点，销售没调过就不算。"""
|
||
baseline = _profit_baseline(src, tms_cost=tms_cost, tms_sale=tms_sale, business_line=business_line)
|
||
profit = compute_profit(x=x, baseline=baseline)
|
||
if profit is not None:
|
||
out["profit_point"] = profit
|
||
if baseline:
|
||
out.setdefault("profit_baseline", baseline)
|
||
line = (business_line or str(src.get("business_line") or "")).upper()
|
||
if line:
|
||
out["business_line"] = line
|
||
return profit
|
||
|
||
|
||
def _single_line_same_currency(rows: list[dict[str, Any]], total_ccy: str) -> Optional[int]:
|
||
"""
|
||
价格明细只有一项，且该项币种与总价一致：总价和明细绑在一起。
|
||
|
||
只看出项数和币种，不看出价是否相等。多项或币种不同：不算绑定。
|
||
"""
|
||
if len(rows) != 1:
|
||
return None
|
||
line_ccy = row_currency(rows[0], "")
|
||
if line_ccy and total_ccy and line_ccy != total_ccy:
|
||
return None
|
||
return 0
|
||
|
||
|
||
def _apply_total_only(
|
||
src: dict[str, Any],
|
||
extract: AdjustExtract,
|
||
*,
|
||
multi: bool,
|
||
total_ccy: str,
|
||
business_line: str,
|
||
tms_cost: str,
|
||
tms_sale: str,
|
||
) -> ApplyResult:
|
||
"""
|
||
改合计。多项或明细币种与总价不同：只改合计数字。
|
||
|
||
明细只有一项且币种与总价一致：明细跟着改，不再提醒分摊。
|
||
"""
|
||
item = next((x for x in extract.items if x is not None), None)
|
||
old_total, prefix = parse_money(src.get("total"))
|
||
if item is None or old_total is None:
|
||
return ApplyResult(ok=False, code="need_lines", quote=src, lines_hint=not multi, currency_hint=multi)
|
||
if item.kind == "target":
|
||
new_amt = float(item.value)
|
||
how = f"改到 {item.value:g}"
|
||
elif item.direction == "down":
|
||
new_amt = old_total * (1 - float(item.value) / 100.0)
|
||
how = f"下调 {item.value:g}%"
|
||
else:
|
||
new_amt = old_total * (1 + float(item.value) / 100.0)
|
||
how = f"上调 {item.value:g}%"
|
||
out = dict(src)
|
||
rows = [dict(x) for x in quote_fee_rows(src)]
|
||
mirrored = _single_line_same_currency(rows, total_ccy or prefix)
|
||
label = total_price_label(business_line, quote={"source": "SALES_ADJUST"})
|
||
applied = [{"name": label, "how": how, "amount": format_money(new_amt, sample=str(src.get("total") or ""), currency=total_ccy or prefix)}]
|
||
if mirrored is not None:
|
||
row = dict(rows[mirrored])
|
||
line_ccy = row_currency(row, total_ccy) or prefix
|
||
_apply_amount(row, format_money(new_amt, sample=str(row.get("amount") or ""), currency=line_ccy))
|
||
rows[mirrored] = row
|
||
applied.append({"name": str(row.get("name") or "明细"), "how": how, "amount": row["amount"]})
|
||
out["fee_rows"] = rows
|
||
out["fee_lines"] = list(rows)
|
||
out["total"] = format_money(new_amt, sample=str(src.get("total") or ""), currency=total_ccy or prefix)
|
||
profit = _attach_profit(
|
||
out,
|
||
src,
|
||
x=new_amt,
|
||
tms_cost=tms_cost or str(src.get("tms_baseline_cost") or ""),
|
||
tms_sale=tms_sale or str(src.get("tms_baseline_sale") or ""),
|
||
business_line=business_line,
|
||
)
|
||
_mark_sales_adjust(out)
|
||
return ApplyResult(
|
||
ok=True,
|
||
code="applied",
|
||
quote=out,
|
||
applied=applied,
|
||
currency_hint=multi,
|
||
lines_hint=bool((not multi) and mirrored is None),
|
||
profit_point=profit,
|
||
)
|
||
|
||
|
||
def apply_adjust(
|
||
quote: dict[str, Any] | None,
|
||
extract: AdjustExtract,
|
||
*,
|
||
business_line: str = "",
|
||
tms_cost: str = "",
|
||
tms_sale: str = "",
|
||
) -> ApplyResult:
|
||
"""
|
||
按抽项改当前生效报价副本。
|
||
|
||
明细只有一项且币种与总价一致：改总价或改明细都一起调。
|
||
多项同币种改合计：只改合计数字。多币种：只改点名的那一格。
|
||
副作用：无。原 quote 不被修改。
|
||
"""
|
||
src = dict(quote or {})
|
||
rows = [dict(x) for x in quote_fee_rows(src)]
|
||
if not extract.is_adjust:
|
||
return ApplyResult(ok=False, code="not_adjust", quote=src)
|
||
header = str(src.get("currency") or "").strip()
|
||
total_ccy = row_currency({"currency": header, "amount": src.get("total")}, header)
|
||
currencies = {row_currency(r, header) for r in rows if row_currency(r, header)}
|
||
if total_ccy:
|
||
currencies.add(total_ccy)
|
||
multi = len(currencies) > 1
|
||
if extract.wants_total:
|
||
return _apply_total_only(src, extract, multi=multi, total_ccy=total_ccy, business_line=business_line, tms_cost=tms_cost, tms_sale=tms_sale)
|
||
if not extract.items:
|
||
return ApplyResult(ok=False, code="need_fee", quote=src)
|
||
applied: list[dict[str, Any]] = []
|
||
unmatched: list[str] = []
|
||
ambiguous: list[str] = []
|
||
currency_hint = bool(multi)
|
||
total_amt, _ = parse_money(src.get("total"))
|
||
new_total = total_amt
|
||
single_same_ccy = _single_line_same_currency(rows, total_ccy or header) is not None
|
||
|
||
for item in extract.items:
|
||
spoken = (item.fee_name or "").strip()
|
||
if not spoken:
|
||
unmatched.append(spoken or "未点名费用")
|
||
continue
|
||
hits = match_fee_names(spoken, rows)
|
||
if len(hits) > 1:
|
||
ambiguous.append(spoken)
|
||
continue
|
||
if not hits:
|
||
unmatched.append(spoken)
|
||
continue
|
||
idx = next(i for i, r in enumerate(rows) if str(r.get("name") or "") == hits[0])
|
||
row = dict(rows[idx])
|
||
old_amt, prefix = parse_money(row.get("amount"))
|
||
if old_amt is None:
|
||
unmatched.append(spoken)
|
||
continue
|
||
line_ccy = row_currency(row, header) or prefix
|
||
if item.kind == "target":
|
||
new_amt = float(item.value)
|
||
how = f"改到 {item.value:g}"
|
||
elif item.direction == "down":
|
||
new_amt = old_amt * (1 - float(item.value) / 100.0)
|
||
how = f"下调 {item.value:g}%"
|
||
else:
|
||
new_amt = old_amt * (1 + float(item.value) / 100.0)
|
||
how = f"上调 {item.value:g}%"
|
||
_apply_amount(row, format_money(new_amt, sample=str(row.get("amount") or ""), currency=line_ccy or prefix))
|
||
rows[idx] = row
|
||
applied.append({"name": hits[0], "how": how, "amount": row["amount"]})
|
||
if (not multi) and new_total is not None:
|
||
if single_same_ccy:
|
||
new_total = new_amt
|
||
else:
|
||
new_total = new_total + (new_amt - old_amt)
|
||
|
||
out = dict(src)
|
||
out["fee_rows"] = rows
|
||
out["fee_lines"] = rows
|
||
if applied and (not multi) and new_total is not None and new_total != total_amt:
|
||
out["total"] = format_money(new_total, sample=str(src.get("total") or ""), currency=total_ccy)
|
||
if not applied:
|
||
code = "ambiguous" if ambiguous else "none_matched"
|
||
return ApplyResult(
|
||
ok=False,
|
||
code=code,
|
||
quote=src,
|
||
unmatched=unmatched,
|
||
ambiguous=ambiguous,
|
||
)
|
||
sale_now, _ = parse_money(out.get("total"))
|
||
profit = _attach_profit(
|
||
out,
|
||
src,
|
||
x=sale_now,
|
||
tms_cost=tms_cost or str(src.get("tms_baseline_cost") or ""),
|
||
tms_sale=tms_sale or str(src.get("tms_baseline_sale") or ""),
|
||
business_line=business_line,
|
||
)
|
||
if tms_cost and "tms_baseline_cost" not in out:
|
||
out["tms_baseline_cost"] = tms_cost
|
||
if tms_sale and "tms_baseline_sale" not in out:
|
||
out["tms_baseline_sale"] = tms_sale
|
||
_mark_sales_adjust(out)
|
||
logger.info(
|
||
"quote_adjust applied=%s unmatched=%s hint=%s profit=%s",
|
||
[x.get("name") for x in applied],
|
||
unmatched,
|
||
currency_hint,
|
||
profit,
|
||
)
|
||
return ApplyResult(
|
||
ok=True,
|
||
code="applied",
|
||
quote=out,
|
||
applied=applied,
|
||
unmatched=unmatched,
|
||
ambiguous=ambiguous,
|
||
currency_hint=currency_hint,
|
||
profit_point=profit,
|
||
)
|