Spaces:
Runtime error
Runtime error
File size: 5,119 Bytes
2a7171f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | """Thread-safe expense ledger with HuggingFace Hub CSV persistence."""
import os
import logging
import threading
import pandas as pd
from pathlib import Path
logger = logging.getLogger(__name__)
COLUMNS = ["Date", "Description", "Category", "Amount"]
CSV_NAME = "ledger.csv"
CACHE_PATH = Path("/tmp/finance_ledger.csv")
class Ledger:
def __init__(self):
self._lock = threading.RLock()
self.token = os.getenv("HF_TOKEN")
self.repo = os.getenv("HF_LEDGER_REPO")
self.enabled = bool(self.token and self.repo)
self.df = self._load()
# ── persistence ──────────────────────────────────────────────────────────
def _load(self) -> pd.DataFrame:
if self.enabled:
try:
self._ensure_repo()
from huggingface_hub import hf_hub_download
path = hf_hub_download(
self.repo, CSV_NAME, repo_type="dataset",
token=self.token, local_dir="/tmp",
)
return self._read_csv(path)
except Exception as e:
logger.warning(f"HF load failed ({e}), falling back to local cache")
if CACHE_PATH.exists():
try:
return self._read_csv(CACHE_PATH)
except Exception as e:
logger.warning(f"Local cache load failed: {e}")
return pd.DataFrame(columns=COLUMNS)
def _read_csv(self, path) -> pd.DataFrame:
df = pd.read_csv(path)
df["Date"] = pd.to_datetime(df["Date"])
df["Amount"] = pd.to_numeric(df["Amount"])
return df.sort_values("Date", ascending=False).reset_index(drop=True)
def _ensure_repo(self):
from huggingface_hub import repo_exists, create_repo
if not repo_exists(self.repo, repo_type="dataset", token=self.token):
create_repo(self.repo, repo_type="dataset", private=True,
token=self.token, exist_ok=True)
def _persist(self):
df_copy = self.df.copy()
if not df_copy.empty:
df_copy["Date"] = df_copy["Date"].dt.strftime("%Y-%m-%d")
df_copy.to_csv(CACHE_PATH, index=False)
if not self.enabled:
return
try:
from huggingface_hub import upload_file
upload_file(
path_or_fileobj=str(CACHE_PATH),
path_in_repo=CSV_NAME,
repo_id=self.repo,
repo_type="dataset",
token=self.token,
commit_message="ledger update",
)
except Exception as e:
logger.error(f"HF upload failed: {e}")
# ── mutations ─────────────────────────────────────────────────────────────
def add(self, date: str, description: str, category: str, amount: float) -> bool:
with self._lock:
try:
row = pd.DataFrame({
"Date": [pd.to_datetime(date)],
"Description": [description],
"Category": [category],
"Amount": [float(amount)],
})
self.df = pd.concat([self.df, row], ignore_index=True)
self.df = self.df.sort_values("Date", ascending=False).reset_index(drop=True)
self._persist()
return True
except Exception as e:
logger.error(f"add failed: {e}")
return False
def delete_last(self) -> bool:
with self._lock:
if self.df.empty:
return False
self.df = self.df.iloc[1:].reset_index(drop=True)
self._persist()
return True
# ── queries ───────────────────────────────────────────────────────────────
def total(self) -> float:
return float(self.df["Amount"].sum()) if not self.df.empty else 0.0
def by_category(self) -> dict[str, float]:
if self.df.empty:
return {}
return self.df.groupby("Category")["Amount"].sum().to_dict()
def recent(self, n: int = 50) -> pd.DataFrame:
df = self.df.head(n).copy()
if not df.empty:
df["Date"] = df["Date"].dt.strftime("%Y-%m-%d")
return df
@property
def status(self) -> str:
return f"✅ HF Hub: `{self.repo}`" if self.enabled else "⚠️ Local cache only"
# ── singleton ─────────────────────────────────────────────────────────────────
_instance: Ledger | None = None
def get_ledger() -> Ledger:
global _instance
if _instance is None:
_instance = Ledger()
return _instance
|