J.A.R.V.I.S-py/expenses_handlers.py

"""Personal finance tracker — Python parity for the Rust `expenses/` pack.

Stores entries in `<here>/expenses.json` (atomic write-through), exposes
the same four voice entry points as the Rust pack:

    do_expenses_log         "потратил 500 на еду"
    do_expenses_today       "сколько потратил сегодня"
    do_expenses_week        "сколько на неделе"
    do_expenses_breakdown   "куда ушли деньги"

State shape: { "entries": [{"ts": int, "amount": float, "category": str}, ...] }
"""

import json
import os
import re
import time as _time
from typing import Callable, Optional

_HERE = os.path.dirname(os.path.abspath(__file__))
_PATH = os.path.join(_HERE, 'expenses.json')

_speak: Optional[Callable[[str], None]] = None


def set_speak(fn: Callable[[str], None]) -> None:
    global _speak
    _speak = fn


def _say(text: str) -> None:
    if _speak is not None:
        _speak(text)
    else:
        print(f"[expenses] {text}")


def load() -> dict:
    """Return current state. Empty-shape dict if file missing or unreadable."""
    if not os.path.isfile(_PATH):
        return {'entries': []}
    try:
        with open(_PATH, encoding='utf-8') as f:
            data = json.load(f)
        if not isinstance(data, dict) or not isinstance(data.get('entries'), list):
            return {'entries': []}
        return data
    except (OSError, json.JSONDecodeError):
        return {'entries': []}


def _save(data: dict) -> None:
    tmp = _PATH + '.tmp'
    try:
        with open(tmp, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        os.replace(tmp, _PATH)
    except OSError as e:
        print(f"[expenses] save failed: {e}")


# Pulled out so tests can exercise parsing without going through TTS.
_TRIGGERS = [
    "потратил", "записать трату", "запиши трату", "купил за", "заплатил за",
    "spent", "log expense", "paid for", "bought",
]


def _strip_trigger(phrase: str) -> str:
    p = (phrase or "").lower().strip()
    for t in _TRIGGERS:
        if p.startswith(t):
            return p[len(t):].strip()
    return p


def parse_expense(phrase: str):
    """Return (amount: float, category: str) or None if the phrase doesn't
    contain a parseable expense. Pure function — used by both `do_expenses_log`
    and the test suite."""
    rest = _strip_trigger(phrase)
    m = re.match(r"(-?\d+[.,]?\d*)\s*(.*)", rest)
    if not m:
        return None
    amount_str = m.group(1).replace(",", ".")
    try:
        amount = float(amount_str)
    except ValueError:
        return None
    if amount <= 0:
        return None
    after = m.group(2) or ""
    category = "разное"
    # "на X" (RU) / "for X" / "on X" (EN)
    cat_match = re.search(r"\bна\s+(\S+)", after) \
                or re.search(r"\bfor\s+(\S+)", after) \
                or re.search(r"\bon\s+(\S+)", after)
    if cat_match:
        category = cat_match.group(1).strip(".,!?")
    return amount, category


def do_expenses_log(action=None, voice=None) -> None:
    parsed = parse_expense(voice or "")
    if parsed is None:
        _say("Не понял сумму, сэр.")
        return
    amount, category = parsed
    data = load()
    data['entries'].append({
        'ts': int(_time.time()),
        'amount': amount,
        'category': category,
    })
    _save(data)
    _say(f"Записал: {amount:g} на {category}.")


def _today_bounds() -> tuple[int, int]:
    now = _time.localtime()
    start = _time.mktime((now.tm_year, now.tm_mon, now.tm_mday,
                          0, 0, 0, 0, 0, -1))
    return int(start), int(start) + 86400


def do_expenses_today(action=None, voice=None) -> None:
    data = load()
    if not data['entries']:
        _say("Сегодня ничего не тратили, сэр.")
        return
    start, end = _today_bounds()
    total = sum(e['amount'] for e in data['entries']
                if isinstance(e.get('ts'), int) and start <= e['ts'] < end)
    if total == 0:
        _say("Сегодня ничего не тратили, сэр.")
    else:
        _say(f"Сегодня потратили {total:g}.")


def do_expenses_week(action=None, voice=None) -> None:
    data = load()
    if not data['entries']:
        _say("На этой неделе ничего не тратили.")
        return
    cutoff = int(_time.time()) - 7 * 86400
    rows = [e for e in data['entries']
            if isinstance(e.get('ts'), int) and e['ts'] >= cutoff]
    if not rows:
        _say("На этой неделе тратами не похваляешься, сэр.")
        return
    total = sum(e['amount'] for e in rows)
    _say(f"За неделю {len(rows)} трат на сумму {total:g}.")


def do_expenses_breakdown(action=None, voice=None) -> None:
    data = load()
    if not data['entries']:
        _say("Нечего разбивать, сэр.")
        return
    cutoff = int(_time.time()) - 7 * 86400
    by_cat: dict[str, float] = {}
    for e in data['entries']:
        if isinstance(e.get('ts'), int) and e['ts'] >= cutoff:
            cat = e.get('category', 'разное')
            by_cat[cat] = by_cat.get(cat, 0.0) + e['amount']
    if not by_cat:
        _say("На этой неделе тишина по тратам, сэр.")
        return
    rows = sorted(by_cat.items(), key=lambda kv: kv[1], reverse=True)
    top = rows[:5]
    rest = rows[5:]
    parts = [f"{cat} {amt:g}" for cat, amt in top]
    tail = ""
    if rest:
        rest_total = sum(amt for _, amt in rest)
        tail = f" и другое {rest_total:g}"
    _say(f"По категориям: {', '.join(parts)}{tail}.")