Bot & Automation
Activity Reminder Bot
/root/hermes-projects/Activity Reminder Bot
packages/reminder_core/parsers/natural_datetime.py
text
import re
from dataclasses import dataclass
from datetime import datetime, time, timedelta
from activity_reminder_bot.models import ReminderRepeat
from dateparser.search import search_dates
WEEKDAY_ALIASES = {
"senin": 0,
"selasa": 1,
"rabu": 2,
"kamis": 3,
"jumat": 4,
"jum'at": 4,
"sabtu": 5,
"minggu": 6,
}
REPEAT_PATTERNS = [
(re.compile(r"\b(setiap hari|harian|daily)\b", re.IGNORECASE), ReminderRepeat.daily),
(re.compile(r"\b(setiap minggu|mingguan|weekly)\b", re.IGNORECASE), ReminderRepeat.weekly),
(re.compile(r"\b(setiap bulan|bulanan|monthly)\b", re.IGNORECASE), ReminderRepeat.monthly),
]
@dataclass(frozen=True)
class ParsedReminder:
activity_name: str
description: str
due_at: datetime
repeat: ReminderRepeat
class ReminderTextParser:
def __init__(self, timezone_name: str = "Asia/Jakarta") -> None:
self.timezone_name = timezone_name
def parse(self, text: str, now: datetime | None = None) -> ParsedReminder:
clean_text = " ".join(text.strip().split())
if not clean_text:
raise ValueError("Tulis kegiatan dan waktu reminder terlebih dahulu.")
base_now = now or datetime.now()
repeat = self._detect_repeat(clean_text)
due_at = self._parse_datetime(clean_text, base_now, repeat)
if due_at <= base_now:
raise ValueError("Tanggal dan jam reminder harus di masa depan.")
activity_name = self._extract_activity_name(clean_text)
if len(activity_name) < 3:
raise ValueError("Nama kegiatan terlalu pendek. Contoh: Besok jam 08:00 meeting klien.")
return ParsedReminder(activity_name=activity_name, description=clean_text, due_at=due_at, repeat=repeat)
def _detect_repeat(self, text: str) -> ReminderRepeat:
for pattern, repeat in REPEAT_PATTERNS:
if pattern.search(text):
return repeat
if re.search(r"\bsetiap\s+(senin|selasa|rabu|kamis|jumat|jum'at|sabtu|minggu)\b", text, re.IGNORECASE):
return ReminderRepeat.weekly
return ReminderRepeat.none
def _parse_datetime(self, text: str, now: datetime, repeat: ReminderRepeat) -> datetime:
weekday_due = self._parse_weekday_repeat(text, now)
time_hint = self._extract_time(text)
if weekday_due and time_hint:
return datetime.combine(weekday_due.date(), time_hint)
day_of_month_due = self._parse_day_of_month(text, now, time_hint)
if day_of_month_due:
return day_of_month_due
settings = {
"PREFER_DATES_FROM": "future",
"RELATIVE_BASE": now,
"TIMEZONE": self.timezone_name,
"RETURN_AS_TIMEZONE_AWARE": False,
}
found = search_dates(text, languages=["id", "en"], settings=settings) or []
if found:
parsed = max((candidate for _, candidate in found), key=lambda item: self._score_datetime(item, now))
parsed = self._merge_time_hint(parsed, time_hint)
if parsed.date() == now.date() and parsed.time() == time.min and time_hint:
parsed = datetime.combine(parsed.date(), time_hint)
return parsed
if time_hint:
candidate = datetime.combine(now.date(), time_hint)
if candidate <= now:
candidate += timedelta(days=1)
if repeat == ReminderRepeat.daily:
return candidate
raise ValueError("Saya belum bisa membaca tanggal/jam. Contoh: Besok jam 08:00 meeting klien.")
def _parse_weekday_repeat(self, text: str, now: datetime) -> datetime | None:
match = re.search(r"\b(?:setiap\s+)?(senin|selasa|rabu|kamis|jumat|jum'at|sabtu|minggu)\b", text, re.IGNORECASE)
if not match:
return None
weekday = WEEKDAY_ALIASES[match.group(1).lower()]
days_ahead = (weekday - now.weekday()) % 7
if days_ahead == 0:
days_ahead = 7
return now + timedelta(days=days_ahead)
def _parse_day_of_month(self, text: str, now: datetime, time_hint: time | None) -> datetime | None:
match = re.search(r"\btanggal\s+(\d{1,2})\b", text, re.IGNORECASE)
if not match:
return None
day = int(match.group(1))
hour_time = time_hint or time(hour=9)
month = now.month
year = now.year
for _ in range(14):
try:
candidate = datetime(year, month, day, hour_time.hour, hour_time.minute)
except ValueError:
month += 1
if month > 12:
month = 1
year += 1
continue
if candidate > now:
return candidate
month += 1
if month > 12:
month = 1
year += 1
return None
def _extract_time(self, text: str) -> time | None:
match = re.search(r"\b(?:jam|pukul)?\s*(\d{1,2})(?:[:.](\d{2}))\b", text, re.IGNORECASE)
if not match:
return None
hour = int(match.group(1))
minute = int(match.group(2))
if hour > 23 or minute > 59:
raise ValueError("Format jam tidak valid.")
return time(hour=hour, minute=minute)
def _merge_time_hint(self, parsed: datetime, time_hint: time | None) -> datetime:
if not time_hint:
return parsed
return parsed.replace(hour=time_hint.hour, minute=time_hint.minute, second=0, microsecond=0)
def _score_datetime(self, parsed: datetime, now: datetime) -> float:
delta = abs((parsed - now).total_seconds())
has_explicit_time = parsed.hour != 0 or parsed.minute != 0
return delta + (0 if has_explicit_time else 1_000_000)
def _extract_activity_name(self, text: str) -> str:
result = text
cleanup_patterns = [
r"\b(besok|lusa|hari ini|nanti|tanggal\s+\d{1,2})\b",
r"\b(setiap hari|setiap minggu|setiap bulan|harian|mingguan|bulanan)\b",
r"\bsetiap\s+(senin|selasa|rabu|kamis|jumat|jum'at|sabtu|minggu)\b",
r"\b(jam|pukul)\s*\d{1,2}[:.]\d{2}\b",
r"\b\d{1,2}[:.]\d{2}\b",
]
for pattern in cleanup_patterns:
result = re.sub(pattern, " ", result, flags=re.IGNORECASE)
result = re.sub(r"\s+", " ", result).strip(" ,-")
return result[:1].upper() + result[1:] if result else text