diff --git a/backend/app/database.py b/backend/app/database.py index 6dfca2e..b55e5f0 100644 --- a/backend/app/database.py +++ b/backend/app/database.py @@ -41,7 +41,7 @@ def get_db(): def init_db(): """Create all tables.""" # import models so SQLAlchemy registers them - from app.models import admin_user, upstream, snapshot, webhook_config, notification_log, custom_page, website, revoked_token, upstream_key, external_api_log # noqa: F401 + from app.models import admin_user, upstream, snapshot, webhook_config, notification_log, custom_page, website, revoked_token, upstream_key, external_api_log, finance_daily_summary # noqa: F401 Base.metadata.create_all(bind=engine) _ensure_indexes() _migrate_custom_pages() diff --git a/backend/app/models/finance_daily_summary.py b/backend/app/models/finance_daily_summary.py new file mode 100644 index 0000000..7b73c2b --- /dev/null +++ b/backend/app/models/finance_daily_summary.py @@ -0,0 +1,30 @@ +"""Finance daily summary snapshot model. + +Stores pre-computed daily reconciliation results in a single JSON column. +This keeps the schema stable: adding new summary fields requires no migration. +""" +from datetime import date, datetime, timezone +from typing import Optional + +from sqlalchemy import Boolean, Date, DateTime, Integer, Text, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +class FinanceDailySummary(Base): + __tablename__ = "finance_daily_summaries" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + stat_date: Mapped[date] = mapped_column(Date, unique=True, nullable=False, index=True) + summary_json: Mapped[str] = mapped_column(Text, nullable=False) + success: Mapped[bool] = mapped_column(Boolean, default=False, index=True) + computed_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow) + + __table_args__ = ( + UniqueConstraint("stat_date", name="uq_finance_summary_date"), + ) diff --git a/backend/app/routers/finance.py b/backend/app/routers/finance.py index f1e14ce..515c046 100644 --- a/backend/app/routers/finance.py +++ b/backend/app/routers/finance.py @@ -1,42 +1,223 @@ -"""Finance router — daily revenue vs cost reconciliation.""" +"""Finance router — daily revenue vs cost reconciliation with snapshot storage.""" from __future__ import annotations import datetime as _dt -from typing import Any, Optional +import json +from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from app.database import get_db -from app.models.upstream import Upstream -from app.models.website import Website -from app.services.finance_service import get_daily_summary, yesterday_shanghai +from app.services.finance_service import ( + compute_daily_summary, + get_or_create_daily_summary, + overwrite_daily_summary, + today_shanghai, + yesterday_shanghai, +) from app.utils.auth import get_current_user router = APIRouter(prefix="/api/finance", tags=["finance"]) +def _resolve_date(date_str: str | None) -> _dt.date: + """Parse optional date string; return yesterday Asia/Shanghai if omitted.""" + if date_str: + try: + return _dt.date.fromisoformat(date_str) + except ValueError: + raise HTTPException(400, f"Invalid date format: {date_str!r}. Expected YYYY-MM-DD.") + return yesterday_shanghai() + + +def _validate_not_future(target: _dt.date) -> None: + """Reject future dates — we can't save snapshots for days that haven't happened.""" + if target > today_shanghai(): + raise HTTPException(400, f"Cannot query future date: {target.isoformat()}") + + @router.get("/daily-summary") def daily_summary( - date: Optional[str] = Query(None, description="YYYY-MM-DD, defaults to yesterday Asia/Shanghai"), + date: str | None = Query(None, description="YYYY-MM-DD, defaults to yesterday Asia/Shanghai"), db: Session = Depends(get_db), _: Any = Depends(get_current_user), ) -> dict: - """Return daily revenue vs cost summary. + """Return stored daily summary snapshot. - Never raises for individual item failures — each failed website/upstream - is included in the response with status='failed' and an error message. - Only raises HTTP errors for bad request parameters. + If no snapshot exists for the requested date, computes one on the fly, + saves it, and returns it. Never raises for individual item failures. """ - if date: - try: - target = _dt.date.fromisoformat(date) - except ValueError: - raise HTTPException(400, f"Invalid date format: {date!r}. Expected YYYY-MM-DD.") + target = _resolve_date(date) + _validate_not_future(target) + return get_or_create_daily_summary(db, target) + + +@router.post("/daily-summary/compare") +def compare_daily_summary( + date: str | None = Query(None, description="YYYY-MM-DD, defaults to yesterday Asia/Shanghai"), + db: Session = Depends(get_db), + _: Any = Depends(get_current_user), +) -> dict: + """Compare stored snapshot against a fresh real-time computation. + + Returns {stored, current, has_difference, diff_items}. Does NOT write to DB. + """ + target = _resolve_date(date) + _validate_not_future(target) + + from app.models.finance_daily_summary import FinanceDailySummary + + stored_row = db.query(FinanceDailySummary).filter( + FinanceDailySummary.stat_date == target + ).first() + + current = compute_daily_summary(db, target) + + if stored_row: + stored_data = json.loads(stored_row.summary_json) + stored_data["computed_at"] = stored_row.computed_at.isoformat() + stored_data["from_snapshot"] = True + diff_items, has_diff = _compute_diff_items(stored_data, current) else: - target = yesterday_shanghai() + stored_data = None + diff_items = [] + has_diff = True # no stored data = definitely different - websites = db.query(Website).filter(Website.enabled == True).all() - upstreams = db.query(Upstream).filter(Upstream.enabled == True).all() + return { + "stored": stored_data, + "current": current, + "has_difference": has_diff, + "diff_items": diff_items, + } - return get_daily_summary(websites, upstreams, target) + +@router.post("/daily-summary/overwrite") +def overwrite_daily_summary_endpoint( + date: str | None = Query(None, description="YYYY-MM-DD, defaults to yesterday Asia/Shanghai"), + db: Session = Depends(get_db), + _: Any = Depends(get_current_user), +) -> dict: + """Recompute and overwrite the stored snapshot for the requested date.""" + target = _resolve_date(date) + _validate_not_future(target) + return overwrite_daily_summary(db, target) + + +def _has_difference(stored: dict, current: dict) -> bool: + """Compare stored vs current summary at the total and item level. + + Items are compared by (id, upstream_type) identity rather than by array + position, so that a replaced website/upstream doesn't falsely appear + identical, and reordered items don't falsely appear different. + """ + for key in ("total_revenue", "total_cost", "net_income", "success"): + if stored.get(key) != current.get(key): + return True + for side in ("website_items", "upstream_items"): + s_list = stored.get(side, []) + c_list = current.get(side, []) + if len(s_list) != len(c_list): + return True + # Build lookup map by (id, upstream_type) + s_map = {(_item_id(s), s.get("upstream_type")): s for s in s_list} + c_map = {(_item_id(c), c.get("upstream_type")): c for c in c_list} + # Check that keys match exactly + if s_map.keys() != c_map.keys(): + return True + # Compare each item by its stable identity + for key_ident, s_item in s_map.items(): + c_item = c_map[key_ident] + for attr in ("status", "amount", "error"): + if s_item.get(attr) != c_item.get(attr): + return True + return False + + +def _item_id(item: dict) -> int | str: + """Return item id, defaulting to a sentinel for items that lack one.""" + v = item.get("id") + return v if v is not None else -1 + + +def _compute_diff_items(stored: dict, current: dict) -> tuple[list[dict], bool]: + """Compare stored vs current and return (diff_items, has_difference). + + diff_items is a list of changes at the item level, each with: + side, name, id, upstream_type, old_{status,amount,error}, new_{status,amount,error}. + + has_difference also covers total-level changes and count mismatches. + """ + diff_items: list[dict] = [] + has_diff = False + + # Total-level checks + for key in ("total_revenue", "total_cost", "net_income", "success"): + if stored.get(key) != current.get(key): + has_diff = True + + # Item-level checks + for side in ("website_items", "upstream_items"): + s_list = stored.get(side, []) + c_list = current.get(side, []) + if len(s_list) != len(c_list): + has_diff = True + s_map = {(_item_id(s), s.get("upstream_type")): s for s in s_list} + c_map = {(_item_id(c), c.get("upstream_type")): c for c in c_list} + if s_map.keys() != c_map.keys(): + has_diff = True + # Items in both sides — compare attributes + for key_ident, s_item in s_map.items(): + c_item = c_map.get(key_ident) + if c_item is None: + continue # already counted as key mismatch + for attr in ("status", "amount", "error"): + if s_item.get(attr) != c_item.get(attr): + has_diff = True + diff_items.append({ + "side": side, + "id": _item_id(s_item), + "name": s_item.get("name", ""), + "upstream_type": s_item.get("upstream_type", ""), + "old_status": s_item.get("status"), + "new_status": c_item.get("status"), + "old_amount": s_item.get("amount"), + "new_amount": c_item.get("amount"), + "old_error": s_item.get("error"), + "new_error": c_item.get("error"), + }) + break # one diff entry per item + # Items only in stored (removed) + for key_ident in s_map.keys() - c_map.keys(): + has_diff = True + s_item = s_map[key_ident] + diff_items.append({ + "side": side, + "id": _item_id(s_item), + "name": s_item.get("name", ""), + "upstream_type": s_item.get("upstream_type", ""), + "old_status": s_item.get("status"), + "new_status": None, + "old_amount": s_item.get("amount"), + "new_amount": None, + "old_error": s_item.get("error"), + "new_error": None, + }) + # Items only in current (added) + for key_ident in c_map.keys() - s_map.keys(): + has_diff = True + c_item = c_map[key_ident] + diff_items.append({ + "side": side, + "id": _item_id(c_item), + "name": c_item.get("name", ""), + "upstream_type": c_item.get("upstream_type", ""), + "old_status": None, + "new_status": c_item.get("status"), + "old_amount": None, + "new_amount": c_item.get("amount"), + "old_error": None, + "new_error": c_item.get("error"), + }) + + return diff_items, has_diff diff --git a/backend/app/services/finance_service.py b/backend/app/services/finance_service.py index 9e42633..0f3b351 100644 --- a/backend/app/services/finance_service.py +++ b/backend/app/services/finance_service.py @@ -7,18 +7,22 @@ Architecture: within the client context — retains cookies, 401 refresh, external API log. - Upstream cost (New-API/Nox-API): same UpstreamClient pattern. -No DB writes. Each item fails independently; failures do NOT contribute to -totals, and the UI must treat any failed item as incomplete accounting data. +New service functions (get_or_create_daily_summary, compute_daily_summary, +overwrite_daily_summary) wrap get_daily_summary with DB-backed snapshot storage. +Each item fails independently; failures do NOT contribute to totals, and the UI +must treat any failed item as incomplete accounting data. """ from __future__ import annotations import json import logging -from datetime import date, datetime, timedelta +from datetime import date, datetime, timedelta, timezone from typing import Any import pytz +from sqlalchemy.orm import Session + from app.models.upstream import Upstream from app.models.website import Website from app.services.upstream_client import UpstreamClient @@ -26,7 +30,7 @@ from app.services.website_client import Sub2ApiWebsiteClient, WebsiteError logger = logging.getLogger(__name__) -_SHANGHAI = pytz.timezone("Asia/Shanghai") +SHANGHAI_TZ = pytz.timezone("Asia/Shanghai") # ───────────────────────────────────────────── @@ -63,13 +67,17 @@ def classify_upstream(upstream: Upstream) -> str: # ───────────────────────────────────────────── def yesterday_shanghai() -> date: - return datetime.now(_SHANGHAI).date() - timedelta(days=1) + return datetime.now(SHANGHAI_TZ).date() - timedelta(days=1) + + +def today_shanghai() -> date: + return datetime.now(SHANGHAI_TZ).date() def date_to_shanghai_timestamps(d: date) -> tuple[int, int]: """Return (start_unix, end_unix) for 00:00:00..23:59:59 Asia/Shanghai on date d.""" - start = _SHANGHAI.localize(datetime(d.year, d.month, d.day, 0, 0, 0)) - end = _SHANGHAI.localize(datetime(d.year, d.month, d.day, 23, 59, 59)) + start = SHANGHAI_TZ.localize(datetime(d.year, d.month, d.day, 0, 0, 0)) + end = SHANGHAI_TZ.localize(datetime(d.year, d.month, d.day, 23, 59, 59)) return int(start.timestamp()), int(end.timestamp()) @@ -294,3 +302,142 @@ def get_daily_summary( "failed_count": failed_count, "success": failed_count == 0, } + + +# ───────────────────────────────────────────── +# DB-backed snapshot storage +# ───────────────────────────────────────────── + +def _truncate_errors(summary: dict[str, Any], max_len: int = 500) -> None: + """Truncate error messages in place to keep summary_json from growing unbounded.""" + for item in summary.get("website_items", []) + summary.get("upstream_items", []): + err = item.get("error") + if err and len(err) > max_len: + item["error"] = err[:max_len] + "..." + + +def _save_summary(db: Session, date_key: date, summary: dict[str, Any]) -> datetime: + """Save a computed summary to the database. Returns computed_at.""" + _truncate_errors(summary) + from app.models.finance_daily_summary import FinanceDailySummary + now = datetime.now(timezone.utc) + row = FinanceDailySummary( + stat_date=date_key, + summary_json=json.dumps(summary, ensure_ascii=False, default=str), + success=summary.get("success", False), + computed_at=now, + ) + db.add(row) + db.commit() + return now + + +def compute_daily_summary( + db: Session, + target_date: date, +) -> dict[str, Any]: + """Compute a fresh daily summary (no DB write). Used for compare.""" + websites = db.query(Website).filter(Website.enabled == True).all() + upstreams = db.query(Upstream).filter(Upstream.enabled == True).all() + return get_daily_summary(websites, upstreams, target_date) + + +def _load_summary(row: Any) -> dict[str, Any]: + """Load summary dict from a FinanceDailySummary row, adding metadata.""" + data = json.loads(row.summary_json) + data["computed_at"] = row.computed_at.isoformat() + data["from_snapshot"] = True + return data + + +def get_or_create_daily_summary( + db: Session, + target_date: date, +) -> dict[str, Any]: + """Return stored summary for target_date, or compute + save if missing. + + If a concurrent request already saved a row between our query and commit, + we rollback and return the existing row. This prevents duplicate-key errors. + """ + from app.models.finance_daily_summary import FinanceDailySummary + + existing = db.query(FinanceDailySummary).filter( + FinanceDailySummary.stat_date == target_date + ).first() + if existing: + return _load_summary(existing) + + # Compute and save + summary = compute_daily_summary(db, target_date) + try: + now = _save_summary(db, target_date, summary) + except Exception: + db.rollback() + # Race: another request may have inserted this date. Read again. + existing = db.query(FinanceDailySummary).filter( + FinanceDailySummary.stat_date == target_date + ).first() + if existing: + return _load_summary(existing) + # Not a race — real error. Re-raise. + raise + + summary["computed_at"] = now.isoformat() + summary["from_snapshot"] = False + return summary + + +def save_summary_if_absent( + db: Session, + target_date: date, +) -> bool: + """Compute and save summary for target_date if no snapshot exists yet. + + Returns True if a new snapshot was saved, False if one already existed. + Safe against concurrent callers: on IntegrityError rolls back and + re-checks. Used by the scheduler cron job. + """ + from app.models.finance_daily_summary import FinanceDailySummary + + existing = db.query(FinanceDailySummary).filter( + FinanceDailySummary.stat_date == target_date + ).first() + if existing: + return False + + summary = compute_daily_summary(db, target_date) + try: + _save_summary(db, target_date, summary) + return True + except Exception: + db.rollback() + # Race: another caller may have inserted this date. + existing = db.query(FinanceDailySummary).filter( + FinanceDailySummary.stat_date == target_date + ).first() + if existing: + return False + raise + + +def overwrite_daily_summary( + db: Session, + target_date: date, +) -> dict[str, Any]: + """Recompute and overwrite the stored snapshot for target_date. + + Used when the user explicitly requests a refresh. + Deletes any existing row for this date first, then inserts the new one. + """ + from app.models.finance_daily_summary import FinanceDailySummary + + db.query(FinanceDailySummary).filter( + FinanceDailySummary.stat_date == target_date + ).delete() + db.flush() + + summary = compute_daily_summary(db, target_date) + now = _save_summary(db, target_date, summary) + summary["computed_at"] = now.isoformat() + summary["from_snapshot"] = False + return summary diff --git a/backend/app/services/scheduler.py b/backend/app/services/scheduler.py index c1f8365..f043953 100644 --- a/backend/app/services/scheduler.py +++ b/backend/app/services/scheduler.py @@ -1,4 +1,4 @@ -"""APScheduler background scheduler for upstream checks.""" +"""APScheduler background scheduler for upstream checks and daily tasks.""" from __future__ import annotations import json @@ -8,6 +8,7 @@ from typing import Any from apscheduler.executors.pool import ThreadPoolExecutor from apscheduler.schedulers.background import BackgroundScheduler +from apscheduler.triggers.cron import CronTrigger from sqlalchemy.orm import Session from app.database import SessionLocal @@ -365,6 +366,16 @@ def start_scheduler() -> None: max_instances=1, misfire_grace_time=3600, ) + # Daily finance summary at 00:05 Asia/Shanghai + _scheduler.add_job( + _compute_finance_daily_summary, + trigger=CronTrigger(timezone="Asia/Shanghai", hour=0, minute=5), + id="finance_daily_summary", + replace_existing=True, + coalesce=True, + max_instances=1, + misfire_grace_time=3600, + ) logger.info("scheduler started with %d upstream job(s)", len(upstreams)) finally: db.close() @@ -383,6 +394,28 @@ def _cleanup_external_logs() -> None: db.close() +def _compute_finance_daily_summary() -> None: + """Compute and save finance daily summary for yesterday (Asia/Shanghai). + + Runs daily at 00:05 Asia/Shanghai. Delegates to save_summary_if_absent + which handles the skip-if-exists logic and concurrent-save race. + """ + from app.services.finance_service import yesterday_shanghai, save_summary_if_absent + + target = yesterday_shanghai() + db = SessionLocal() + try: + saved = save_summary_if_absent(db, target) + if saved: + logger.info("finance daily summary for %s saved", target) + else: + logger.info("finance daily summary for %s already exists, skipping", target) + except Exception: + logger.exception("failed to compute finance daily summary for %s", target) + finally: + db.close() + + def stop_scheduler() -> None: if _scheduler.running: _scheduler.shutdown(wait=True) diff --git a/backend/test_finance_daily_summary.py b/backend/test_finance_daily_summary.py index 3857d31..e5f3a62 100644 --- a/backend/test_finance_daily_summary.py +++ b/backend/test_finance_daily_summary.py @@ -1,6 +1,7 @@ """Tests for finance daily summary service.""" from __future__ import annotations +import datetime as _dt import json from datetime import date from unittest.mock import MagicMock @@ -404,3 +405,383 @@ def test_get_daily_summary_all_success(monkeypatch): assert abs(result["total_cost"] - 40.0) < 1e-6 assert abs(result["net_income"] - 60.0) < 1e-6 assert abs(result["margin_percent"] - 60.0) < 1e-6 + + +# ───────────────────────────────────────────── +# DB-backed snapshot tests (FinanceDailySummary) +# ───────────────────────────────────────────── + +def _make_inmemory_db(): + """Create an isolated in-memory SQLite session with all tables.""" + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + from app.database import Base + + # Import models so they register with Base.metadata + from app.models import finance_daily_summary # noqa: F401 + from app.models.website import Website, WebsiteGroupBinding, WebsiteSyncLog # noqa: F401 + from app.models.upstream import Upstream # noqa: F401 + + engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}) + Base.metadata.create_all(bind=engine) + Session = sessionmaker(bind=engine) + return Session() + + +def _add_minimal_test_data(db): + """Add one Website and one Upstream so compute_daily_summary has data to iterate.""" + from app.models.website import Website + from app.models.upstream import Upstream + + w = Website( + id=1, name="TestSite", base_url="http://t1", api_prefix="/api/v1/admin", + auth_type="api_key", auth_config_json="{}", enabled=True, + ) + u = Upstream( + id=1, name="TestUp", base_url="http://u1", api_prefix="api/v1", + auth_type="bearer", auth_config_json='{"token":"t"}', enabled=True, + ) + db.add_all([w, u]) + db.commit() + + +def test_get_or_create_no_existing_computes_and_saves(monkeypatch): + """Missing date triggers compute and save, returning from_snapshot=False.""" + db = _make_inmemory_db() + _add_minimal_test_data(db) + monkeypatch.setattr("app.services.finance_service.fetch_website_revenue", lambda *_: (100.0, None)) + monkeypatch.setattr("app.services.finance_service.fetch_upstream_cost_sub2api", lambda *_: (40.0, None)) + + from app.services.finance_service import get_or_create_daily_summary + + result = get_or_create_daily_summary(db, date(2026, 7, 2)) + + assert result["from_snapshot"] is False + assert abs(result["total_revenue"] - 100.0) < 1e-6 + assert abs(result["total_cost"] - 40.0) < 1e-6 + + # Verify it was saved + from app.models.finance_daily_summary import FinanceDailySummary + row = db.query(FinanceDailySummary).filter( + FinanceDailySummary.stat_date == date(2026, 7, 2) + ).first() + assert row is not None + assert row.success is True + db.close() + + +def test_get_or_create_existing_returns_stored(monkeypatch): + """Existing date returns stored data without calling upstream fetchers.""" + from app.models.finance_daily_summary import FinanceDailySummary + from app.services.finance_service import get_or_create_daily_summary + + db = _make_inmemory_db() + + # Pre-save a row + stored_data = { + "date": "2026-07-02", "total_revenue": 200.0, "total_cost": 80.0, + "net_income": 120.0, "margin_percent": 60.0, + "website_items": [], "upstream_items": [], + "failed_count": 0, "success": True, + } + row = FinanceDailySummary( + stat_date=date(2026, 7, 2), + summary_json=json.dumps(stored_data), + success=True, + ) + db.add(row) + db.commit() + + # If it calls fetch_*, the monkeypatched function will raise + called = [] + def _fail(*a, **kw): + called.append(True) + raise RuntimeError("should not be called") + monkeypatch.setattr("app.services.finance_service.fetch_website_revenue", _fail) + monkeypatch.setattr("app.services.finance_service.fetch_upstream_cost_sub2api", _fail) + + result = get_or_create_daily_summary(db, date(2026, 7, 2)) + + assert result["from_snapshot"] is True + assert abs(result["total_revenue"] - 200.0) < 1e-6 + assert len(called) == 0 + db.close() + + +def test_compare_does_not_write(monkeypatch): + """POST /daily-summary/compare must NOT write to the database.""" + from app.routers.finance import compare_daily_summary + + db = _make_inmemory_db() + _add_minimal_test_data(db) + monkeypatch.setattr("app.services.finance_service.fetch_website_revenue", lambda *_: (100.0, None)) + monkeypatch.setattr("app.services.finance_service.fetch_upstream_cost_sub2api", lambda *_: (40.0, None)) + + result = compare_daily_summary(date="2026-07-02", db=db, _=None) + + assert result["has_difference"] is True # no stored data + assert result["stored"] is None + assert abs(result["current"]["total_revenue"] - 100.0) < 1e-6 + + # Verify NOT saved + from app.models.finance_daily_summary import FinanceDailySummary + row = db.query(FinanceDailySummary).filter( + FinanceDailySummary.stat_date == date(2026, 7, 2) + ).first() + assert row is None + db.close() + + +def test_overwrite_replaces_existing(monkeypatch): + """Overwrite updates the same date's record.""" + from app.models.finance_daily_summary import FinanceDailySummary + from app.services.finance_service import overwrite_daily_summary + + db = _make_inmemory_db() + _add_minimal_test_data(db) + monkeypatch.setattr("app.services.finance_service.fetch_website_revenue", lambda *_: (100.0, None)) + monkeypatch.setattr("app.services.finance_service.fetch_upstream_cost_sub2api", lambda *_: (40.0, None)) + + # Save once + overwrite_daily_summary(db, date(2026, 7, 2)) + + # Change revenue for second save + monkeypatch.setattr("app.services.finance_service.fetch_website_revenue", lambda *_: (200.0, None)) + result2 = overwrite_daily_summary(db, date(2026, 7, 2)) + + assert abs(result2["total_revenue"] - 200.0) < 1e-6 + + # Only one row + rows = db.query(FinanceDailySummary).filter( + FinanceDailySummary.stat_date == date(2026, 7, 2) + ).all() + assert len(rows) == 1 + assert rows[0].success is True + db.close() + + +def test_error_truncation(monkeypatch): + """Error messages over 500 chars are truncated.""" + from app.services.finance_service import get_or_create_daily_summary + + db = _make_inmemory_db() + _add_minimal_test_data(db) + long_err = "x" * 1000 + monkeypatch.setattr( + "app.services.finance_service.fetch_website_revenue", + lambda *_: (0.0, long_err), + ) + monkeypatch.setattr( + "app.services.finance_service.fetch_upstream_cost_sub2api", + lambda *_: (40.0, None), + ) + + result = get_or_create_daily_summary(db, date(2026, 7, 2)) + + err = result["website_items"][0]["error"] + assert len(err) == 503 # 500 + "..." + assert err.endswith("...") + + from app.models.finance_daily_summary import FinanceDailySummary + row = db.query(FinanceDailySummary).filter( + FinanceDailySummary.stat_date == date(2026, 7, 2) + ).first() + stored = json.loads(row.summary_json) + assert len(stored["website_items"][0]["error"]) == 503 + db.close() + + +def test_future_date_rejected(): + """Future dates must return 400.""" + from app.routers.finance import _validate_not_future + from fastapi import HTTPException + import pytest + + from app.services.finance_service import today_shanghai + + with pytest.raises(HTTPException) as exc: + _validate_not_future(today_shanghai() + _dt.timedelta(days=1)) + assert exc.value.status_code == 400 + + +def test_invalid_date_format_rejected(): + """Invalid date format must return 400.""" + from app.routers.finance import _resolve_date + from fastapi import HTTPException + import pytest + + with pytest.raises(HTTPException) as exc: + _resolve_date("not-a-date") + assert exc.value.status_code == 400 + + +def test_partial_failure_still_saved(monkeypatch): + """Partially failed results are saved, with success=False and failed_count.""" + from app.services.finance_service import get_or_create_daily_summary + + db = _make_inmemory_db() + _add_minimal_test_data(db) + + # Override the first website to fail + original_revenue_calls = [] + def mock_website_revenue(website, _date): + original_revenue_calls.append(website.id) + if website.id == 1: + return (0.0, "timeout") + return (50.0, None) + + monkeypatch.setattr("app.services.finance_service.fetch_website_revenue", mock_website_revenue) + monkeypatch.setattr("app.services.finance_service.fetch_upstream_cost_sub2api", lambda *_: (40.0, None)) + + result = get_or_create_daily_summary(db, date(2026, 7, 2)) + + assert result["success"] is False + assert result["failed_count"] >= 1 + + from app.models.finance_daily_summary import FinanceDailySummary + row = db.query(FinanceDailySummary).filter( + FinanceDailySummary.stat_date == date(2026, 7, 2) + ).first() + assert row is not None + assert row.success is False + db.close() + + +def test_save_summary_if_absent_saves_new(monkeypatch): + """save_summary_if_absent computes and saves when no snapshot exists.""" + from app.services.finance_service import save_summary_if_absent + + db = _make_inmemory_db() + _add_minimal_test_data(db) + monkeypatch.setattr("app.services.finance_service.fetch_website_revenue", lambda *_: (100.0, None)) + monkeypatch.setattr("app.services.finance_service.fetch_upstream_cost_sub2api", lambda *_: (40.0, None)) + + saved = save_summary_if_absent(db, date(2026, 7, 2)) + + assert saved is True + from app.models.finance_daily_summary import FinanceDailySummary + row = db.query(FinanceDailySummary).filter( + FinanceDailySummary.stat_date == date(2026, 7, 2) + ).first() + assert row is not None + db.close() + + +def test_save_summary_if_absent_skips_existing(monkeypatch): + """save_summary_if_absent returns False and does nothing when snapshot exists.""" + from app.models.finance_daily_summary import FinanceDailySummary + from app.services.finance_service import save_summary_if_absent + + db = _make_inmemory_db() + + # Pre-save a row + row = FinanceDailySummary( + stat_date=date(2026, 7, 2), + summary_json='{"date":"2026-07-02"}', + success=True, + ) + db.add(row) + db.commit() + + # If it tries to compute, the monkeypatched function would raise + def _fail(*a, **kw): + raise RuntimeError("should not be called") + monkeypatch.setattr("app.services.finance_service.fetch_website_revenue", _fail) + + saved = save_summary_if_absent(db, date(2026, 7, 2)) + + assert saved is False + db.close() + + +def _make_test_summary(website_items, upstream_items=None): + """Return a summary dict with deep-copied items to prevent cross-contamination.""" + return { + "total_revenue": 100.0, "total_cost": 40.0, + "net_income": 60.0, "success": True, + "website_items": [dict(it) for it in website_items], + "upstream_items": [dict(it) for it in (upstream_items or [ + {"id": 1, "name": "U1", "upstream_type": "sub2api", "amount": 40.0, "status": "success", "error": None}, + ])], + } + + +def test_has_difference_by_identity(): + """_has_difference compares items by (id, upstream_type), not by position.""" + from app.routers.finance import _has_difference + + items = [ + {"id": 1, "name": "A", "upstream_type": "sub2api", "amount": 60.0, "status": "success", "error": None}, + {"id": 2, "name": "B", "upstream_type": "sub2api", "amount": 40.0, "status": "success", "error": None}, + ] + stored = _make_test_summary(items) + current = _make_test_summary(list(reversed(items))) + + assert _has_difference(stored, current) is False + + # Different amount on same identity — should be different + current["website_items"][1]["amount"] = 50.0 # item id=1 amount changed + assert _has_difference(stored, current) is True + + # Different item count — should be different + current["website_items"].pop() + assert _has_difference(stored, current) is True + + +def test_compute_diff_items_unchanged(): + """_compute_diff_items returns empty diff_items when data matches.""" + from app.routers.finance import _compute_diff_items + + items = [ + {"id": 1, "name": "A", "upstream_type": "sub2api", "amount": 60.0, "status": "success", "error": None}, + {"id": 2, "name": "B", "upstream_type": "sub2api", "amount": 40.0, "status": "success", "error": None}, + ] + stored = _make_test_summary(items) + current = _make_test_summary(list(reversed(items))) + + diff_items, has_diff = _compute_diff_items(stored, current) + assert has_diff is False + assert diff_items == [] + + +def test_compute_diff_items_amount_change(): + """_compute_diff_items returns the changed item when amount differs.""" + from app.routers.finance import _compute_diff_items + + stored = _make_test_summary([ + {"id": 1, "name": "A", "upstream_type": "sub2api", "amount": 60.0, "status": "success", "error": None}, + ]) + current = _make_test_summary([ + {"id": 1, "name": "A", "upstream_type": "sub2api", "amount": 55.0, "status": "success", "error": None}, + ]) + + diff_items, has_diff = _compute_diff_items(stored, current) + assert has_diff is True + assert len(diff_items) == 1 + d = diff_items[0] + assert d["side"] == "website_items" + assert d["id"] == 1 + assert d["name"] == "A" + assert d["old_amount"] == 60.0 + assert d["new_amount"] == 55.0 + + +def test_compute_diff_items_added_and_removed(): + """_compute_diff_items handles items added or removed between snapshots.""" + from app.routers.finance import _compute_diff_items + + stored = _make_test_summary([ + {"id": 1, "name": "A", "upstream_type": "sub2api", "amount": 60.0, "status": "success", "error": None}, + ]) + current = _make_test_summary([ + {"id": 1, "name": "A", "upstream_type": "sub2api", "amount": 60.0, "status": "success", "error": None}, + {"id": 2, "name": "B", "upstream_type": "sub2api", "amount": 40.0, "status": "success", "error": None}, + ]) + + diff_items, has_diff = _compute_diff_items(stored, current) + assert has_diff is True + # B is new — find it + added = [d for d in diff_items if d["name"] == "B"] + assert len(added) == 1 + assert added[0]["old_amount"] is None + assert added[0]["new_amount"] == 40.0 diff --git a/frontend/src/views/Finance.vue b/frontend/src/views/Finance.vue index 48f873e..ec6d938 100644 --- a/frontend/src/views/Finance.vue +++ b/frontend/src/views/Finance.vue @@ -7,7 +7,7 @@ 财务对账 -

实时计算，数据不入库；失败项不计入利润，避免虚高

+

已统计入库数据，定时快照不自动覆盖

刷新 + + + 重新统计并比对 +
@@ -34,12 +43,17 @@
网站收入
{{ formatAmount(summary.total_revenue) }}
-
{{ summary.date }}
+
+ {{ summary.date }} + + · 快照 {{ formatTime(summary.computed_at) }} + +
上游消费
{{ formatAmount(summary.total_cost) }}
-
{{ enabledUpstreamCount }} 个上游
+
{{ upstreamItemCount }} 个上游
净收入
@@ -70,8 +84,7 @@ 网站收入明细 - {{ summary.website_items.filter(i => i.status === 'success').length }} / - {{ summary.website_items.length }} 成功 + {{ successWebsiteCount }} / {{ summary.website_items.length }} 成功
@@ -108,8 +121,7 @@ 上游消费明细 - {{ summary.upstream_items.filter(i => i.status === 'success').length }} / - {{ summary.upstream_items.length }} 成功 + {{ successUpstreamCount }} / {{ summary.upstream_items.length }} 成功 @@ -148,13 +160,128 @@ + + + + + +