Add Personal Finance Manager with HF Hub CSV storage

.gitignore

@@ -0,0 +1,80 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+pip-wheel-metadata/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# Virtual Environment
+venv/
+ENV/
+env/
+.venv
+env.bak/
+venv.bak/
+
+# IDE
+.vscode/
+.idea/
+*.swp
+*.swo
+*~
+.DS_Store
+
+# Environment
+.env
+.env.local
+.env.*.local
+
+# Application
+ledger.csv
+*.csv
+*.xlsx
+*.xls
+
+# Logs
+*.log
+logs/
+
+# Cache
+.cache/
+.pytest_cache/
+.mypy_cache/
+
+# Gradio
+gradio_cached_examples/
+flagged/
+
+# Cache directory
+cache/
+
+# Deployment docs (not needed in Space)
+DEPLOY_AND_TEST.md
+DEPLOYMENT_QUICK_START.sh
+DEVELOPMENT.md
+HF_STORAGE_SETUP.md
+QUICKSTART.md
+SPACES_DEPLOYMENT.md
+PUSH_TO_SPACE.md
+spaces_config.yaml
+test_app.py
+test_standalone.py
+.env.example

README.md

@@ -0,0 +1,222 @@
+---
+title: Money Manager
+emoji: 💸
+colorFrom: green
+colorTo: blue
+sdk: gradio
+sdk_version: 4.44.0
+app_file: app.py
+pinned: false
+---
+
+# 💸 Personal Finance Manager
+
+A Gradio-based web application for managing personal finances with LLM-powered natural language expense logging. Log expenses like "Spent $15 on a burrito at Chipotle" and let AI parse them into organized ledger entries.
+
+## Features
+
+✨ **Natural Language Parsing**: Describe expenses in your own words—the LLM handles extraction
+📊 **Dynamic Ledger**: Real-time table showing all expenses with sorting and filtering
+💰 **Total Tracking**: Automatically calculated total spending that updates instantly
+🏷️ **Smart Categorization**: Expenses are automatically categorized (Food, Transportation, Utilities, etc.)
+🎨 **Clean Dashboard**: Financial-themed UI using Gradio's Soft theme
+🔄 **Session Persistence**: Ledger data persists throughout your session
+⚡ **Fallback Parser**: Works even without LLM API keys using rule-based parsing
+
+## Tech Stack
+
+- **Frontend**: Gradio (Python web framework)
+- **Data**: Pandas (DataFrames)
+- **LLM**: LangChain with HuggingFace Hub or OpenAI
+- **Language**: Python 3.8+
+
+## Setup
+
+### 1. Clone the Repository
+```bash
+cd financemanager
+```
+
+### 2. Create Virtual Environment
+```bash
+python3 -m venv venv
+source venv/bin/activate  # On Windows: venv\Scripts\activate
+```
+
+### 3. Install Dependencies
+```bash
+pip install -r requirements.txt
+```
+
+### 4. Configure API Keys (Optional)
+
+Copy `.env.example` to `.env` and add your API keys:
+
+```bash
+cp .env.example .env
+```
+
+Edit `.env`:
+- **HuggingFace**: Get token from https://huggingface.co/settings/tokens
+- **OpenAI**: Get key from https://platform.openai.com/api-keys
+
+If you don't configure any API keys, the app will use the fallback rule-based parser.
+
+### 5. Run the Application
+
+```bash
+python app.py
+```
+
+The app will launch at `http://localhost:7860`
+
+## Usage
+
+1. **Describe Your Expense**: Type a natural language description in the input box
+   - Examples:
+     - "Spent $15 on a burrito at Chipotle"
+     - "Paid $1200 for rent today"
+     - "Gas: $45.50"
+     - "Movie tickets $32"
+
+2. **Click Log Expense** or press Enter
+3. **View Results**: 
+   - Status message confirms the entry
+   - Table updates with the new expense
+   - Total spending updates automatically
+   - Expenses sorted by date (newest first)
+
+## How It Works
+
+### LLM-Based Parsing (Recommended)
+When an LLM is configured, the app sends your input to the model with this prompt:
+
+```
+Parse this expense and return JSON with:
+- date (YYYY-MM-DD)
+- description (what was purchased)
+- category (Food, Transportation, Utilities, etc.)
+- amount (numeric value)
+```
+
+The LLM returns structured JSON that the app parses and stores.
+
+### Fallback Parser
+Without an LLM, the app uses:
+- **Regex** to extract dollar amounts
+- **Keyword matching** for category detection
+- **Current date** for entries without explicit dates
+
+## Expense Data Structure
+
+Each entry contains:
+
+| Field | Example | Notes |
+|-------|---------|-------|
+| Date | 2024-05-01 | YYYY-MM-DD format |
+| Description | Burrito at Chipotle | What was purchased |
+| Category | Food | Auto-categorized |
+| Amount | 15.00 | Dollar amount |
+
+## Supported Categories
+
+- **Food**: Restaurant, groceries, coffee
+- **Transportation**: Gas, Uber, parking, taxi
+- **Utilities**: Electric, water, internet, phone
+- **Entertainment**: Movies, concerts, books, games
+- **Rent**: Rent, mortgage, apartment
+- **Other**: Uncategorized expenses
+
+## Deployment to HuggingFace Spaces
+
+### 1. Create a Space
+- Go to https://huggingface.co/spaces
+- Click "Create new Space"
+- Choose "Gradio" as the SDK
+- Set repository visibility to public/private
+
+### 2. Upload Files
+```bash
+git clone https://huggingface.co/spaces/your-username/your-space
+cd your-space
+# Copy app.py, requirements.txt, .env to this directory
+git add .
+git commit -m "Add finance manager"
+git push
+```
+
+### 3. Add Secrets
+In your Space's Settings → Repository secrets, add:
+- `HUGGINGFACEHUB_API_TOKEN`
+- `OPENAI_API_KEY` (if using OpenAI)
+
+The space will auto-deploy and be accessible at: `https://huggingface.co/spaces/your-username/your-space`
+
+## Customization
+
+### Change Theme
+In `app.py`, line 214:
+```python
+with gr.Blocks(theme=gr.themes.Soft()) as demo:
+```
+
+Try: `Default()`, `Glass()`, `Monochrome()`, `Soft()`, `Base()`
+
+### Add More Categories
+Edit `parse_expense_fallback()` function, around line 149:
+```python
+categories = {
+    "Shopping": ["amazon", "mall", "store", "buy"],
+    "Medical": ["doctor", "pharmacy", "clinic"],
+    # Add more...
+}
+```
+
+### Change LLM Model
+In `initialize_llm()`, line 68:
+```python
+repo_id="mistralai/Mistral-7B-Instruct-v0.2",
+# Try: "HuggingFaceH4/zephyr-7b-beta", "meta-llama/Llama-2-7b-chat-hf"
+```
+
+## Limitations
+
+- ⚠️ Session data is not persisted between app restarts (no database)
+- ⚠️ All amounts are in USD (no multi-currency support)
+- ⚠️ LLM parsing may fail for very ambiguous inputs
+- ⚠️ No built-in authentication (use for personal/private deployments)
+
+## Future Enhancements
+
+- [ ] CSV export functionality
+- [ ] Monthly/yearly summaries with charts
+- [ ] Budget alerts
+- [ ] Receipt image upload
+- [ ] Multi-currency support
+- [ ] SQLite database for persistence
+- [ ] User authentication for Spaces deployment
+
+## Troubleshooting
+
+### "LLM not available" Warning
+The app works without an LLM. This just means it's using the fallback parser. Add an API key to `.env` to enable intelligent parsing.
+
+### "JSON parsing error"
+The LLM response format was unexpected. Try rephrasing your expense description or check your API key.
+
+### App Hangs on Startup
+- Check that your API keys are correct
+- Ensure you have internet connectivity
+- Try disabling the LLM by not setting environment variables
+
+## License
+
+MIT License - feel free to modify and deploy!
+
+## Support
+
+For issues or suggestions, please check the code comments or modify as needed.
+
+---
+
+**Happy budgeting! 💰**

app.py

@@ -0,0 +1,298 @@
+import gradio as gr
+import pandas as pd
+import json
+from datetime import datetime
+from typing import Tuple, Dict, Any
+import os
+import logging
+
+try:
+    from langchain.llms import HuggingFaceHub
+    from langchain.prompts import PromptTemplate
+    from langchain.chains import LLMChain
+except ImportError:
+    # Fallback: try OpenAI or basic mock
+    pass
+
+from hf_storage import HFHubLedger
+
+# Setup logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class ExpenseManager:
+    """Manages ledger entries and DataFrame operations."""
+
+    def __init__(self):
+        """Initialize the expense manager with an empty DataFrame."""
+        self.df = pd.DataFrame(
+            columns=["Date", "Description", "Category", "Amount"]
+        )
+        self.df["Date"] = pd.to_datetime(self.df["Date"])
+        self.df["Amount"] = pd.to_numeric(self.df["Amount"])
+
+    def add_entry(self, date: str, description: str, category: str, amount: float) -> bool:
+        """Add a new expense entry to the ledger."""
+        try:
+            new_entry = pd.DataFrame({
+                "Date": [pd.to_datetime(date)],
+                "Description": [description],
+                "Category": [category],
+                "Amount": [float(amount)]
+            })
+            self.df = pd.concat([self.df, new_entry], ignore_index=True)
+            self.df = self.df.sort_values("Date", ascending=False).reset_index(drop=True)
+            return True
+        except Exception as e:
+            print(f"Error adding entry: {e}")
+            return False
+
+    def get_dataframe(self) -> pd.DataFrame:
+        """Return the current DataFrame."""
+        return self.df.copy()
+
+    def get_total_spending(self) -> float:
+        """Calculate and return total spending."""
+        if self.df.empty:
+            return 0.0
+        return self.df["Amount"].sum()
+
+    def get_category_summary(self) -> Dict[str, float]:
+        """Get spending summary by category."""
+        if self.df.empty:
+            return {}
+        return self.df.groupby("Category")["Amount"].sum().to_dict()
+
+
+def initialize_llm():
+    """Initialize the LLM. Supports HuggingFace or OpenAI."""
+    try:
+        # Try HuggingFace
+        api_token = os.getenv("HUGGINGFACEHUB_API_TOKEN")
+        if api_token:
+            llm = HuggingFaceHub(
+                repo_id="mistralai/Mistral-7B-Instruct-v0.2",
+                huggingfacehub_api_token=api_token,
+                model_kwargs={"temperature": 0.1, "max_length": 200}
+            )
+            return llm
+    except Exception as e:
+        print(f"HuggingFace initialization failed: {e}")
+
+    try:
+        # Fallback to OpenAI
+        from langchain.llms import OpenAI
+        api_key = os.getenv("OPENAI_API_KEY")
+        if api_key:
+            return OpenAI(temperature=0.1, max_tokens=200)
+    except Exception as e:
+        print(f"OpenAI initialization failed: {e}")
+
+    return None
+
+
+def parse_expense_with_llm(user_input: str, llm) -> Dict[str, Any]:
+    """
+    Parse natural language input into structured expense data using LLM.
+    Returns a dictionary with keys: date, description, category, amount
+    """
+    if not llm:
+        return parse_expense_fallback(user_input)
+
+    prompt_template = PromptTemplate(
+        input_variables=["user_input"],
+        template="""Parse the following expense entry and extract the information into a JSON object.
+        
+User input: {user_input}
+
+Return ONLY a valid JSON object with these fields (use today's date if not specified):
+- date (YYYY-MM-DD format)
+- description (what was purchased)
+- category (e.g., Food, Transportation, Utilities, Entertainment, Other)
+- amount (numeric value without currency symbol)
+
+JSON:"""
+    )
+
+    chain = LLMChain(llm=llm, prompt=prompt_template)
+    response = chain.run(user_input=user_input)
+
+    try:
+        # Extract JSON from response
+        json_str = response.strip()
+        # Find JSON object in response
+        start_idx = json_str.find("{")
+        end_idx = json_str.rfind("}") + 1
+        if start_idx != -1 and end_idx > start_idx:
+            json_str = json_str[start_idx:end_idx]
+        parsed = json.loads(json_str)
+        return parsed
+    except json.JSONDecodeError as e:
+        print(f"JSON parsing error: {e}")
+        return parse_expense_fallback(user_input)
+
+
+def parse_expense_fallback(user_input: str) -> Dict[str, Any]:
+    """
+    Fallback parser using regex and heuristics when LLM is unavailable.
+    """
+    import re
+
+    result = {
+        "date": datetime.now().strftime("%Y-%m-%d"),
+        "description": user_input,
+        "category": "Other",
+        "amount": 0.0
+    }
+
+    # Try to extract amount
+    amount_pattern = r"\$?(\d+(?:\.\d{2})?)"
+    amount_match = re.search(amount_pattern, user_input)
+    if amount_match:
+        result["amount"] = float(amount_match.group(1))
+
+    # Simple category detection
+    categories = {
+        "Food": ["food", "lunch", "dinner", "breakfast", "coffee", "restaurant", "burrito", "pizza", "eat"],
+        "Transportation": ["gas", "uber", "lyft", "taxi", "bus", "train", "parking", "car"],
+        "Utilities": ["electric", "water", "gas", "internet", "phone", "utility"],
+        "Entertainment": ["movie", "concert", "game", "book", "music"],
+        "Rent": ["rent", "apartment", "mortgage"],
+    }
+
+    user_lower = user_input.lower()
+    for category, keywords in categories.items():
+        if any(keyword in user_lower for keyword in keywords):
+            result["category"] = category
+            break
+
+    return result
+
+
+def process_expense_entry(
+    user_input: str,
+    manager: ExpenseManager,
+    llm,
+    hf_ledger: HFHubLedger = None
+) -> Tuple[pd.DataFrame, str, str]:
+    """
+    Process user input, parse it, add to ledger, and return updated table.
+    """
+    if not user_input.strip():
+        return manager.get_dataframe(), "", "Please enter an expense description."
+
+    try:
+        # Parse the expense
+        parsed = parse_expense_with_llm(user_input, llm)
+
+        # Validate parsed data
+        if not parsed.get("amount") or parsed["amount"] <= 0:
+            return manager.get_dataframe(), "", "❌ Error: Could not extract valid amount. Try again."
+
+        # Add to ledger
+        success = manager.add_entry(
+            date=parsed.get("date", datetime.now().strftime("%Y-%m-%d")),
+            description=parsed.get("description", user_input),
+            category=parsed.get("category", "Other"),
+            amount=float(parsed["amount"])
+        )
+
+        if success:
+            # Sync to HF Hub if enabled
+            if hf_ledger:
+                hf_ledger.save(manager.df)
+            
+            total = manager.get_total_spending()
+            message = f"✅ Logged: ${parsed['amount']:.2f} - {parsed['description']}"
+            return manager.get_dataframe(), "", message
+        else:
+            return manager.get_dataframe(), "", "❌ Error adding entry. Please try again."
+
+    except Exception as e:
+        return manager.get_dataframe(), "", f"❌ Error: {str(e)}"
+
+
+def build_interface(manager, llm, hf_ledger: HFHubLedger):
+    """Build the Gradio interface."""
+
+    def log_expense_callback(user_input: str) -> Tuple[pd.DataFrame, str, str]:
+        """Callback for log expense button."""
+        df, cleared_input, message = process_expense_entry(user_input, manager, llm, hf_ledger)
+        total = manager.get_total_spending()
+        total_md = f"### 💰 Total Spending: ${total:.2f}"
+        return df, cleared_input, message, total_md
+
+    with gr.Blocks(theme=gr.themes.Soft()) as demo:
+        gr.Markdown("# 💸 Personal Finance Manager")
+        gr.Markdown("Log your expenses using natural language. The AI will parse and categorize them for you.")
+        gr.Markdown(f"**Storage Status:** {hf_ledger.get_status()}")
+
+        with gr.Row():
+            with gr.Column(scale=3):
+                user_input = gr.Textbox(
+                    label="Describe your expense",
+                    placeholder="e.g., 'Spent $15 on a burrito at Chipotle' or 'Paid $1200 for rent'",
+                    lines=2
+                )
+            with gr.Column(scale=1):
+                log_button = gr.Button("Log Expense", variant="primary", scale=1)
+
+        status_output = gr.Textbox(
+            label="Status",
+            interactive=False,
+            max_lines=1
+        )
+
+        total_display = gr.Markdown("### 💰 Total Spending: $0.00")
+
+        gr.Markdown("## 📊 Ledger")
+        ledger_table = gr.Dataframe(
+            value=manager.get_dataframe(),
+            interactive=False,
+            label="Expense Entries",
+            datatype=["str", "str", "str", "number"],
+        )
+
+        # Connect button click to callback
+        log_button.click(
+            fn=log_expense_callback,
+            inputs=[user_input],
+            outputs=[ledger_table, user_input, status_output, total_display]
+        )
+
+        # Allow Enter key to submit
+        user_input.submit(
+            fn=log_expense_callback,
+            inputs=[user_input],
+            outputs=[ledger_table, user_input, status_output, total_display]
+        )
+
+    return demo
+
+
+def main():
+    """Main entry point."""
+    # Initialize HuggingFace Hub ledger
+    hf_ledger = HFHubLedger()
+    
+    # Initialize components
+    manager = ExpenseManager()
+    
+    # Load existing data from HF Hub if available
+    if hf_ledger.df is not None and not hf_ledger.df.empty:
+        manager.df = hf_ledger.df.copy()
+        logger.info(f"Loaded {len(manager.df)} entries from persistent storage")
+    
+    llm = initialize_llm()
+
+    if not llm:
+        logger.warning("⚠️  Warning: LLM not available. Using fallback parser.")
+
+    # Build and launch interface
+    demo = build_interface(manager, llm, hf_ledger)
+    demo.launch(share=False)
+
+
+if __name__ == "__main__":
+    main()

hf_storage.py

@@ -0,0 +1,267 @@
+"""HuggingFace Hub storage integration for persistent ledger management."""
+
+import os
+import time
+import pandas as pd
+import tempfile
+from pathlib import Path
+from typing import Optional
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class HFHubLedger:
+    """Manages ledger CSV persistence using HuggingFace Hub storage."""
+
+    def __init__(
+        self,
+        hf_token: Optional[str] = None,
+        repo_id: Optional[str] = None,
+        repo_type: str = "dataset",
+        csv_filename: str = "ledger.csv",
+        local_cache_dir: str = "./cache",
+        max_retries: int = 3,
+        retry_delay: float = 1.0,
+    ):
+        """
+        Initialize HuggingFace Hub ledger storage.
+
+        Args:
+            hf_token: HuggingFace API token (uses HF_TOKEN env var if not provided)
+            repo_id: Repository ID in format "username/repo-name"
+            repo_type: Type of repo ("dataset", "model", "space")
+            csv_filename: Name of the CSV file in the repo
+            local_cache_dir: Local directory for caching
+            max_retries: Maximum number of upload retries
+            retry_delay: Initial delay between retries (exponential backoff)
+        """
+        self.hf_token = hf_token or os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACEHUB_API_TOKEN")
+        self.repo_id = repo_id or os.getenv("HF_REPO_ID")
+        self.repo_type = repo_type
+        self.csv_filename = csv_filename
+        self.local_cache_dir = local_cache_dir
+        self.max_retries = max_retries
+        self.retry_delay = retry_delay
+        self.enabled = self.hf_token and self.repo_id
+        self.df = None
+
+        # Create local cache directory
+        Path(self.local_cache_dir).mkdir(parents=True, exist_ok=True)
+        self.local_csv_path = Path(self.local_cache_dir) / self.csv_filename
+
+        if self.enabled:
+            logger.info(f"HF Hub storage enabled: {self.repo_id}")
+            self._ensure_repo_exists()
+            self._load_from_hub()
+        else:
+            logger.warning("HF Hub storage disabled. Set HF_TOKEN and HF_REPO_ID to enable.")
+            self._load_local_or_create()
+
+    def _ensure_repo_exists(self) -> bool:
+        """
+        Ensure the HuggingFace Hub repository exists.
+
+        Returns:
+            True if repo exists or was created, False otherwise
+        """
+        try:
+            from huggingface_hub import create_repo, repo_exists
+            
+            if repo_exists(self.repo_id, repo_type=self.repo_type, token=self.hf_token):
+                logger.info(f"Repository {self.repo_id} exists")
+                return True
+
+            # Create repo if it doesn't exist
+            repo_url = create_repo(
+                self.repo_id,
+                repo_type=self.repo_type,
+                private=True,
+                exist_ok=True,
+                token=self.hf_token,
+            )
+            logger.info(f"Created repository: {repo_url}")
+            return True
+        except Exception as e:
+            logger.error(f"Failed to ensure repo exists: {e}")
+            return False
+
+    def _load_from_hub(self) -> bool:
+        """
+        Download and load CSV from HuggingFace Hub.
+
+        Returns:
+            True if successful, False otherwise
+        """
+        try:
+            from huggingface_hub import hf_hub_download
+            
+            logger.info(f"Attempting to download {self.csv_filename} from {self.repo_id}")
+            
+            file_path = hf_hub_download(
+                repo_id=self.repo_id,
+                filename=self.csv_filename,
+                repo_type=self.repo_type,
+                token=self.hf_token,
+                cache_dir=self.local_cache_dir,
+            )
+            
+            # Load CSV
+            self.df = pd.read_csv(file_path)
+            self.df["Date"] = pd.to_datetime(self.df["Date"])
+            self.df["Amount"] = pd.to_numeric(self.df["Amount"])
+            self.df = self.df.sort_values("Date", ascending=False).reset_index(drop=True)
+            
+            logger.info(f"Loaded {len(self.df)} entries from HF Hub")
+            return True
+            
+        except Exception as e:
+            logger.warning(f"Could not load from Hub: {e}. Starting fresh.")
+            self._load_local_or_create()
+            return False
+
+    def _load_local_or_create(self) -> bool:
+        """
+        Load CSV from local cache or create new DataFrame.
+
+        Returns:
+            True if loaded, False if created new
+        """
+        if self.local_csv_path.exists():
+            try:
+                self.df = pd.read_csv(self.local_csv_path)
+                self.df["Date"] = pd.to_datetime(self.df["Date"])
+                self.df["Amount"] = pd.to_numeric(self.df["Amount"])
+                logger.info(f"Loaded {len(self.df)} entries from local cache")
+                return True
+            except Exception as e:
+                logger.warning(f"Failed to load local CSV: {e}")
+
+        # Create new empty DataFrame
+        self.df = pd.DataFrame(columns=["Date", "Description", "Category", "Amount"])
+        self.df["Date"] = pd.to_datetime(self.df["Date"])
+        self.df["Amount"] = pd.to_numeric(self.df["Amount"])
+        logger.info("Created new empty ledger")
+        return False
+
+    def save(self, df: pd.DataFrame) -> bool:
+        """
+        Save DataFrame to local cache and optionally to HF Hub.
+
+        Args:
+            df: DataFrame to save
+
+        Returns:
+            True if successful, False otherwise
+        """
+        try:
+            # Save locally first
+            df_copy = df.copy()
+            df_copy["Date"] = df_copy["Date"].dt.strftime("%Y-%m-%d")
+            df_copy.to_csv(self.local_csv_path, index=False)
+            self.df = df
+            
+            # Upload to Hub if enabled
+            if self.enabled:
+                self._upload_to_hub_with_retry()
+            
+            return True
+        except Exception as e:
+            logger.error(f"Failed to save ledger: {e}")
+            return False
+
+    def _upload_to_hub_with_retry(self) -> bool:
+        """
+        Upload CSV to HuggingFace Hub with exponential backoff retry.
+
+        Returns:
+            True if successful, False otherwise
+        """
+        for attempt in range(self.max_retries):
+            try:
+                from huggingface_hub import upload_file
+                
+                logger.info(f"Uploading to HF Hub (attempt {attempt + 1}/{self.max_retries})")
+                
+                upload_file(
+                    path_or_fileobj=str(self.local_csv_path),
+                    path_in_repo=self.csv_filename,
+                    repo_id=self.repo_id,
+                    repo_type=self.repo_type,
+                    token=self.hf_token,
+                    commit_message=f"Auto-save ledger at {pd.Timestamp.now()}",
+                )
+                
+                logger.info("Successfully uploaded to HF Hub")
+                return True
+                
+            except Exception as e:
+                wait_time = self.retry_delay * (2 ** attempt)  # Exponential backoff
+                logger.warning(f"Upload failed (attempt {attempt + 1}): {e}")
+                
+                if attempt < self.max_retries - 1:
+                    logger.info(f"Retrying in {wait_time:.1f}s...")
+                    time.sleep(wait_time)
+                else:
+                    logger.error(f"Failed to upload after {self.max_retries} attempts")
+                    return False
+
+        return False
+
+    def get_dataframe(self) -> pd.DataFrame:
+        """Return a copy of the current DataFrame."""
+        if self.df is None:
+            return pd.DataFrame(columns=["Date", "Description", "Category", "Amount"])
+        return self.df.copy()
+
+    def add_entry(self, date: str, description: str, category: str, amount: float) -> bool:
+        """
+        Add a new entry and save.
+
+        Args:
+            date: Date in YYYY-MM-DD format
+            description: Expense description
+            category: Expense category
+            amount: Amount in dollars
+
+        Returns:
+            True if successful, False otherwise
+        """
+        try:
+            new_entry = pd.DataFrame({
+                "Date": [pd.to_datetime(date)],
+                "Description": [description],
+                "Category": [category],
+                "Amount": [float(amount)]
+            })
+            self.df = pd.concat([self.df, new_entry], ignore_index=True)
+            self.df = self.df.sort_values("Date", ascending=False).reset_index(drop=True)
+            
+            # Save immediately
+            return self.save(self.df)
+        except Exception as e:
+            logger.error(f"Failed to add entry: {e}")
+            return False
+
+    def get_total_spending(self) -> float:
+        """Calculate and return total spending."""
+        if self.df is None or self.df.empty:
+            return 0.0
+        return float(self.df["Amount"].sum())
+
+    def get_category_summary(self) -> dict:
+        """Get spending summary by category."""
+        if self.df is None or self.df.empty:
+            return {}
+        return self.df.groupby("Category")["Amount"].sum().to_dict()
+
+    def is_enabled(self) -> bool:
+        """Check if HF Hub storage is enabled."""
+        return self.enabled
+
+    def get_status(self) -> str:
+        """Get human-readable status string."""
+        if self.enabled:
+            return f"✅ HF Hub: {self.repo_id}"
+        else:
+            return "⚠️ Local cache only (HF Hub disabled)"

requirements.txt

@@ -0,0 +1,5 @@
+gradio>=4.0.0
+pandas>=2.0.0
+langchain>=0.1.0
+huggingface-hub>=0.17.0
+python-dotenv>=1.0.0

utils.py

@@ -0,0 +1,225 @@
+"""Utility functions for the Finance Manager application."""
+
+import pandas as pd
+import os
+from datetime import datetime
+from typing import Optional
+
+
+class CSVLedger:
+    """Handles CSV persistence for the expense ledger."""
+
+    def __init__(self, filepath: str = "ledger.csv"):
+        """
+        Initialize the CSV ledger handler.
+
+        Args:
+            filepath: Path to the CSV file
+        """
+        self.filepath = filepath
+        self.df = self._load_or_create()
+
+    def _load_or_create(self) -> pd.DataFrame:
+        """Load existing CSV or create new DataFrame."""
+        if os.path.exists(self.filepath):
+            try:
+                df = pd.read_csv(self.filepath)
+                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)
+            except Exception as e:
+                print(f"Error loading CSV: {e}. Creating new ledger.")
+
+        return pd.DataFrame(columns=["Date", "Description", "Category", "Amount"])
+
+    def save(self, df: pd.DataFrame) -> bool:
+        """
+        Save DataFrame to CSV.
+
+        Args:
+            df: DataFrame to save
+
+        Returns:
+            True if successful, False otherwise
+        """
+        try:
+            # Convert datetime to string for CSV
+            df_copy = df.copy()
+            df_copy["Date"] = df_copy["Date"].dt.strftime("%Y-%m-%d")
+            df_copy.to_csv(self.filepath, index=False)
+            return True
+        except Exception as e:
+            print(f"Error saving CSV: {e}")
+            return False
+
+    def append_from_dataframe(self, df: pd.DataFrame) -> bool:
+        """
+        Append DataFrame entries to CSV.
+
+        Args:
+            df: DataFrame with new entries
+
+        Returns:
+            True if successful, False otherwise
+        """
+        self.df = pd.concat([self.df, df], ignore_index=True)
+        self.df = self.df.sort_values("Date", ascending=False).reset_index(drop=True)
+        return self.save(self.df)
+
+
+def format_currency(amount: float) -> str:
+    """
+    Format amount as USD currency.
+
+    Args:
+        amount: Numeric amount
+
+    Returns:
+        Formatted string like "$123.45"
+    """
+    return f"${amount:,.2f}"
+
+
+def parse_date_flexible(date_str: Optional[str]) -> str:
+    """
+    Parse various date formats and return ISO format (YYYY-MM-DD).
+
+    Args:
+        date_str: Date string in various formats or None
+
+    Returns:
+        ISO format date string
+    """
+    if not date_str or date_str.lower() == "today" or date_str.lower() == "now":
+        return datetime.now().strftime("%Y-%m-%d")
+
+    # Try common formats
+    formats = [
+        "%Y-%m-%d",
+        "%m/%d/%Y",
+        "%m/%d/%y",
+        "%m-%d-%Y",
+        "%d/%m/%Y",
+        "%Y/%m/%d",
+    ]
+
+    for fmt in formats:
+        try:
+            dt = datetime.strptime(date_str.strip(), fmt)
+            return dt.strftime("%Y-%m-%d")
+        except ValueError:
+            continue
+
+    # Default to today
+    return datetime.now().strftime("%Y-%m-%d")
+
+
+def get_spending_summary(df: pd.DataFrame) -> dict:
+    """
+    Generate spending summary by category.
+
+    Args:
+        df: Expense DataFrame
+
+    Returns:
+        Dictionary with category totals
+    """
+    if df.empty:
+        return {}
+
+    summary = df.groupby("Category")["Amount"].agg(["sum", "count"]).to_dict("index")
+    return {
+        cat: {
+            "total": values["sum"],
+            "count": int(values["count"]),
+            "average": values["sum"] / values["count"]
+        }
+        for cat, values in summary.items()
+    }
+
+
+def get_daily_summary(df: pd.DataFrame) -> pd.DataFrame:
+    """
+    Generate daily spending summary.
+
+    Args:
+        df: Expense DataFrame
+
+    Returns:
+        DataFrame with daily totals
+    """
+    if df.empty:
+        return pd.DataFrame(columns=["Date", "Total", "Count"])
+
+    daily = df.groupby(df["Date"].dt.date).agg({
+        "Amount": ["sum", "count"]
+    }).reset_index()
+    daily.columns = ["Date", "Total", "Count"]
+    return daily.sort_values("Date", ascending=False)
+
+
+def validate_expense_data(date: str, description: str, category: str, amount: float) -> tuple[bool, str]:
+    """
+    Validate expense entry data.
+
+    Args:
+        date: Date string
+        description: Expense description
+        category: Expense category
+        amount: Amount in dollars
+
+    Returns:
+        Tuple of (is_valid, error_message)
+    """
+    errors = []
+
+    # Validate date
+    if not date:
+        errors.append("Date is required")
+    else:
+        try:
+            datetime.strptime(date, "%Y-%m-%d")
+        except ValueError:
+            errors.append("Date must be in YYYY-MM-DD format")
+
+    # Validate description
+    if not description or len(description.strip()) == 0:
+        errors.append("Description is required")
+    elif len(description) > 500:
+        errors.append("Description is too long (max 500 characters)")
+
+    # Validate category
+    if not category or len(category.strip()) == 0:
+        errors.append("Category is required")
+
+    # Validate amount
+    if amount is None or amount <= 0:
+        errors.append("Amount must be greater than 0")
+    elif amount > 999999.99:
+        errors.append("Amount is too large (max $999,999.99)")
+
+    if errors:
+        return False, "\n".join(errors)
+
+    return True, ""
+
+
+def export_to_csv(df: pd.DataFrame, filepath: str) -> bool:
+    """
+    Export DataFrame to CSV file.
+
+    Args:
+        df: DataFrame to export
+        filepath: Output file path
+
+    Returns:
+        True if successful, False otherwise
+    """
+    try:
+        df_copy = df.copy()
+        df_copy["Date"] = df_copy["Date"].dt.strftime("%Y-%m-%d")
+        df_copy.to_csv(filepath, index=False)
+        return True
+    except Exception as e:
+        print(f"Error exporting to CSV: {e}")
+        return False