"""Finance service — daily revenue vs cost reconciliation. Architecture: - Website revenue: Sub2ApiWebsiteClient._request("/usage/stats", params=...) Reuses all existing error handling, logging, auth (api_key / bearer). - Upstream cost (Sub2API): UpstreamClient._send_request(url, params=...) within the client context — retains cookies, 401 refresh, external API log. - Upstream cost (New-API/Nox-API): same UpstreamClient pattern. 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, timezone 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 logger = logging.getLogger(__name__) SHANGHAI_TZ = pytz.timezone("Asia/Shanghai") # ───────────────────────────────────────────── # Upstream type classification # ───────────────────────────────────────────── def classify_upstream(upstream: Upstream) -> str: """Return 'sub2api' | 'new_api' | 'unknown'.""" prefix = (upstream.api_prefix or "").strip("/") if prefix == "api/v1": return "sub2api" auth = upstream.auth_type or "" if auth in ("new_api_token", "nox_token"): return "new_api" if auth in ("cookie", "login_password"): try: cfg = json.loads(upstream.auth_config_json or "{}") except Exception: cfg = {} if ( prefix == "" and ( auth == "cookie" or "/api/user/login" in (cfg.get("login_path") or "") or cfg.get("provider") in {"new-api", "nox-api"} ) ): return "new_api" return "unknown" # ───────────────────────────────────────────── # Date helpers # ───────────────────────────────────────────── def yesterday_shanghai() -> date: 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_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()) 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 # ───────────────────────────────────────────── def fetch_website_revenue(website: Website, target_date: date) -> tuple[float, str | None]: """Fetch website revenue for target_date. Uses Sub2ApiWebsiteClient (existing auth, error handling, external API log). GET /usage/stats?start_date=...&end_date=...&timezone=Asia/Shanghai&nocache=true """ date_str = target_date.isoformat() try: with Sub2ApiWebsiteClient( base_url=website.base_url, api_prefix=website.api_prefix, auth_type=website.auth_type, auth_config=json.loads(website.auth_config_json or "{}"), timeout=float(website.timeout_seconds), target_id=website.id, target_name=website.name, ) as c: data = c._request( "GET", "/usage/stats", params={ "start_date": date_str, "end_date": date_str, "timezone": "Asia/Shanghai", "nocache": "true", }, ) # Sub2API wraps payload in `data` key: {data: {total_actual_cost: ...}} payload = data.get("data", data) if isinstance(data, dict) else {} if isinstance(payload, dict): cost = payload.get("total_actual_cost") if cost is not None: return float(cost), None return 0.0, "响应中找不到 total_actual_cost 字段" except WebsiteError as e: return 0.0, str(e) except Exception as e: return 0.0, str(e) # ───────────────────────────────────────────── # Upstream cost — Sub2API # ───────────────────────────────────────────── def fetch_upstream_cost_sub2api(upstream: Upstream, target_date: date) -> tuple[float, str | None]: """Fetch cost for Sub2API-type upstream via /usage/stats (user-facing endpoint). Calls UpstreamClient._send_request inside the context so that cookies, 401 refresh, and external API logging are all preserved. Sub2API's end_date is exclusive (+1 day internally), so passing the same date for start and end yields the full day's data. """ date_str = target_date.isoformat() auth_config = json.loads(upstream.auth_config_json or "{}") try: with UpstreamClient( base_url=upstream.base_url, api_prefix=upstream.api_prefix, auth_type=upstream.auth_type, auth_config=auth_config, timeout=float(upstream.timeout_seconds), on_auth_config_update=lambda _: None, # read-only, no DB write target_id=upstream.id, target_name=upstream.name, ) as client: client.ensure_authenticated() params = { "start_date": date_str, "end_date": date_str, "timezone": "Asia/Shanghai", } resp = client._send_request("GET", client._url("/usage/stats"), params=params) resp.raise_for_status() data = resp.json() payload = data.get("data", data) if isinstance(data, dict) else {} if isinstance(payload, dict): cost = payload.get("total_actual_cost") if cost is not None: return float(cost), None return 0.0, "响应中找不到 total_actual_cost 字段" except Exception as e: return 0.0, str(e) # ───────────────────────────────────────────── # Upstream cost — New-API / Nox-API # ───────────────────────────────────────────── def _new_api_failure_message(data: Any) -> str | None: if not isinstance(data, dict): return None message = data.get("message") or data.get("detail") or data.get("error") if data.get("success") is False: return str(message or "上游返回 success=false") if message and data.get("error"): return str(message) return None def fetch_upstream_cost_new_api(upstream: Upstream, target_date: date) -> tuple[float, str | None]: """Fetch cost for New-API/Nox-API upstream via /api/log/self/stat. Keeps UpstreamClient open for the full request so that: - login_password token refresh is available if the stat call gets 401 - cookies are maintained throughout - external API call is logged via _send_request → _do_request Timestamp range: 00:00:00 to 23:59:59 Asia/Shanghai (closed interval per New-API source: `created_at <= endTimestamp`). """ start_ts, end_ts = date_to_shanghai_timestamps(target_date) auth_config = json.loads(upstream.auth_config_json or "{}") try: with UpstreamClient( base_url=upstream.base_url, api_prefix=upstream.api_prefix, auth_type=upstream.auth_type, auth_config=auth_config, timeout=float(upstream.timeout_seconds), on_auth_config_update=lambda _: None, # read-only, no DB write target_id=upstream.id, target_name=upstream.name, ) as client: client.ensure_authenticated() quota_per_unit = client._new_api_quota_per_unit() params = {"type": 2, "start_timestamp": start_ts, "end_timestamp": end_ts} resp = client._send_request("GET", client._url("/api/log/self/stat"), params=params) resp.raise_for_status() data = resp.json() failure_message = _new_api_failure_message(data) if failure_message: return 0.0, failure_message payload = data.get("data", data) if isinstance(data, dict) else {} if isinstance(payload, dict): quota = payload.get("quota") if quota is not None: cost = float(quota) / max(quota_per_unit, 1) return round(cost, 6), None return 0.0, "响应中找不到 quota 字段" except Exception as e: 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 # ───────────────────────────────────────────── 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. Never raises — each website/upstream failure is recorded in the item's status/error fields. Failures do NOT contribute to totals (safe default: under-count rather than over-count profit). """ website_items: list[dict[str, Any]] = [] upstream_items: list[dict[str, Any]] = [] total_revenue = 0.0 total_cost = 0.0 for w in websites: amount, err = fetch_website_revenue(w, target_date) item: dict[str, Any] = { "id": w.id, "name": w.name, "upstream_type": "sub2api", "amount": round(amount, 6), "status": "failed" if err else "success", "error": err, } website_items.append(item) if not err: total_revenue += amount for u in upstreams: 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: 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, "name": u.name, "upstream_type": utype, "amount": round(amount, 6), "status": "failed" if err else "success", "error": err, } upstream_items.append(item) if not err: total_cost += amount net = total_revenue - total_cost margin = (net / total_revenue * 100) if total_revenue > 0 else 0.0 failed_count = sum(1 for i in website_items + upstream_items if i["status"] == "failed") return { "date": target_date.isoformat(), "total_revenue": round(total_revenue, 6), "total_cost": round(total_cost, 6), "net_income": round(net, 6), "margin_percent": round(margin, 2), "website_items": website_items, "upstream_items": upstream_items, "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, db=db) 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