feat(summary): Smart Weekly Digest — GET /summary/weekly (Issue #121)

packages/backend/app/routes/__init__.py

@@ -7,6 +7,7 @@
 from .categories import bp as categories_bp
 from .docs import bp as docs_bp
 from .dashboard import bp as dashboard_bp
+from .summary import bp as summary_bp
 
 
 def register_routes(app: Flask):
@@ -18,3 +19,4 @@ def register_routes(app: Flask):
     app.register_blueprint(categories_bp, url_prefix="/categories")
     app.register_blueprint(docs_bp, url_prefix="/docs")
     app.register_blueprint(dashboard_bp, url_prefix="/dashboard")
+    app.register_blueprint(summary_bp, url_prefix="/summary")

packages/backend/app/routes/summary.py

@@ -0,0 +1,189 @@
+"""
+routes/summary.py — Smart Weekly Digest endpoint
+Issue #121 · rohitdash08/FinMind
+Author: Xavier Abraham Sandoval <abrahamsandoval.as@gmail.com>
+"""
+
+from datetime import date, timedelta
+from decimal import Decimal
+
+from flask import Blueprint, jsonify, request
+from flask_jwt_extended import get_jwt_identity, jwt_required
+from sqlalchemy import func
+
+from ..extensions import db
+from ..models import Category, Expense
+
+bp = Blueprint("summary", __name__)
+
+
+def _week_bounds(offset: int = 0) -> tuple[date, date]:
+    """Return (monday, sunday) for the current week minus `offset` weeks."""
+    today = date.today()
+    monday = today - timedelta(days=today.weekday()) - timedelta(weeks=offset)
+    sunday = monday + timedelta(days=6)
+    return monday, sunday
+
+
+def _aggregate_week(user_id: int, start: date, end: date) -> dict:
+    """Aggregate expenses and income for a user in [start, end]."""
+    rows = (
+        db.session.query(
+            Expense.expense_type,
+            func.sum(Expense.amount).label("total"),
+            func.count(Expense.id).label("count"),
+        )
+        .filter(
+            Expense.user_id == user_id,
+            Expense.spent_at >= start,
+            Expense.spent_at <= end,
+        )
+        .group_by(Expense.expense_type)
+        .all()
+    )
+
+    totals = {"EXPENSE": Decimal("0"), "INCOME": Decimal("0")}
+    counts = {"EXPENSE": 0, "INCOME": 0}
+    for row in rows:
+        key = row.expense_type.upper()
+        if key in totals:
+            totals[key] = row.total or Decimal("0")
+            counts[key] = row.count or 0
+
+    net = totals["INCOME"] - totals["EXPENSE"]
+    return {
+        "total_expenses": float(totals["EXPENSE"]),
+        "total_income": float(totals["INCOME"]),
+        "net": float(net),
+        "expense_count": counts["EXPENSE"],
+        "income_count": counts["INCOME"],
+    }
+
+
+def _category_breakdown(user_id: int, start: date, end: date) -> list[dict]:
+    """Return per-category spending sorted by amount desc."""
+    rows = (
+        db.session.query(
+            Category.name,
+            func.sum(Expense.amount).label("total"),
+            func.count(Expense.id).label("count"),
+        )
+        .join(Category, Expense.category_id == Category.id, isouter=True)
+        .filter(
+            Expense.user_id == user_id,
+            Expense.expense_type == "EXPENSE",
+            Expense.spent_at >= start,
+            Expense.spent_at <= end,
+        )
+        .group_by(Category.name)
+        .order_by(func.sum(Expense.amount).desc())
+        .all()
+    )
+
+    return [
+        {
+            "category": row.name or "Uncategorized",
+            "amount": float(row.total or 0),
+            "count": row.count or 0,
+        }
+        for row in rows
+    ]
+
+
+def _daily_trend(user_id: int, start: date, end: date) -> list[dict]:
+    """Return daily expense totals for sparkline charts."""
+    rows = (
+        db.session.query(
+            Expense.spent_at,
+            func.sum(Expense.amount).label("total"),
+        )
+        .filter(
+            Expense.user_id == user_id,
+            Expense.expense_type == "EXPENSE",
+            Expense.spent_at >= start,
+            Expense.spent_at <= end,
+        )
+        .group_by(Expense.spent_at)
+        .order_by(Expense.spent_at)
+        .all()
+    )
+
+    # Fill all days even if no data
+    day_map = {row.spent_at: float(row.total or 0) for row in rows}
+    trend = []
+    current = start
+    while current <= end:
+        trend.append({"date": current.isoformat(), "amount": day_map.get(current, 0.0)})
+        current += timedelta(days=1)
+    return trend
+
+
+def _generate_insights(current: dict, previous: dict) -> list[str]:
+    """Generate human-readable insight strings comparing two weeks."""
+    insights = []
+
+    exp_curr = current["total_expenses"]
+    exp_prev = previous["total_expenses"]
+
+    if exp_prev > 0:
+        pct = ((exp_curr - exp_prev) / exp_prev) * 100
+        direction = "increased" if pct > 0 else "decreased"
+        insights.append(
+            f"Total spending {direction} by {abs(pct):.1f}% vs last week "
+            f"({exp_prev:.2f} → {exp_curr:.2f})."
+        )
+    elif exp_curr > 0:
+        insights.append(f"First week of recorded expenses: {exp_curr:.2f} total.")
+
+    net = current["net"]
+    if net >= 0:
+        insights.append(f"Positive cash flow this week: +{net:.2f}.")
+    else:
+        insights.append(f"Negative cash flow this week: {net:.2f}. Consider reviewing discretionary spend.")
+
+    return insights
+
+
+# ── Public endpoint ────────────────────────────────────────────────────────────
+
+@bp.get("/weekly")
+@jwt_required()
+def weekly_digest():
+    """
+    GET /summary/weekly?weeks_back=0
+
+    Returns a structured weekly financial digest for the authenticated user.
+    weeks_back=0  → current week
+    weeks_back=1  → last week
+    weeks_back=N  → N weeks ago
+    """
+    uid = int(get_jwt_identity())
+
+    try:
+        weeks_back = max(0, int(request.args.get("weeks_back", "0")))
+    except ValueError:
+        return jsonify(error="weeks_back must be an integer"), 400
+
+    curr_start, curr_end = _week_bounds(offset=weeks_back)
+    prev_start, prev_end = _week_bounds(offset=weeks_back + 1)
+
+    current = _aggregate_week(uid, curr_start, curr_end)
+    previous = _aggregate_week(uid, prev_start, prev_end)
+    categories = _category_breakdown(uid, curr_start, curr_end)
+    daily_trend = _daily_trend(uid, curr_start, curr_end)
+    insights = _generate_insights(current, previous)
+
+    return jsonify(
+        {
+            "period": {
+                "start": curr_start.isoformat(),
+                "end": curr_end.isoformat(),
+                "label": f"Week of {curr_start.strftime('%b %d, %Y')}",
+            },
+            "summary": current,
+            "previous_week": previous,
+            "category_breakdown": categories,
+            "daily_trend": daily_trend,
+            "insights": insights,
+        }
+    )

packages/backend/tests/test_summary.py

@@ -0,0 +1,117 @@
+"""
+tests/test_summary.py — Unit tests for Smart Weekly Digest
+Issue #121 · rohitdash08/FinMind
+Author: Xavier Abraham Sandoval <abrahamsandoval.as@gmail.com>
+"""
+
+from datetime import date, timedelta
+from decimal import Decimal
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.routes.summary import (
+    _aggregate_week,
+    _category_breakdown,
+    _daily_trend,
+    _generate_insights,
+    _week_bounds,
+)
+
+
+# ── Helpers ────────────────────────────────────────────────────────────────────
+
+def _make_row(**kwargs):
+    row = MagicMock()
+    for k, v in kwargs.items():
+        setattr(row, k, v)
+    return row
+
+
+# ── _week_bounds ───────────────────────────────────────────────────────────────
+
+class TestWeekBounds:
+    def test_returns_monday_to_sunday(self):
+        start, end = _week_bounds(offset=0)
+        assert start.weekday() == 0  # Monday
+        assert end.weekday() == 6    # Sunday
+        assert (end - start).days == 6
+
+    def test_offset_goes_back(self):
+        curr_start, _ = _week_bounds(offset=0)
+        prev_start, _ = _week_bounds(offset=1)
+        assert curr_start - prev_start == timedelta(weeks=1)
+
+
+# ── _generate_insights ─────────────────────────────────────────────────────────
+
+class TestGenerateInsights:
+    def _curr(self, expenses=100.0, income=200.0):
+        return {"total_expenses": expenses, "total_income": income, "net": income - expenses}
+
+    def _prev(self, expenses=80.0, income=150.0):
+        return {"total_expenses": expenses, "total_income": income, "net": income - expenses}
+
+    def test_spending_increased(self):
+        insights = _generate_insights(self._curr(expenses=100), self._prev(expenses=80))
+        assert any("increased" in i for i in insights)
+
+    def test_spending_decreased(self):
+        insights = _generate_insights(self._curr(expenses=60), self._prev(expenses=80))
+        assert any("decreased" in i for i in insights)
+
+    def test_positive_cashflow(self):
+        insights = _generate_insights(self._curr(expenses=100, income=200), self._prev())
+        assert any("Positive" in i for i in insights)
+
+    def test_negative_cashflow(self):
+        insights = _generate_insights(self._curr(expenses=300, income=100), self._prev())
+        assert any("Negative" in i for i in insights)
+
+    def test_first_week_no_previous(self):
+        insights = _generate_insights(self._curr(expenses=50), self._prev(expenses=0))
+        assert any("First week" in i for i in insights)
+
+
+# ── _aggregate_week ────────────────────────────────────────────────────────────
+
+class TestAggregateWeek:
+    @patch("app.routes.summary.db")
+    def test_aggregation_structure(self, mock_db):
+        mock_rows = [
+            _make_row(expense_type="EXPENSE", total=Decimal("150.00"), count=5),
+            _make_row(expense_type="INCOME", total=Decimal("500.00"), count=2),
+        ]
+        mock_db.session.query.return_value.filter.return_value.group_by.return_value.all.return_value = mock_rows
+
+        result = _aggregate_week(1, date.today(), date.today())
+
+        assert result["total_expenses"] == pytest.approx(150.0)
+        assert result["total_income"] == pytest.approx(500.0)
+        assert result["net"] == pytest.approx(350.0)
+        assert result["expense_count"] == 5
+        assert result["income_count"] == 2
+
+    @patch("app.routes.summary.db")
+    def test_empty_week(self, mock_db):
+        mock_db.session.query.return_value.filter.return_value.group_by.return_value.all.return_value = []
+        result = _aggregate_week(1, date.today(), date.today())
+        assert result["total_expenses"] == 0.0
+        assert result["net"] == 0.0
+
+
+# ── _daily_trend ───────────────────────────────────────────────────────────────
+
+class TestDailyTrend:
+    @patch("app.routes.summary.db")
+    def test_fills_all_days(self, mock_db):
+        start = date(2026, 3, 16)  # Monday
+        end = date(2026, 3, 22)    # Sunday
+        mock_db.session.query.return_value.filter.return_value.group_by.return_value.order_by.return_value.all.return_value = []
+
+        trend = _daily_trend(1, start, end)
+
+        assert len(trend) == 7
+        assert all(t["amount"] == 0.0 for t in trend)
+        assert trend[0]["date"] == "2026-03-16"
+        assert trend[-1]["date"] == "2026-03-22"