83 lines
2.1 KiB
Python
83 lines
2.1 KiB
Python
import json
|
|
import os
|
|
from datetime import date
|
|
|
|
DATA_DIR = os.environ.get("DATA_DIR", os.path.join(os.path.dirname(__file__), "data"))
|
|
INCOME_FILE = os.path.join(DATA_DIR, "income_log.json")
|
|
|
|
INCOME_STREAMS = [
|
|
{"name": "Auto King", "owner": "Tony", "type": "paycheck"},
|
|
{"name": "Studio", "owner": "Bonna", "type": "business"},
|
|
{"name": "Bakery", "owner": "Bonna", "type": "business"},
|
|
{"name": "Car Flipping","owner": "Tony", "type": "business"},
|
|
{"name": "Misc", "owner": "Both", "type": "misc"},
|
|
]
|
|
|
|
OWNER_COLORS = {
|
|
"Bonna": "#D4B8C5",
|
|
"Tony": "#A0B8C5",
|
|
"Both": "#B8A8D4",
|
|
}
|
|
|
|
|
|
def load_income_log():
|
|
if os.path.exists(INCOME_FILE):
|
|
with open(INCOME_FILE) as f:
|
|
return json.load(f)
|
|
return []
|
|
|
|
|
|
def save_income_log(log):
|
|
with open(INCOME_FILE, "w") as f:
|
|
json.dump(log, f, indent=2)
|
|
|
|
|
|
def month_key(year, month):
|
|
return f"{year}-{month:02d}"
|
|
|
|
|
|
def income_by_month(year, month):
|
|
log = load_income_log()
|
|
mk = month_key(year, month)
|
|
entries = [e for e in log if e.get("month") == mk]
|
|
by_stream = {}
|
|
for e in entries:
|
|
s = e["stream"]
|
|
by_stream[s] = by_stream.get(s, 0) + float(e.get("amount", 0))
|
|
return by_stream, entries
|
|
|
|
|
|
def income_ytd(year):
|
|
log = load_income_log()
|
|
by_stream = {}
|
|
by_month = {}
|
|
for e in log:
|
|
mk = e.get("month", "")
|
|
if not mk.startswith(str(year)):
|
|
continue
|
|
s = e["stream"]
|
|
amt = float(e.get("amount", 0))
|
|
by_stream[s] = by_stream.get(s, 0) + amt
|
|
by_month[mk] = by_month.get(mk, 0) + amt
|
|
return by_stream, by_month
|
|
|
|
|
|
def log_income_entry(stream, amount, month_str, note="", entry_id=None):
|
|
import uuid
|
|
log = load_income_log()
|
|
log.append({
|
|
"id": entry_id or str(uuid.uuid4()),
|
|
"stream": stream,
|
|
"amount": float(amount),
|
|
"month": month_str,
|
|
"date": date.today().isoformat(),
|
|
"note": note,
|
|
})
|
|
save_income_log(log)
|
|
|
|
|
|
def delete_income_entry(entry_id):
|
|
log = load_income_log()
|
|
log = [e for e in log if e.get("id") != entry_id]
|
|
save_income_log(log)
|