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