Spaces:
Runtime error
Runtime error
Download src/parser.py from BinbomJared/DM-Agent: direct link, hf CLI and curl.
- Browser
- Download file 7.58 kB
-
https://huggingface.co/spaces/BinbomJared/DM-Agent/resolve/main/src/parser.py
- Command line
-
hf download hf://spaces/BinbomJared/DM-Agent/src/parser.py
-
curl -L -o parser.py https://huggingface.co/spaces/BinbomJared/DM-Agent/resolve/main/src/parser.py
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周** — 指定周 | |
| - **帮助** — 显示本说明""" | |