137 lines
4.2 KiB
Python
137 lines
4.2 KiB
Python
"""风控预警自然语言筛选条件解析。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from datetime import UTC, date, datetime, time, timedelta
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from app.core.risk_contracts import RiskAlertQuery
|
||
|
||
SHANGHAI = ZoneInfo("Asia/Shanghai")
|
||
|
||
|
||
def parse_risk_alert_filters(
|
||
message: str,
|
||
*,
|
||
now: datetime | None = None,
|
||
) -> dict[str, str]:
|
||
"""把常见中文筛选表达转换为风控查询工具参数。"""
|
||
current = now or datetime.now(SHANGHAI)
|
||
if current.tzinfo is None:
|
||
current = current.replace(tzinfo=SHANGHAI)
|
||
|
||
filters: dict[str, str] = {}
|
||
level = re.search(r"(低|中|高)(?:风险)?", message)
|
||
if level:
|
||
filters["risk_level"] = level.group(1)
|
||
|
||
rule = re.search(r"RW-\d{3}", message, re.IGNORECASE)
|
||
if rule:
|
||
filters["rule_code"] = rule.group(0).upper()
|
||
|
||
customer = re.search(
|
||
r"(?:客户编号|客户号|客户账户|客户)\s*[:：#]?\s*"
|
||
r"([A-Za-z0-9_-]{1,64})",
|
||
message,
|
||
)
|
||
if customer:
|
||
filters["customer_no"] = customer.group(1)
|
||
elif matched := re.search(r"\b(CUST-\d+)\b", message, re.IGNORECASE):
|
||
filters["customer_no"] = matched.group(1).upper()
|
||
|
||
product_code = re.search(
|
||
r"(?:产品编号|产品代码|产品编码)\s*[:：#]?\s*"
|
||
r"([A-Za-z0-9_-]{1,64})",
|
||
message,
|
||
)
|
||
if product_code:
|
||
filters["product_code"] = product_code.group(1)
|
||
elif matched := re.search(r"\b(P-[A-Za-z0-9_-]{1,60})\b", message):
|
||
filters["product_code"] = matched.group(1)
|
||
|
||
product_name = re.search(
|
||
r"产品名称\s*[:：#]?\s*([^\s，,。；;？！]{1,128})",
|
||
message,
|
||
)
|
||
if product_name is None:
|
||
product_name = re.search(
|
||
r"产品\s*[:：]\s*([^\s，,。；;？！]{1,128})",
|
||
message,
|
||
)
|
||
if product_name:
|
||
candidate = product_name.group(1).strip()
|
||
if candidate not in {"编号", "代码", "编码", "名称"}:
|
||
filters["product_name"] = candidate
|
||
|
||
start_time, end_time = _parse_time_range(message, current)
|
||
if start_time is not None:
|
||
filters["start_time"] = _to_utc_iso(start_time)
|
||
if end_time is not None:
|
||
filters["end_time"] = _to_utc_iso(end_time)
|
||
|
||
validated = RiskAlertQuery.model_validate(filters)
|
||
return validated.model_dump(mode="json", exclude_none=True)
|
||
|
||
|
||
def _parse_time_range(
|
||
message: str,
|
||
current: datetime,
|
||
) -> tuple[datetime | None, datetime | None]:
|
||
explicit = re.search(
|
||
r"(\d{4}-\d{2}-\d{2})\s*(?:至|到|~|—)\s*(\d{4}-\d{2}-\d{2})",
|
||
message,
|
||
)
|
||
if explicit:
|
||
start_date = _parse_date(explicit.group(1))
|
||
end_date = _parse_date(explicit.group(2))
|
||
if start_date is not None and end_date is not None:
|
||
return (
|
||
datetime.combine(start_date, time.min, tzinfo=SHANGHAI),
|
||
datetime.combine(end_date, time.max, tzinfo=SHANGHAI),
|
||
)
|
||
|
||
if "今天" in message or "今日" in message:
|
||
start = current.replace(hour=0, minute=0, second=0, microsecond=0)
|
||
return start, current
|
||
|
||
if "本月" in message:
|
||
start = current.replace(
|
||
day=1,
|
||
hour=0,
|
||
minute=0,
|
||
second=0,
|
||
microsecond=0,
|
||
)
|
||
return start, current
|
||
|
||
recent = re.search(r"(?:近|最近)\s*(\d+)\s*(天|日)", message)
|
||
if recent:
|
||
days = int(recent.group(1))
|
||
return current - timedelta(days=days), current
|
||
|
||
since = re.search(r"(\d{4}-\d{2}-\d{2})\s*(?:以来|之后|起)", message)
|
||
if since:
|
||
start_date = _parse_date(since.group(1))
|
||
if start_date is not None:
|
||
return datetime.combine(start_date, time.min, tzinfo=SHANGHAI), current
|
||
|
||
until = re.search(r"(?:截至|截止)\s*(\d{4}-\d{2}-\d{2})", message)
|
||
if until:
|
||
end_date = _parse_date(until.group(1))
|
||
if end_date is not None:
|
||
return None, datetime.combine(end_date, time.max, tzinfo=SHANGHAI)
|
||
|
||
return None, None
|
||
|
||
|
||
def _to_utc_iso(value: datetime) -> str:
|
||
return value.astimezone(UTC).replace(tzinfo=None).isoformat()
|
||
|
||
|
||
def _parse_date(value: str) -> date | None:
|
||
try:
|
||
return date.fromisoformat(value)
|
||
except ValueError:
|
||
return None
|