From 1670503720ec6955ab7ccc82a8f5eb1cf063ae7a Mon Sep 17 00:00:00 2001 From: SmartUp Developer Date: Fri, 3 Jul 2026 20:05:28 +0800 Subject: [PATCH] fix finance balance delta recharges --- backend/app/database.py | 6 +- backend/app/models/snapshot.py | 15 +- backend/app/models/upstream.py | 2 + backend/app/models/upstream_recharge_event.py | 24 +++ backend/app/routers/upstreams.py | 125 +++++++++++- backend/app/schemas/upstream.py | 29 ++- backend/app/services/finance_service.py | 119 ++++++++++- backend/app/services/scheduler.py | 59 +++++- backend/app/services/snapshot_service.py | 15 +- backend/test_finance_daily_summary.py | 169 +++++++++++++++- frontend/src/api/index.ts | 22 +++ frontend/src/views/Upstreams.vue | 184 +++++++++++++++++- 12 files changed, 745 insertions(+), 24 deletions(-) create mode 100644 backend/app/models/upstream_recharge_event.py diff --git a/backend/app/database.py b/backend/app/database.py index b55e5f0..8b830e4 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, finance_daily_summary # 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, upstream_recharge_event # noqa: F401 Base.metadata.create_all(bind=engine) _ensure_indexes() _migrate_custom_pages() @@ -60,6 +60,8 @@ _NEW_INDEXES = [ "CREATE INDEX IF NOT EXISTS ix_key_upstream_name ON upstream_generated_keys(upstream_id, key_name)", "CREATE INDEX IF NOT EXISTS ix_external_api_logs_created_at_desc ON external_api_logs(created_at DESC)", "CREATE INDEX IF NOT EXISTS ix_external_api_logs_direction_created ON external_api_logs(direction, created_at DESC)", + "CREATE INDEX IF NOT EXISTS ix_balance_snap_upstream_captured ON upstream_balance_snapshots(upstream_id, captured_at DESC)", + "CREATE INDEX IF NOT EXISTS ix_recharge_upstream_date ON upstream_recharge_events(upstream_id, recharge_date)", ] @@ -132,6 +134,8 @@ def _migrate_upstreams(): conn.execute(text("ALTER TABLE upstreams ADD COLUMN balance_alert_threshold FLOAT")) if "balance_alert_notified" not in columns: conn.execute(text("ALTER TABLE upstreams ADD COLUMN balance_alert_notified BOOLEAN NOT NULL DEFAULT 0")) + if "finance_cost_mode" not in columns: + conn.execute(text("ALTER TABLE upstreams ADD COLUMN finance_cost_mode VARCHAR(32) NOT NULL DEFAULT 'usage_stats'")) def _migrate_upstream_generated_keys(): diff --git a/backend/app/models/snapshot.py b/backend/app/models/snapshot.py index e32dbcd..c04ef32 100644 --- a/backend/app/models/snapshot.py +++ b/backend/app/models/snapshot.py @@ -1,5 +1,5 @@ from datetime import datetime, timezone -from sqlalchemy import Index, Integer, Text, DateTime, ForeignKey, text +from sqlalchemy import Float, Index, Integer, Text, DateTime, ForeignKey, text from sqlalchemy.orm import mapped_column, Mapped from app.database import Base @@ -15,3 +15,16 @@ class UpstreamRateSnapshot(Base): __table_args__ = ( Index("ix_snapshot_upstream_captured", "upstream_id", text("captured_at DESC")), ) + + +class UpstreamBalanceSnapshot(Base): + __tablename__ = "upstream_balance_snapshots" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + upstream_id: Mapped[int] = mapped_column(Integer, ForeignKey("upstreams.id", ondelete="CASCADE"), index=True) + balance: Mapped[float] = mapped_column(Float, nullable=False) + captured_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc), index=True) + + __table_args__ = ( + Index("ix_balance_snap_upstream_captured", "upstream_id", text("captured_at DESC")), + ) diff --git a/backend/app/models/upstream.py b/backend/app/models/upstream.py index 07cf5eb..3283723 100644 --- a/backend/app/models/upstream.py +++ b/backend/app/models/upstream.py @@ -35,6 +35,8 @@ class Upstream(Base): # Balance alert balance_alert_threshold: Mapped[Optional[float]] = mapped_column(Float, nullable=True) balance_alert_notified: Mapped[bool] = mapped_column(Boolean, default=False) + # Finance cost mode: "usage_stats" (default) or "balance_delta" + finance_cost_mode: Mapped[str] = mapped_column(String(32), default="usage_stats") created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc)) updated_at: Mapped[datetime] = mapped_column( DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc) diff --git a/backend/app/models/upstream_recharge_event.py b/backend/app/models/upstream_recharge_event.py new file mode 100644 index 0000000..f65fcf5 --- /dev/null +++ b/backend/app/models/upstream_recharge_event.py @@ -0,0 +1,24 @@ +from datetime import date, datetime, timezone + +from sqlalchemy import Date, DateTime, Float, ForeignKey, Index, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base + + +class UpstreamRechargeEvent(Base): + __tablename__ = "upstream_recharge_events" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + upstream_id: Mapped[int] = mapped_column(Integer, ForeignKey("upstreams.id", ondelete="CASCADE"), index=True) + recharge_date: Mapped[date] = mapped_column(Date, nullable=False) + amount: Mapped[float] = mapped_column(Float, nullable=False) + note: Mapped[str] = mapped_column(Text, default="") + created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc)) + updated_at: Mapped[datetime] = mapped_column( + DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + Index("ix_recharge_upstream_date", "upstream_id", "recharge_date"), + ) diff --git a/backend/app/routers/upstreams.py b/backend/app/routers/upstreams.py index 36a33a5..cf7eabc 100644 --- a/backend/app/routers/upstreams.py +++ b/backend/app/routers/upstreams.py @@ -4,18 +4,19 @@ from __future__ import annotations import json import logging import re -from datetime import datetime, timezone +from datetime import date as date_type, datetime, timezone from typing import Any, List logger = logging.getLogger(__name__) -from fastapi import APIRouter, Depends, HTTPException, status +from fastapi import APIRouter, Depends, HTTPException, Query as QueryParam, status from sqlalchemy.orm import Session from app.database import SessionLocal, get_db from app.models.admin_user import AdminUser from app.models.upstream import Upstream from app.models.upstream_key import UpstreamGeneratedKey +from app.models.upstream_recharge_event import UpstreamRechargeEvent from app.models.snapshot import UpstreamRateSnapshot from app.schemas.upstream import ( GenerateKeysByGroupsRequest, @@ -23,10 +24,12 @@ from app.schemas.upstream import ( GeneratedUpstreamKeyResponse, UpstreamCreate, UpstreamUpdate, UpstreamResponse, SnapshotResponse, TestResult, UpstreamBatchActionItem, UpstreamBatchActionSummary, UpstreamBatchActionResponse, + UpstreamRechargeCreate, UpstreamRechargeUpdate, UpstreamRechargeResponse, ) from app.services.upstream_client import UpstreamClient, UpstreamError, _PendingKeyError, build_snapshot, build_new_api_token_name, mask_secret, _extract_key_value from app.services.auth_config import MASK, mask_auth_config, normalize_auth_config -from app.services.snapshot_service import diff_snapshots +from app.services.finance_service import today_shanghai +from app.services.snapshot_service import diff_snapshots, write_balance_snapshot from app.services import scheduler as sched_svc from app.services import webhook_service from app.services import website_sync @@ -138,11 +141,29 @@ def _to_response(u: Upstream) -> UpstreamResponse: balance_response_path=u.balance_response_path or "", balance_divisor=u.balance_divisor or 1.0, balance_alert_threshold=u.balance_alert_threshold, + finance_cost_mode=u.finance_cost_mode or "usage_stats", created_at=u.created_at, updated_at=u.updated_at, ) +def _recharge_response(row: UpstreamRechargeEvent) -> UpstreamRechargeResponse: + return UpstreamRechargeResponse( + id=row.id, + upstream_id=row.upstream_id, + date=row.recharge_date, + amount=row.amount, + note=row.note or "", + created_at=row.created_at, + updated_at=row.updated_at, + ) + + +def _validate_recharge_date(value: date_type) -> None: + if value > today_shanghai(): + raise HTTPException(422, "date 不允许是未来日期") + + @router.get("", response_model=List[UpstreamResponse]) def list_upstreams(db: Session = Depends(get_db), _=Depends(get_current_user)): return [_to_response(u) for u in db.query(Upstream).order_by(Upstream.id).all()] @@ -694,6 +715,7 @@ def _test_upstream_core(db: Session, u: Upstream) -> UpstreamBatchActionItem: raw_balance = client.get_balance(u.balance_endpoint, u.balance_response_path) if raw_balance is not None: u.balance = raw_balance / (u.balance_divisor or 1.0) + write_balance_snapshot(db, u.id, u.balance) u.balance_updated_at = datetime.now(timezone.utc) if raw_balance is not None else None except Exception as exc: logger.warning("test-all: upstream %s balance failed: %s", u.name, exc) @@ -738,6 +760,7 @@ def _check_now_core(db: Session, u: Upstream) -> tuple[str, bool]: raw_balance = client.get_balance(u.balance_endpoint, u.balance_response_path) if raw_balance is not None: u.balance = raw_balance / (u.balance_divisor or 1.0) + write_balance_snapshot(db, u.id, u.balance) u.balance_updated_at = datetime.now(timezone.utc) if raw_balance is not None else None except Exception as exc: logger.warning("check-now: upstream %s balance failed: %s", u.name, exc) @@ -919,6 +942,7 @@ def create_upstream( balance_response_path=body.balance_response_path, balance_divisor=body.balance_divisor, balance_alert_threshold=body.balance_alert_threshold, + finance_cost_mode=body.finance_cost_mode, ) db.add(u) db.commit() @@ -935,6 +959,99 @@ def get_upstream(uid: int, db: Session = Depends(get_db), _=Depends(get_current_ return _to_response(u) +@router.get("/{uid}/recharges", response_model=List[UpstreamRechargeResponse]) +def list_recharges( + uid: int, + date: date_type | None = QueryParam(None), + db: Session = Depends(get_db), + _=Depends(get_current_user), +): + u = db.query(Upstream).filter(Upstream.id == uid).first() + if not u: + raise HTTPException(404, "upstream not found") + query = db.query(UpstreamRechargeEvent).filter(UpstreamRechargeEvent.upstream_id == uid) + if date is not None: + query = query.filter(UpstreamRechargeEvent.recharge_date == date) + rows = query.order_by(UpstreamRechargeEvent.recharge_date.desc(), UpstreamRechargeEvent.id.desc()).all() + return [_recharge_response(row) for row in rows] + + +@router.post("/{uid}/recharges", response_model=UpstreamRechargeResponse, status_code=201) +def create_recharge( + uid: int, + body: UpstreamRechargeCreate, + db: Session = Depends(get_db), + _=Depends(get_current_user), +): + u = db.query(Upstream).filter(Upstream.id == uid).first() + if not u: + raise HTTPException(404, "upstream not found") + _validate_recharge_date(body.date) + row = UpstreamRechargeEvent( + upstream_id=uid, + recharge_date=body.date, + amount=body.amount, + note=body.note or "", + ) + db.add(row) + db.commit() + db.refresh(row) + return _recharge_response(row) + + +@router.put("/{uid}/recharges/{rid}", response_model=UpstreamRechargeResponse) +def update_recharge( + uid: int, + rid: int, + body: UpstreamRechargeUpdate, + db: Session = Depends(get_db), + _=Depends(get_current_user), +): + u = db.query(Upstream).filter(Upstream.id == uid).first() + if not u: + raise HTTPException(404, "upstream not found") + row = ( + db.query(UpstreamRechargeEvent) + .filter(UpstreamRechargeEvent.id == rid, UpstreamRechargeEvent.upstream_id == uid) + .first() + ) + if not row: + raise HTTPException(404, "recharge event not found") + data = body.model_dump(exclude_unset=True) + if "date" in data: + _validate_recharge_date(data["date"]) + row.recharge_date = data["date"] + if "amount" in data: + row.amount = data["amount"] + if "note" in data: + row.note = data["note"] or "" + row.updated_at = datetime.now(timezone.utc) + db.commit() + db.refresh(row) + return _recharge_response(row) + + +@router.delete("/{uid}/recharges/{rid}", status_code=204) +def delete_recharge( + uid: int, + rid: int, + db: Session = Depends(get_db), + _=Depends(get_current_user), +): + u = db.query(Upstream).filter(Upstream.id == uid).first() + if not u: + raise HTTPException(404, "upstream not found") + row = ( + db.query(UpstreamRechargeEvent) + .filter(UpstreamRechargeEvent.id == rid, UpstreamRechargeEvent.upstream_id == uid) + .first() + ) + if not row: + raise HTTPException(404, "recharge event not found") + db.delete(row) + db.commit() + + @router.put("/{uid}", response_model=UpstreamResponse) def update_upstream( uid: int, @@ -1029,8 +1146,6 @@ def latest_snapshot(uid: int, db: Session = Depends(get_db), _=Depends(get_curre ) -from fastapi import Query as QueryParam - @router.get("/{uid}/snapshots", response_model=List[SnapshotResponse]) def list_snapshots( uid: int, diff --git a/backend/app/schemas/upstream.py b/backend/app/schemas/upstream.py index 1aeb85a..0063a83 100644 --- a/backend/app/schemas/upstream.py +++ b/backend/app/schemas/upstream.py @@ -1,5 +1,5 @@ -from datetime import datetime -from typing import Optional, Any +from datetime import date, datetime +from typing import Optional, Any, Literal from pydantic import BaseModel, Field @@ -37,6 +37,7 @@ class UpstreamCreate(BaseModel): balance_response_path: str = "" balance_divisor: float = 1.0 balance_alert_threshold: Optional[float] = None + finance_cost_mode: Literal["usage_stats", "balance_delta"] = "usage_stats" class UpstreamUpdate(BaseModel): @@ -54,6 +55,7 @@ class UpstreamUpdate(BaseModel): balance_response_path: Optional[str] = None balance_divisor: Optional[float] = None balance_alert_threshold: Optional[float] = None + finance_cost_mode: Optional[Literal["usage_stats", "balance_delta"]] = None class UpstreamResponse(BaseModel): @@ -77,12 +79,35 @@ class UpstreamResponse(BaseModel): balance_response_path: str = "" balance_divisor: float = 1.0 balance_alert_threshold: Optional[float] = None + finance_cost_mode: Literal["usage_stats", "balance_delta"] = "usage_stats" created_at: datetime updated_at: datetime model_config = {"from_attributes": True} +class UpstreamRechargeCreate(BaseModel): + date: date + amount: float = Field(gt=0) + note: str = "" + + +class UpstreamRechargeUpdate(BaseModel): + date: Optional[date] = None + amount: Optional[float] = Field(default=None, gt=0) + note: Optional[str] = None + + +class UpstreamRechargeResponse(BaseModel): + id: int + upstream_id: int + date: date + amount: float + note: str = "" + created_at: datetime + updated_at: datetime + + class SnapshotResponse(BaseModel): id: int upstream_id: int diff --git a/backend/app/services/finance_service.py b/backend/app/services/finance_service.py index 0f3b351..72c7f5a 100644 --- a/backend/app/services/finance_service.py +++ b/backend/app/services/finance_service.py @@ -21,10 +21,13 @@ from typing import Any import pytz +from sqlalchemy import func from sqlalchemy.orm import Session from app.models.upstream import Upstream from app.models.website import Website +from app.models.snapshot import UpstreamBalanceSnapshot +from app.models.upstream_recharge_event import UpstreamRechargeEvent from app.services.upstream_client import UpstreamClient from app.services.website_client import Sub2ApiWebsiteClient, WebsiteError @@ -81,6 +84,13 @@ def date_to_shanghai_timestamps(d: date) -> tuple[int, int]: return int(start.timestamp()), int(end.timestamp()) +def date_to_shanghai_utc_range(d: date) -> tuple[datetime, datetime]: + """Return UTC datetimes for [start, next_day_start) of a Shanghai calendar day.""" + start = SHANGHAI_TZ.localize(datetime(d.year, d.month, d.day, 0, 0, 0)) + end = start + timedelta(days=1) + return start.astimezone(timezone.utc), end.astimezone(timezone.utc) + + # ───────────────────────────────────────────── # Website revenue # ───────────────────────────────────────────── @@ -229,6 +239,86 @@ def fetch_upstream_cost_new_api(upstream: Upstream, target_date: date) -> tuple[ return 0.0, str(e) +# ───────────────────────────────────────────── +# Balance-delta cost (local balance snapshots) +# ───────────────────────────────────────────── + +def fetch_upstream_cost_balance_delta( + upstream: Upstream, + target_date: date, + db: Session, +) -> tuple[float, str | None]: + """Compute daily cost from local balance snapshots for balance_delta upstreams. + + Algorithm: + 1. Find the most recent snapshot BEFORE the start of target_date (baseline). + 2. Find snapshots DURING target_date and use the last one as ending balance. + 3. Sum manual recharge events for target_date. + 4. cost = baseline + recharge_total - ending. Negative values fail. + """ + start_dt, end_dt = date_to_shanghai_utc_range(target_date) + + # Baseline: latest snapshot before start of target date + baseline = ( + db.query(UpstreamBalanceSnapshot) + .filter( + UpstreamBalanceSnapshot.upstream_id == upstream.id, + UpstreamBalanceSnapshot.captured_at < start_dt, + ) + .order_by(UpstreamBalanceSnapshot.captured_at.desc()) + .first() + ) + + # Intra-day snapshots during target date + intra_day = ( + db.query(UpstreamBalanceSnapshot) + .filter( + UpstreamBalanceSnapshot.upstream_id == upstream.id, + UpstreamBalanceSnapshot.captured_at >= start_dt, + UpstreamBalanceSnapshot.captured_at < end_dt, + ) + .order_by(UpstreamBalanceSnapshot.captured_at.asc()) + .all() + ) + + if baseline is None: + return 0.0, "缺少目标日期前的余额基线样本，无法进行差分统计" + + if not intra_day: + return 0.0, "目标日期内无余额样本，无法进行差分统计" + + recharge_total = ( + db.query(func.coalesce(func.sum(UpstreamRechargeEvent.amount), 0.0)) + .filter( + UpstreamRechargeEvent.upstream_id == upstream.id, + UpstreamRechargeEvent.recharge_date == target_date, + ) + .scalar() + ) + recharge_total = float(recharge_total or 0.0) + + max_balance_rise = 0.0 + prev = float(baseline.balance) + for snap in intra_day: + current = float(snap.balance) + if current > prev: + max_balance_rise = max(max_balance_rise, current - prev) + prev = current + + if max_balance_rise > 0 and recharge_total <= 0: + return 0.0, "检测到余额上涨，但当天无充值记录；请补录充值后重新对账" + + if max_balance_rise > 0 and recharge_total + 1e-9 < max_balance_rise: + return 0.0, "当天充值总额小于观察到的最大余额上涨，充值金额可能漏填" + + ending = float(intra_day[-1].balance) + total = float(baseline.balance) + recharge_total - ending + if total < -1e-9: + return 0.0, "余额差分计算为负数，请检查余额样本或补录充值记录" + + return round(max(total, 0.0), 6), None + + # ───────────────────────────────────────────── # Orchestration # ───────────────────────────────────────────── @@ -237,6 +327,7 @@ def get_daily_summary( websites: list[Website], upstreams: list[Upstream], target_date: date, + db: Session | None = None, ) -> dict[str, Any]: """Compute daily revenue vs cost summary. @@ -264,16 +355,24 @@ def get_daily_summary( total_revenue += amount for u in upstreams: - utype = classify_upstream(u) - if utype == "sub2api": - amount, err = fetch_upstream_cost_sub2api(u, target_date) - elif utype == "new_api": - amount, err = fetch_upstream_cost_new_api(u, target_date) + cost_mode = getattr(u, "finance_cost_mode", "usage_stats") + if cost_mode == "balance_delta": + if db is None: + amount, err = 0.0, "余额差分统计需要数据库连接" + else: + amount, err = fetch_upstream_cost_balance_delta(u, target_date, db) + utype = classify_upstream(u) else: - amount, err = 0.0, ( - f"未知上游类型（auth_type={u.auth_type}, api_prefix={u.api_prefix}），" - "无法统计消费。请检查上游配置。" - ) + utype = classify_upstream(u) + if utype == "sub2api": + amount, err = fetch_upstream_cost_sub2api(u, target_date) + elif utype == "new_api": + amount, err = fetch_upstream_cost_new_api(u, target_date) + else: + amount, err = 0.0, ( + f"未知上游类型（auth_type={u.auth_type}, api_prefix={u.api_prefix}），" + "无法统计消费。请检查上游配置。" + ) item = { "id": u.id, @@ -339,7 +438,7 @@ def compute_daily_summary( """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) + return get_daily_summary(websites, upstreams, target_date, db=db) def _load_summary(row: Any) -> dict[str, Any]: diff --git a/backend/app/services/scheduler.py b/backend/app/services/scheduler.py index f043953..c03c28c 100644 --- a/backend/app/services/scheduler.py +++ b/backend/app/services/scheduler.py @@ -16,7 +16,7 @@ from app.models.upstream import Upstream from app.models.snapshot import UpstreamRateSnapshot from app.services.auth_config import normalize_auth_config from app.services.upstream_client import UpstreamClient, build_snapshot -from app.services.snapshot_service import diff_snapshots, prune_snapshots +from app.services.snapshot_service import diff_snapshots, prune_snapshots, write_balance_snapshot from app.services import webhook_service from app.services import website_sync from app.config import get_settings @@ -102,6 +102,7 @@ def _check_upstream(upstream_id: int) -> None: if balance is not None: upstream.balance = balance upstream.balance_updated_at = datetime.now(timezone.utc) + write_balance_snapshot(db, upstream.id, balance) # ── 余额告警阈值检查 ── threshold = upstream.balance_alert_threshold if threshold is not None and threshold > 0: @@ -376,6 +377,16 @@ def start_scheduler() -> None: max_instances=1, misfire_grace_time=3600, ) + # Daily balance snapshot capture at 23:59 Asia/Shanghai + _scheduler.add_job( + _capture_balance_snapshots, + trigger=CronTrigger(timezone="Asia/Shanghai", hour=23, minute=59), + id="balance_snapshot_daily", + replace_existing=True, + coalesce=True, + max_instances=1, + misfire_grace_time=600, + ) logger.info("scheduler started with %d upstream job(s)", len(upstreams)) finally: db.close() @@ -416,6 +427,52 @@ def _compute_finance_daily_summary() -> None: db.close() +def _capture_balance_snapshots() -> None: + """Fetch balance for all balance_delta-mode upstreams and write a snapshot. + + Runs daily at 23:59 Asia/Shanghai to capture the end-of-day boundary balance. + Only writes a snapshot if balance fetch succeeds. Does NOT update the + upstream.balance field (that's the regular check's job). + """ + db = SessionLocal() + try: + upstreams = db.query(Upstream).filter( + Upstream.enabled == True, + Upstream.finance_cost_mode == "balance_delta", + Upstream.balance_endpoint != "", + Upstream.balance_response_path != "", + ).all() + for u in upstreams: + try: + auth_config = json.loads(u.auth_config_json or "{}") + with UpstreamClient( + base_url=u.base_url, + api_prefix=u.api_prefix, + auth_type=u.auth_type, + auth_config=auth_config, + timeout=float(u.timeout_seconds), + on_auth_config_update=lambda updated: _persist_auth_config_update(u, updated), + target_id=u.id, + target_name=u.name, + ) as client: + client.ensure_authenticated() + raw_balance = client.get_balance(u.balance_endpoint, u.balance_response_path) + if raw_balance is not None: + balance = raw_balance / (u.balance_divisor or 1.0) + write_balance_snapshot(db, u.id, balance) + db.commit() + logger.info("balance boundary snapshot for upstream %s: %.4f", u.name, balance) + else: + logger.warning("balance boundary snapshot: upstream %s returned None", u.name) + except Exception as exc: + db.rollback() + logger.warning("balance boundary snapshot failed for upstream %s: %s", u.name, exc) + except Exception: + logger.exception("failed to capture balance boundary snapshots") + finally: + db.close() + + def stop_scheduler() -> None: if _scheduler.running: _scheduler.shutdown(wait=True) diff --git a/backend/app/services/snapshot_service.py b/backend/app/services/snapshot_service.py index 9fd05b9..55b3d45 100644 --- a/backend/app/services/snapshot_service.py +++ b/backend/app/services/snapshot_service.py @@ -1,9 +1,10 @@ -"""Snapshot diff logic.""" +"""Snapshot diff logic and balance snapshot writes.""" +from datetime import datetime, timezone from typing import Any, Optional from sqlalchemy.orm import Session -from app.models.snapshot import UpstreamRateSnapshot +from app.models.snapshot import UpstreamBalanceSnapshot, UpstreamRateSnapshot def diff_snapshots( @@ -43,6 +44,16 @@ def diff_snapshots( return changes +def write_balance_snapshot(db: Session, upstream_id: int, balance: float) -> None: + """Write a balance snapshot row. Call only on successful balance fetch.""" + row = UpstreamBalanceSnapshot( + upstream_id=upstream_id, + balance=balance, + captured_at=datetime.now(timezone.utc), + ) + db.add(row) + + def prune_snapshots(db: Session, upstream_id: int, keep: int) -> None: if keep <= 0: return diff --git a/backend/test_finance_daily_summary.py b/backend/test_finance_daily_summary.py index e5f3a62..c9c9e84 100644 --- a/backend/test_finance_daily_summary.py +++ b/backend/test_finance_daily_summary.py @@ -4,6 +4,7 @@ from __future__ import annotations import datetime as _dt import json from datetime import date +from datetime import datetime, timezone from unittest.mock import MagicMock import httpx @@ -13,6 +14,7 @@ from app.services.finance_service import ( date_to_shanghai_timestamps, fetch_upstream_cost_sub2api, fetch_upstream_cost_new_api, + fetch_upstream_cost_balance_delta, fetch_website_revenue, get_daily_summary, ) @@ -31,6 +33,7 @@ class FakeUpstream: self.auth_type = kw.get("auth_type", "bearer") self.auth_config_json = kw.get("auth_config_json", json.dumps({"token": "tok"})) self.timeout_seconds = kw.get("timeout_seconds", 30) + self.finance_cost_mode = kw.get("finance_cost_mode", "usage_stats") self.enabled = True @@ -407,6 +410,163 @@ def test_get_daily_summary_all_success(monkeypatch): assert abs(result["margin_percent"] - 60.0) < 1e-6 +def _add_balance_snapshot(db, upstream_id: int, balance: float, captured_at: datetime): + from app.models.snapshot import UpstreamBalanceSnapshot + + row = UpstreamBalanceSnapshot( + upstream_id=upstream_id, + balance=balance, + captured_at=captured_at, + ) + db.add(row) + db.commit() + return row + + +def _add_recharge_event(db, upstream_id: int, recharge_date: date, amount: float, note: str = ""): + from app.models.upstream_recharge_event import UpstreamRechargeEvent + + row = UpstreamRechargeEvent( + upstream_id=upstream_id, + recharge_date=recharge_date, + amount=amount, + note=note, + ) + db.add(row) + db.commit() + return row + + +def test_balance_delta_cost_uses_manual_recharge_and_ending_balance(): + """Balance-delta cost is baseline + recharge_total - ending.""" + db = _make_inmemory_db() + u = FakeUpstream(id=1, finance_cost_mode="balance_delta") + + _add_balance_snapshot(db, 1, 100.0, datetime(2026, 7, 1, 15, 55, tzinfo=timezone.utc)) + _add_balance_snapshot(db, 1, 90.0, datetime(2026, 7, 1, 16, 30, tzinfo=timezone.utc)) + _add_balance_snapshot(db, 1, 110.0, datetime(2026, 7, 1, 18, 0, tzinfo=timezone.utc)) + _add_balance_snapshot(db, 1, 80.0, datetime(2026, 7, 2, 15, 59, tzinfo=timezone.utc)) + _add_recharge_event(db, 1, date(2026, 7, 2), 20.0) + + amount, err = fetch_upstream_cost_balance_delta(u, date(2026, 7, 2), db) + + assert err is None + assert abs(amount - 40.0) < 1e-6 + db.close() + + +def test_balance_delta_balance_rise_without_recharge_fails(): + db = _make_inmemory_db() + u = FakeUpstream(id=1, finance_cost_mode="balance_delta") + _add_balance_snapshot(db, 1, 100.0, datetime(2026, 7, 1, 15, 55, tzinfo=timezone.utc)) + _add_balance_snapshot(db, 1, 120.0, datetime(2026, 7, 1, 16, 30, tzinfo=timezone.utc)) + + amount, err = fetch_upstream_cost_balance_delta(u, date(2026, 7, 2), db) + + assert amount == 0.0 + assert "无充值记录" in err + db.close() + + +def test_balance_delta_recharge_less_than_observed_rise_fails(): + db = _make_inmemory_db() + u = FakeUpstream(id=1, finance_cost_mode="balance_delta") + _add_balance_snapshot(db, 1, 100.0, datetime(2026, 7, 1, 15, 55, tzinfo=timezone.utc)) + _add_balance_snapshot(db, 1, 130.0, datetime(2026, 7, 1, 16, 30, tzinfo=timezone.utc)) + _add_recharge_event(db, 1, date(2026, 7, 2), 20.0) + + amount, err = fetch_upstream_cost_balance_delta(u, date(2026, 7, 2), db) + + assert amount == 0.0 + assert "充值金额可能漏填" in err + db.close() + + +def test_balance_delta_missing_baseline_fails(): + """Balance-delta requires a pre-day baseline sample.""" + db = _make_inmemory_db() + u = FakeUpstream(id=1, finance_cost_mode="balance_delta") + _add_balance_snapshot(db, 1, 90.0, datetime(2026, 7, 1, 16, 30, tzinfo=timezone.utc)) + + amount, err = fetch_upstream_cost_balance_delta(u, date(2026, 7, 2), db) + + assert amount == 0.0 + assert "基线样本" in err + db.close() + + +def test_get_daily_summary_balance_delta_failure_excluded_from_total(): + """Incomplete balance-delta data must not be counted as zero-cost success.""" + db = _make_inmemory_db() + u = FakeUpstream(id=1, api_prefix="api/v1", auth_type="bearer", finance_cost_mode="balance_delta") + + result = get_daily_summary([], [u], date(2026, 7, 2), db=db) + + assert result["failed_count"] == 1 + assert result["total_cost"] == 0.0 + assert result["upstream_items"][0]["status"] == "failed" + assert "基线样本" in result["upstream_items"][0]["error"] + db.close() + + +def test_upstream_recharge_crud_api_validates_and_scopes_to_upstream(): + from fastapi.testclient import TestClient + + from app.database import get_db + from app.main import app + from app.models.upstream import Upstream + from app.utils.auth import get_current_user + + db = _make_inmemory_db() + db.add_all([ + Upstream(id=1, name="U1", base_url="http://u1", api_prefix="api/v1", auth_type="bearer", auth_config_json="{}"), + Upstream(id=2, name="U2", base_url="http://u2", api_prefix="api/v1", auth_type="bearer", auth_config_json="{}"), + ]) + db.commit() + + app.dependency_overrides[get_db] = lambda: db + app.dependency_overrides[get_current_user] = lambda: None + try: + client = TestClient(app) + + bad_amount = client.post("/api/upstreams/1/recharges", json={"date": "2026-07-02", "amount": 0}) + assert bad_amount.status_code == 422 + + future = client.post("/api/upstreams/1/recharges", json={"date": "2999-01-01", "amount": 10}) + assert future.status_code == 422 + + created = client.post( + "/api/upstreams/1/recharges", + json={"date": "2026-07-02", "amount": 50.5, "note": "manual"}, + ) + assert created.status_code == 201 + body = created.json() + rid = body["id"] + assert body["date"] == "2026-07-02" + assert body["amount"] == 50.5 + + listed = client.get("/api/upstreams/1/recharges", params={"date": "2026-07-02"}) + assert listed.status_code == 200 + assert [row["id"] for row in listed.json()] == [rid] + + wrong_upstream = client.put(f"/api/upstreams/2/recharges/{rid}", json={"amount": 60}) + assert wrong_upstream.status_code == 404 + + updated = client.put(f"/api/upstreams/1/recharges/{rid}", json={"amount": 60, "note": "fixed"}) + assert updated.status_code == 200 + assert updated.json()["amount"] == 60 + assert updated.json()["note"] == "fixed" + + deleted = client.delete(f"/api/upstreams/1/recharges/{rid}") + assert deleted.status_code == 204 + + empty = client.get("/api/upstreams/1/recharges", params={"date": "2026-07-02"}) + assert empty.json() == [] + finally: + app.dependency_overrides.clear() + db.close() + + # ───────────────────────────────────────────── # DB-backed snapshot tests (FinanceDailySummary) # ───────────────────────────────────────────── @@ -415,14 +575,21 @@ 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 sqlalchemy.pool import StaticPool 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 import snapshot # noqa: F401 + from app.models import upstream_recharge_event # 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}) + engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) Base.metadata.create_all(bind=engine) Session = sessionmaker(bind=engine) return Session() diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index 0013f53..10856b0 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -68,6 +68,8 @@ export const authApi = { } // ——— Upstreams ——— +export type FinanceCostMode = 'usage_stats' | 'balance_delta' + export interface UpstreamData { id: number name: string @@ -89,6 +91,7 @@ export interface UpstreamData { balance_response_path: string balance_divisor: number balance_alert_threshold: number | null + finance_cost_mode: FinanceCostMode created_at: string updated_at: string } @@ -108,6 +111,7 @@ export interface UpstreamForm { balance_response_path: string balance_divisor: number balance_alert_threshold: number | null + finance_cost_mode: FinanceCostMode } export interface GeneratedUpstreamKey { @@ -128,6 +132,16 @@ export interface GeneratedUpstreamKey { has_key_value: boolean } +export interface UpstreamRecharge { + id: number + upstream_id: number + date: string + amount: number + note: string + created_at: string + updated_at: string +} + export interface GenerateKeysByGroupsForm { group_ids: string[] name_prefix: string @@ -169,6 +183,14 @@ export const upstreamsApi = { test: (id: number) => api.post<{ success: boolean; message: string; detail?: string }>(`/api/upstreams/${id}/test`), checkNow: (id: number) => api.post<{ success: boolean; message: string }>(`/api/upstreams/${id}/check-now`), generatedKeys: (id: number) => api.get(`/api/upstreams/${id}/generated-keys`), + listRecharges: (id: number, date?: string) => + api.get(`/api/upstreams/${id}/recharges`, { params: date ? { date } : undefined }), + createRecharge: (id: number, data: { date: string; amount: number; note?: string }) => + api.post(`/api/upstreams/${id}/recharges`, data), + updateRecharge: (id: number, rechargeId: number, data: { date?: string; amount?: number; note?: string }) => + api.put(`/api/upstreams/${id}/recharges/${rechargeId}`, data), + deleteRecharge: (id: number, rechargeId: number) => + api.delete(`/api/upstreams/${id}/recharges/${rechargeId}`), generateKeysByGroups: (id: number, data: GenerateKeysByGroupsForm) => api.post<{ success: boolean; message: string; items: GeneratedUpstreamKey[] }>(`/api/upstreams/${id}/keys/generate-by-groups`, data), latestSnapshot: (id: number) => api.get(`/api/upstreams/${id}/snapshots/latest`), diff --git a/frontend/src/views/Upstreams.vue b/frontend/src/views/Upstreams.vue index ae70f15..e461e31 100644 --- a/frontend/src/views/Upstreams.vue +++ b/frontend/src/views/Upstreams.vue @@ -247,6 +247,15 @@
余额低于此值时发送 Webhook 通知，留空/0 表示不监控
+ + + + + +
+ 远端会清理使用记录的上游可选余额差分。需配置余额接口；首个完整采样日后才可用于对账。 +
+ @@ -310,6 +319,76 @@ {{ detailUpstream.last_error }} +
+ + 充值记录 + 列表合计 {{ formatBalance(rechargeTotal) }} +
+ + + + + + + + + + + + + + + + + + + +
+ 取消编辑 + + {{ editingRechargeId ? '保存充值' : '添加充值' }} + +
+ + + + + + + + + + + + + +
已创建 Key @@ -463,7 +542,7 @@ import { ElMessage, ElMessageBox } from 'element-plus' import type { FormInstance } from 'element-plus' import dayjs from 'dayjs' import { Refresh, Plus, Edit, List, Delete, Warning, Clock, ArrowRight, Pointer, Key } from '@element-plus/icons-vue' -import { upstreamsApi, type AuthCaptureCandidate, type GeneratedUpstreamKey, type UpstreamData, type UpstreamBatchActionResponse } from '@/api' +import { upstreamsApi, type AuthCaptureCandidate, type FinanceCostMode, type GeneratedUpstreamKey, type UpstreamData, type UpstreamBatchActionResponse, type UpstreamRecharge } from '@/api' import AuthCaptureDialog from '@/components/AuthCaptureDialog.vue' const list = ref<(UpstreamData & { _testing?: boolean; _checking?: boolean })[]>([]) @@ -551,6 +630,7 @@ const defaultForm = () => ({ balance_response_path: platformDefaults.sub2api.balance_response_path, balance_divisor: platformDefaults.sub2api.balance_divisor, balance_alert_threshold: null as number | null, + finance_cost_mode: 'usage_stats' as FinanceCostMode, }) const form = ref(defaultForm()) const rules = { @@ -862,11 +942,20 @@ const detailVisible = ref(false) const detailUpstream = ref(null) const snapshots = ref([]) const generatedKeys = ref([]) +const recharges = ref([]) const snapshotLoading = ref(false) const keysLoading = ref(false) +const rechargeLoading = ref(false) +const rechargeSaving = ref(false) const expandedId = ref(null) const snapshotOffset = ref(0) const snapshotLimit = 20 +const editingRechargeId = ref(null) +const rechargeForm = ref({ + date: dayjs().format('YYYY-MM-DD'), + amount: null as number | null, + note: '', +}) const keyDialogVisible = ref(false) const keyTarget = ref(null) @@ -900,6 +989,10 @@ const healthyRate = computed(() => { const pendingChecks = computed(() => list.value.filter((item) => !item.last_checked_at).length) +const rechargeTotal = computed(() => + recharges.value.reduce((sum, item) => sum + (Number(item.amount) || 0), 0), +) + function usesTokenEndpointUpstream(row: UpstreamData | null) { if (!row) return false return row.api_prefix === '' @@ -1006,6 +1099,7 @@ function openEdit(row: UpstreamData) { balance_response_path: row.balance_response_path || '', balance_divisor: row.balance_divisor ?? 1.0, balance_alert_threshold: row.balance_alert_threshold ?? null, + finance_cost_mode: row.finance_cost_mode || 'usage_stats', } drawerVisible.value = true } @@ -1080,8 +1174,10 @@ function openDetail(row: UpstreamData) { detailUpstream.value = row snapshots.value = [] generatedKeys.value = [] + recharges.value = [] snapshotOffset.value = 0 expandedId.value = null + resetRechargeForm() _groupRowsCache.clear() detailVisible.value = true } @@ -1100,6 +1196,7 @@ async function loadGeneratedKeys() { async function loadSnapshots() { if (!detailUpstream.value) return loadGeneratedKeys() + loadRecharges() snapshotLoading.value = true try { const res = await upstreamsApi.listSnapshots(detailUpstream.value.id, snapshotLimit, snapshotOffset.value) @@ -1115,6 +1212,79 @@ async function loadSnapshots() { } } +function resetRechargeForm() { + editingRechargeId.value = null + rechargeForm.value = { + date: dayjs().format('YYYY-MM-DD'), + amount: null, + note: '', + } +} + +function disableFutureDate(time: Date) { + return dayjs(time).isAfter(dayjs(), 'day') +} + +async function loadRecharges() { + if (!detailUpstream.value) return + rechargeLoading.value = true + try { + const res = await upstreamsApi.listRecharges(detailUpstream.value.id) + recharges.value = res.data + } catch (e: any) { + ElMessage.error(e.response?.data?.detail || '加载充值记录失败') + } finally { + rechargeLoading.value = false + } +} + +function editRecharge(row: UpstreamRecharge) { + editingRechargeId.value = row.id + rechargeForm.value = { + date: row.date, + amount: row.amount, + note: row.note || '', + } +} + +async function saveRecharge() { + if (!detailUpstream.value || !rechargeForm.value.date || !rechargeForm.value.amount || rechargeForm.value.amount <= 0) return + rechargeSaving.value = true + try { + const payload = { + date: rechargeForm.value.date, + amount: rechargeForm.value.amount, + note: rechargeForm.value.note, + } + if (editingRechargeId.value) { + await upstreamsApi.updateRecharge(detailUpstream.value.id, editingRechargeId.value, payload) + ElMessage.success('充值记录已更新') + } else { + await upstreamsApi.createRecharge(detailUpstream.value.id, payload) + ElMessage.success('充值记录已添加') + } + resetRechargeForm() + await loadRecharges() + } catch (e: any) { + ElMessage.error(e.response?.data?.detail || '保存充值记录失败') + } finally { + rechargeSaving.value = false + } +} + +async function confirmDeleteRecharge(row: UpstreamRecharge) { + if (!detailUpstream.value) return + try { + await ElMessageBox.confirm(`确认删除 ${row.date} 的充值记录？`, '删除确认', { type: 'warning' }) + await upstreamsApi.deleteRecharge(detailUpstream.value.id, row.id) + ElMessage.success('充值记录已删除') + if (editingRechargeId.value === row.id) resetRechargeForm() + await loadRecharges() + } catch { + // noop + } +} + async function openKeyGenerate(row: UpstreamData) { keyTarget.value = row keyResults.value = [] @@ -1520,6 +1690,18 @@ onMounted(loadList) font-weight: 500; } +.recharge-alert, +.recharge-form, +.recharge-table { + margin-bottom: 1rem; +} + +.recharge-actions { + display: flex; + justify-content: flex-end; + gap: 0.5rem; +} + .snapshot-list { display: grid; gap: 0.7rem;