DM-Agent / src /parser.py
unknown
refactor: consolidate repo structure, remove upload_to_github duplication
3291500
Raw History Blame Contribute Delete
7.58 kB
"""
解析用户自然语言时间指令,转为分析参数。
支持的表达式:
"最新一周" / "最新周报" → {"mode": "latest"}
"上周" / "上一周" → {"mode": "prev"}
"W9" / "第9周" → {"mode": "week", "year": current, "week": 9}
"2025 W40" / "2025年第40周" → {"mode": "week", "year": 2025, "week": 40}
"25年12月" / "2025年12月" → {"mode": "month", "year": 2025, "month": 12}
"12月" / "上个月" → {"mode": "month", "year": inferred, "month": 12}
"昨天" / "今天" → {"mode": "day", "date": "2026-03-03"}
"3月3日" / "2026-03-03" → {"mode": "day", "date": "2026-03-03"}
"1.15~3.3" / "1月15日~3月3日" → {"mode": "range", "start": "2026-01-15", "end": "2026-03-03"}
"""
import re
from datetime import datetime, timedelta
from typing import Any
def _current_year() -> int:
return datetime.now().year
def _full_year(y: int) -> int:
"""两位年份 → 四位。"""
if y < 100:
return 2000 + y
return y
def _clean_token(token: str) -> str:
"""去掉日期 token 中常见的中文前缀/后缀干扰词。"""
token = token.strip()
token = re.sub(r"^(分析|查看|看看|帮我看|查一下|看一下|分析一下|对比)\s*", "", token)
token = re.sub(r"(的数据|数据|表现|报告|怎么样|如何)$", "", token)
return token.strip()
def _parse_date_token(token: str) -> str | None:
"""将各种日期片段解析为 YYYY-MM-DD 字符串,无法识别返回 None。"""
token = _clean_token(token)
# ISO: 2026-01-15 / 2026/1/15
m = re.search(r"(\d{4})\s*[\-/]\s*(\d{1,2})\s*[\-/]\s*(\d{1,2})", token)
if m:
y, mo, d = int(m.group(1)), int(m.group(2)), int(m.group(3))
return f"{y:04d}-{mo:02d}-{d:02d}"
# 中文完整: 26年1月15日 / 2026年1月15日
m = re.search(r"(\d{2,4})\s*年\s*(\d{1,2})\s*月\s*(\d{1,2})\s*[日号]?", token)
if m:
y = _full_year(int(m.group(1)))
mo, d = int(m.group(2)), int(m.group(3))
return f"{y:04d}-{mo:02d}-{d:02d}"
# 中文无年: 1月15日
m = re.search(r"(\d{1,2})\s*月\s*(\d{1,2})\s*[日号]", token)
if m:
mo, d = int(m.group(1)), int(m.group(2))
y = _current_year()
return f"{y:04d}-{mo:02d}-{d:02d}"
# 简写: 1.15 / 01.15(整个 token 只剩数字和点)
m = re.match(r"^(\d{1,2})\s*\.\s*(\d{1,2})$", token)
if m:
mo, d = int(m.group(1)), int(m.group(2))
y = _current_year()
return f"{y:04d}-{mo:02d}-{d:02d}"
return None
def _try_parse_range(text: str) -> dict[str, Any] | None:
"""尝试解析自定义时间段。支持 ~ - 到 至 作为分隔符。"""
for sep in [r"[~~—]", r"\s*到\s*", r"\s*至\s*"]:
parts = re.split(sep, text, maxsplit=1)
if len(parts) == 2:
start = _parse_date_token(parts[0])
end = _parse_date_token(parts[1])
if start and end:
return {"mode": "range", "start": start, "end": end}
return None
def parse_query(text: str) -> dict[str, Any] | None:
"""
解析用户输入,返回分析参数 dict,或 None(无法识别)。
"""
text = text.strip()
# 去掉常见指令前缀,留下时间描述
cleaned = re.sub(r"^(分析|查看|看看|帮我看|帮我分析|查一下|看一下|分析一下|对比)\s*", "", text).strip()
# ── 自定义时间段(优先级最高,避免被单日匹配吃掉)──
range_result = _try_parse_range(cleaned)
if not range_result:
range_result = _try_parse_range(text)
if range_result:
return range_result
# ── 昨天 / 前天 / 今天 ──
if re.search(r"昨天|昨日", text):
d = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
return {"mode": "day", "date": d}
if re.search(r"前天|前日", text):
d = (datetime.now() - timedelta(days=2)).strftime("%Y-%m-%d")
return {"mode": "day", "date": d}
if re.search(r"今天|今日", text):
d = datetime.now().strftime("%Y-%m-%d")
return {"mode": "day", "date": d}
# ── 指定日期:2026-03-03 / 2026/3/3 ──
dm = re.search(r"(\d{4})\s*[\-/]\s*(\d{1,2})\s*[\-/]\s*(\d{1,2})", text)
if dm:
y, mo, day = int(dm.group(1)), int(dm.group(2)), int(dm.group(3))
return {"mode": "day", "date": f"{y:04d}-{mo:02d}-{day:02d}"}
# ── 中文日期:3月3日 / 26年3月3日 ──
dm = re.search(r"(?:(\d{2,4})\s*年\s*)?(\d{1,2})\s*月\s*(\d{1,2})\s*[日号]", text)
if dm:
y = _full_year(int(dm.group(1))) if dm.group(1) else _current_year()
mo, day = int(dm.group(2)), int(dm.group(3))
return {"mode": "day", "date": f"{y:04d}-{mo:02d}-{day:02d}"}
# ── 最新一周 ──
if re.search(r"(最新|本周|这周|当前|latest)", text, re.IGNORECASE):
return {"mode": "latest"}
# ── 上一周 ──
if re.search(r"(上周|上一周|last\s*week)", text, re.IGNORECASE):
return {"mode": "prev"}
# ── 指定年+周:2025 W40 / 2025年第40周 / 25年W40 ──
m = re.search(r"(\d{2,4})\s*年?\s*[Ww第]?\s*(\d{1,2})\s*周?", text)
if m:
first, second = int(m.group(1)), int(m.group(2))
# 区分 "25年12月" vs "25年W12"
if re.search(r"[Ww第]", text) and not re.search(r"月", text):
return {"mode": "week", "year": _full_year(first), "week": second}
# ── 指定年+月:25年12月 / 2025年12月 / 2025-12 ──
m = re.search(r"(\d{2,4})\s*[年\-/]\s*(\d{1,2})\s*月?", text)
if m:
y, mo = int(m.group(1)), int(m.group(2))
if 1 <= mo <= 12:
return {"mode": "month", "year": _full_year(y), "month": mo}
# ── 仅月份:12月 / 上个月 ──
if re.search(r"上个?月", text):
now = datetime.now()
mo = now.month - 1
y = now.year
if mo < 1:
mo = 12
y -= 1
return {"mode": "month", "year": y, "month": mo}
m = re.search(r"(\d{1,2})\s*月", text)
if m:
mo = int(m.group(1))
if 1 <= mo <= 12:
y = _current_year()
if mo > datetime.now().month:
y -= 1
return {"mode": "month", "year": y, "month": mo}
# ── 仅周号:W9 / 第9周 ──
m = re.search(r"[Ww第]\s*(\d{1,2})\s*周?", text)
if m:
wk = int(m.group(1))
return {"mode": "week", "year": _current_year(), "week": wk}
return None
def describe_query(params: dict[str, Any]) -> str:
"""将解析结果转为可读描述。"""
mode = params.get("mode")
if mode == "latest":
return "最新一周"
elif mode == "prev":
return "上一周"
elif mode == "week":
return f"{params['year']}年 第{params['week']}周"
elif mode == "month":
return f"{params['year']}年{params['month']}月"
elif mode == "day":
return params["date"]
elif mode == "range":
return f"{params['start']} ~ {params['end']}"
return "未知"
HELP_TEXT = """你可以 @我 发送以下指令:
- **分析最新一周** — 最新周度报告
- **分析上周** — 上一周报告
- **分析昨天** — 昨日日报
- **分析3月3日** — 指定日期
- **分析1.15~3.3** — 自定义时间段
- **分析25年12月** — 指定月份汇总
- **分析W9** 或 **分析第9周** — 指定周
- **帮助** — 显示本说明"""