feat: expense tracker Python parity (Wave 10)
New `expenses_handlers.py`: - parse_expense(phrase) -> (amount, category) | None — pure fn, fully unit-tested. Handles "потратил 500 на еду", comma decimals "12,50", EN "spent 12.5 on coffee", default category "разное". - do_expenses_log: append entry to expenses.json (atomic write-then-rename). - do_expenses_today: sum today's entries (local-day boundary). - do_expenses_week: rolling 7-day total + count. - do_expenses_breakdown: per-category, top-5 + tail collapse. extensions.py / main.py / commands.yaml: 4 new dispatch entries. .gitignore: + expenses.json. tests/test_expenses_handlers.py: 8 unit tests covering parse, log, today/week math, breakdown ordering + tail collapse. Tests: 96 → 104 (+8). Total Python coverage now mirrors Rust pack behaviour for personal finance.
This commit is contained in:
parent
b3cfaa5171
commit
f102b8dc95
6 changed files with 343 additions and 0 deletions
177
expenses_handlers.py
Normal file
177
expenses_handlers.py
Normal file
|
|
@ -0,0 +1,177 @@
|
|||
"""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}.")
|
||||
Loading…
Add table
Add a link
Reference in a new issue