feat: persist finance daily summaries
This commit is contained in:
SmartUp Developer
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user