Compare commits
19 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 4fe117a4f6 | |||
| 0cf5b3816f | |||
| 9e082e6376 | |||
| b6be4b5d95 | |||
| 2289b2720c | |||
| afb5f478c9 | |||
| 615dab0490 | |||
| c386103809 | |||
| 66eaa40355 | |||
| db4647be5b | |||
| d8679b2ee5 | |||
| 211010cadd | |||
| 93cb283c1b | |||
| a43e082391 | |||
| e11ad9449c | |||
| 6ec4cfb4be | |||
| a52fc1973f | |||
| 58b59c264d | |||
| 0e149c83e7 |
@@ -0,0 +1,72 @@
|
||||
"""analytics_selfstat tables
|
||||
|
||||
Revision ID: 11c44afbea58
|
||||
Revises: admin_user_plain_password
|
||||
Create Date: 2026-07-08 16:32:49.351817
|
||||
|
||||
"""
|
||||
from collections.abc import Sequence
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from alembic import op
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = '11c44afbea58'
|
||||
down_revision: str | Sequence[str] | None = 'admin_user_plain_password'
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_table(
|
||||
'analytics_selfstat',
|
||||
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
|
||||
sa.Column('device_id', sa.String(length=64), nullable=False),
|
||||
sa.Column('epoch_id', sa.String(length=64), nullable=False),
|
||||
sa.Column('app_ver', sa.String(length=32), nullable=True),
|
||||
sa.Column('oem', sa.String(length=32), nullable=True),
|
||||
sa.Column('os', sa.String(length=32), nullable=True),
|
||||
sa.Column('batches_attempted', sa.BigInteger(), nullable=False, server_default='0'),
|
||||
sa.Column('batches_ok', sa.BigInteger(), nullable=False, server_default='0'),
|
||||
sa.Column('batches_fail', sa.BigInteger(), nullable=False, server_default='0'),
|
||||
sa.Column('retries', sa.BigInteger(), nullable=False, server_default='0'),
|
||||
sa.Column('queue_depth', sa.Integer(), nullable=False, server_default='0'),
|
||||
sa.Column('sent_at', sa.BigInteger(), nullable=True),
|
||||
sa.Column('created_at', sa.DateTime(timezone=True),
|
||||
server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
|
||||
sa.PrimaryKeyConstraint('id'),
|
||||
)
|
||||
with op.batch_alter_table('analytics_selfstat', schema=None) as batch_op:
|
||||
batch_op.create_index(batch_op.f('ix_analytics_selfstat_created_at'), ['created_at'], unique=False)
|
||||
batch_op.create_index(batch_op.f('ix_analytics_selfstat_device_id'), ['device_id'], unique=False)
|
||||
batch_op.create_index(batch_op.f('ix_analytics_selfstat_epoch_id'), ['epoch_id'], unique=False)
|
||||
|
||||
op.create_table(
|
||||
'analytics_selfstat_event',
|
||||
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
|
||||
sa.Column('snapshot_id', sa.Integer(), nullable=False),
|
||||
sa.Column('event', sa.String(length=64), nullable=False),
|
||||
sa.Column('attempted', sa.BigInteger(), nullable=False, server_default='0'),
|
||||
sa.Column('drop_capture', sa.BigInteger(), nullable=False, server_default='0'),
|
||||
sa.Column('delivered', sa.BigInteger(), nullable=False, server_default='0'),
|
||||
sa.Column('drop_undelivered', sa.BigInteger(), nullable=False, server_default='0'),
|
||||
sa.ForeignKeyConstraint(['snapshot_id'], ['analytics_selfstat.id'], ),
|
||||
sa.PrimaryKeyConstraint('id'),
|
||||
)
|
||||
with op.batch_alter_table('analytics_selfstat_event', schema=None) as batch_op:
|
||||
batch_op.create_index(batch_op.f('ix_analytics_selfstat_event_event'), ['event'], unique=False)
|
||||
batch_op.create_index(batch_op.f('ix_analytics_selfstat_event_snapshot_id'), ['snapshot_id'], unique=False)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
with op.batch_alter_table('analytics_selfstat_event', schema=None) as batch_op:
|
||||
batch_op.drop_index(batch_op.f('ix_analytics_selfstat_event_snapshot_id'))
|
||||
batch_op.drop_index(batch_op.f('ix_analytics_selfstat_event_event'))
|
||||
op.drop_table('analytics_selfstat_event')
|
||||
with op.batch_alter_table('analytics_selfstat', schema=None) as batch_op:
|
||||
batch_op.drop_index(batch_op.f('ix_analytics_selfstat_epoch_id'))
|
||||
batch_op.drop_index(batch_op.f('ix_analytics_selfstat_device_id'))
|
||||
batch_op.drop_index(batch_op.f('ix_analytics_selfstat_created_at'))
|
||||
op.drop_table('analytics_selfstat')
|
||||
@@ -0,0 +1,35 @@
|
||||
"""admin_user 加 pages_override 列(「自定义」权限:按人存专属可见页 key 列表)
|
||||
|
||||
权限管理页新增「自定义」角色:选它时该成员的可见页不跟随任何共享角色,而由逐页勾选决定,
|
||||
存这个人专属的一份页 key 列表。仅 role == "custom" 时有效;普通角色为 None(可见页跟随角色)。
|
||||
PG 用 JSONB,SQLite 退化为通用 JSON(同 admin_audit_log.detail / comparison_record.raw_payload)。
|
||||
|
||||
Revision ID: admin_user_pages_override
|
||||
Revises: admin_user_plain_password
|
||||
Create Date: 2026-07-08 00:00:00.000000
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy.dialects.postgresql import JSONB
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision: str = "admin_user_pages_override"
|
||||
down_revision: str | Sequence[str] | None = "admin_user_plain_password"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
# 与 app/models/admin.py 的 _JSON 一致:PG JSONB / 其它 JSON
|
||||
_JSON = sa.JSON().with_variant(JSONB(), "postgresql")
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
with op.batch_alter_table("admin_user", schema=None) as batch_op:
|
||||
batch_op.add_column(sa.Column("pages_override", _JSON, nullable=True))
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
with op.batch_alter_table("admin_user", schema=None) as batch_op:
|
||||
batch_op.drop_column("pages_override")
|
||||
@@ -26,6 +26,7 @@ from app.admin.routers.cps import router as cps_router
|
||||
from app.admin.routers.dashboard import router as dashboard_router
|
||||
from app.admin.routers.device_liveness import router as device_liveness_router
|
||||
from app.admin.routers.ops_stat_config import router as ops_stat_config_router
|
||||
from app.admin.routers.analytics_health import router as analytics_health_router
|
||||
from app.admin.routers.event_logs import router as event_logs_router
|
||||
from app.admin.routers.feedback import router as feedback_router
|
||||
from app.admin.routers.feedback_qr import router as feedback_qr_router
|
||||
@@ -97,6 +98,7 @@ admin_app.include_router(withdraw_router)
|
||||
admin_app.include_router(price_report_router)
|
||||
admin_app.include_router(feedback_router)
|
||||
admin_app.include_router(event_logs_router)
|
||||
admin_app.include_router(analytics_health_router)
|
||||
admin_app.include_router(feedback_qr_router)
|
||||
admin_app.include_router(admins_router)
|
||||
admin_app.include_router(roles_router)
|
||||
|
||||
@@ -9,6 +9,9 @@ super_admin 为内建全权角色,恒可见全部页(effective_pages 特判)。
|
||||
from __future__ import annotations
|
||||
|
||||
SUPER_ADMIN_ROLE = "super_admin"
|
||||
# 「自定义」哨兵角色:不是 admin_role 表里的行,而是标记「这个人的可见页由 pages_override 决定」。
|
||||
# admin_user.role == CUSTOM_ROLE 时,有效可见页取 admin_user.pages_override(见 auth._admin_out_with_pages)。
|
||||
CUSTOM_ROLE = "custom"
|
||||
|
||||
# 分组镜像前端导航(app/(main)/layout.tsx 的 NAV_GROUPS);key = 路由一级
|
||||
PERMISSION_CATALOG: list[dict] = [
|
||||
|
||||
@@ -385,6 +385,27 @@ def ad_revenue_report(
|
||||
for k, v in type_map.items()
|
||||
}
|
||||
|
||||
# 分场景小计(按 feed_scene:展示条数 + 预估收益),同 type_stats 基于全量 events——
|
||||
# 供数据大盘「领券广告 / 比价广告」卡用。此前大盘是在分页 items 里按 feed_scene 现算,
|
||||
# 2026-07-02 起信息流逐条展示行(唯一带收益 + 场景的行)不再进主表 items,现算恒为 0;
|
||||
# 改为服务端在全量上聚合下发(也顺带不受 limit 分页截断影响)。feed_scene 为空(激励视频 /
|
||||
# 旧数据)不计入任何场景桶。
|
||||
scene_map: dict[str, dict] = {}
|
||||
for e in events:
|
||||
sc = e.get("feed_scene")
|
||||
if not sc:
|
||||
continue
|
||||
s = scene_map.get(sc)
|
||||
if s is None:
|
||||
s = {"impressions": 0, "revenue_yuan": 0.0}
|
||||
scene_map[sc] = s
|
||||
s["impressions"] += e["impressions"]
|
||||
s["revenue_yuan"] += e["revenue_yuan"]
|
||||
scene_stats = {
|
||||
k: {"impressions": v["impressions"], "revenue_yuan": round(v["revenue_yuan"], 6)}
|
||||
for k, v in scene_map.items()
|
||||
}
|
||||
|
||||
# DAU:复用数据大盘活跃用户口径(登录 + 开始比价 + 开始领券,按用户去重),按所选日期区间
|
||||
# 统计(含今日),历史 / 多天区间同样有值。ARPU = 区间预估收益 ÷ 区间活跃用户。全局口径,
|
||||
# 不随 user / ad_type / feed_scene / app_env 筛选变化(活跃用户口径无这些维度)。
|
||||
@@ -418,6 +439,7 @@ def ad_revenue_report(
|
||||
"daily": daily,
|
||||
"hourly": hourly,
|
||||
"type_stats": type_stats,
|
||||
"scene_stats": scene_stats,
|
||||
"dau": dau,
|
||||
"items": main_rows[offset:offset + limit],
|
||||
}
|
||||
|
||||
@@ -26,14 +26,17 @@ def create_admin(
|
||||
password: str,
|
||||
role: str = "operator",
|
||||
plain_password: str | None = None,
|
||||
pages_override: list[str] | None = None,
|
||||
) -> AdminUser:
|
||||
"""建管理员。plain_password 非空则额外留存明文(后台 UI 建的账号传,供权限管理页复看);
|
||||
脚本/起后台建账号不传(留 None → 前端不显示密码)。"""
|
||||
脚本/起后台建账号不传(留 None → 前端不显示密码)。
|
||||
pages_override:role=="custom" 时传专属可见页 key 列表;普通角色为 None。"""
|
||||
admin = AdminUser(
|
||||
username=username,
|
||||
password_hash=hash_password(password),
|
||||
role=role,
|
||||
plain_password=plain_password,
|
||||
pages_override=pages_override,
|
||||
)
|
||||
db.add(admin)
|
||||
db.commit()
|
||||
|
||||
@@ -0,0 +1,146 @@
|
||||
"""埋点健康度聚合(埋点成功率 / 上报成功率)。
|
||||
|
||||
只存原始累计快照,查询时在 Python 侧差分聚合(admin 低频、量级小,跨 PG/SQLite 无方言坑;
|
||||
与 cps.py / coupon_data.py 同款约定)。差分按 (device_id, epoch_id, event) 分区、created_at
|
||||
升序,相邻做差、负值夹 0;每增量按其快照 created_at 归入北京天桶。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core import rewards
|
||||
from app.models.analytics_selfstat import AnalyticsSelfStat as H
|
||||
from app.models.analytics_selfstat import AnalyticsSelfStatEvent as E
|
||||
|
||||
_COUNTS = ("attempted", "drop_capture", "delivered", "drop_undelivered")
|
||||
|
||||
|
||||
def diff_snapshots(rows: list[dict]) -> list[dict]:
|
||||
"""累计快照行 → 每快照增量行(纯逻辑)。
|
||||
|
||||
rows 每行含 device_id/epoch_id/event/created_at/app_ver/oem/os + 四个累计计数。
|
||||
返回每行含 dims + created_at + 四个增量 d_*(分区首行增量=累计值;负值夹 0)。
|
||||
"""
|
||||
parts: dict[tuple, list[dict]] = defaultdict(list)
|
||||
for r in rows:
|
||||
parts[(r["device_id"], r["epoch_id"], r["event"])].append(r)
|
||||
|
||||
out: list[dict] = []
|
||||
for group in parts.values():
|
||||
group.sort(key=lambda r: (r["created_at"], r.get("id", 0)))
|
||||
prev = {k: 0 for k in _COUNTS}
|
||||
for r in group:
|
||||
deltas = {f"d_{k}": max(0, int(r[k]) - prev[k]) for k in _COUNTS}
|
||||
out.append({
|
||||
"device_id": r["device_id"], "epoch_id": r["epoch_id"], "event": r["event"],
|
||||
"created_at": r["created_at"], "app_ver": r["app_ver"],
|
||||
"oem": r["oem"], "os": r["os"], **deltas,
|
||||
})
|
||||
prev = {k: int(r[k]) for k in _COUNTS}
|
||||
return out
|
||||
|
||||
|
||||
def _cn_day(dt: datetime) -> str:
|
||||
"""created_at(UTC 口径)→ 北京日期字符串 YYYY-MM-DD。naive 当 UTC,tz-aware 直接换算。"""
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=UTC)
|
||||
return dt.astimezone(rewards.CN_TZ).date().isoformat()
|
||||
|
||||
|
||||
def _rates(sums: dict) -> dict:
|
||||
"""由四个增量和派生两段率(分母 0 → None)。"""
|
||||
persisted_denom = sums["attempted"]
|
||||
report_denom = sums["delivered"] + sums["drop_undelivered"]
|
||||
return {
|
||||
**sums,
|
||||
"track_success_rate": (
|
||||
(sums["attempted"] - sums["drop_capture"]) / persisted_denom
|
||||
if persisted_denom else None
|
||||
),
|
||||
"report_success_rate": (
|
||||
sums["delivered"] / report_denom if report_denom else None
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def _sum_deltas(deltas: list[dict]) -> dict:
|
||||
return {k: sum(d[f"d_{k}"] for d in deltas) for k in _COUNTS}
|
||||
|
||||
|
||||
def _fetch_rows(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
|
||||
"""取 [from, to) 区间行 + 每分区在 from 左侧的最后一条基线行(供第一条区间增量做差)。"""
|
||||
cols = (
|
||||
H.id, H.device_id, H.epoch_id, E.event, H.created_at,
|
||||
H.app_ver, H.oem, H.os,
|
||||
E.attempted, E.drop_capture, E.delivered, E.drop_undelivered,
|
||||
)
|
||||
in_range = db.execute(
|
||||
select(*cols).join(E, E.snapshot_id == H.id)
|
||||
.where(H.created_at >= date_from, H.created_at < date_to)
|
||||
).mappings().all()
|
||||
|
||||
# 注:基线子查询无下界扫 from 左侧全量(spec §7 已接受的取舍;量级变大再上物化 rollup)。
|
||||
# 用 max(id) 而非 max(created_at) 选"最新一条":id 严格单调,避免 SQLite 秒级时间戳撞车时选歧义。
|
||||
sub = (
|
||||
select(H.device_id, H.epoch_id, E.event, func.max(H.id).label("max_id"))
|
||||
.join(E, E.snapshot_id == H.id)
|
||||
.where(H.created_at < date_from)
|
||||
.group_by(H.device_id, H.epoch_id, E.event)
|
||||
.subquery()
|
||||
)
|
||||
baseline = db.execute(
|
||||
select(*cols).join(E, E.snapshot_id == H.id).join(
|
||||
sub, sub.c.max_id == H.id
|
||||
)
|
||||
).mappings().all()
|
||||
|
||||
return [dict(r) for r in list(baseline) + list(in_range)]
|
||||
|
||||
|
||||
def _in_range_deltas(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
|
||||
"""差分后只保留 created_at ∈ [from, to) 的增量(基线行被差分用后丢弃)。
|
||||
|
||||
Python 侧过滤需对齐 tz 口径:SQLite 返回 naive UTC,PG 返回 aware UTC。
|
||||
统一转成 naive UTC 再比较,兼容两种后端。
|
||||
"""
|
||||
def _to_naive_utc(dt: datetime) -> datetime:
|
||||
if dt.tzinfo is not None:
|
||||
return dt.astimezone(UTC).replace(tzinfo=None)
|
||||
return dt
|
||||
|
||||
from_naive = _to_naive_utc(date_from)
|
||||
to_naive = _to_naive_utc(date_to)
|
||||
deltas = diff_snapshots(_fetch_rows(db, date_from, date_to))
|
||||
return [d for d in deltas if from_naive <= _to_naive_utc(d["created_at"]) < to_naive]
|
||||
|
||||
|
||||
def overview(db: Session, date_from: datetime, date_to: datetime) -> dict:
|
||||
deltas = _in_range_deltas(db, date_from, date_to)
|
||||
return _rates(_sum_deltas(deltas))
|
||||
|
||||
|
||||
def trend(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
|
||||
deltas = _in_range_deltas(db, date_from, date_to)
|
||||
by_day: dict[str, list[dict]] = defaultdict(list)
|
||||
for d in deltas:
|
||||
by_day[_cn_day(d["created_at"])].append(d)
|
||||
return [
|
||||
{"day": day, **_rates(_sum_deltas(items))}
|
||||
for day, items in sorted(by_day.items())
|
||||
]
|
||||
|
||||
|
||||
def breakdown(db: Session, date_from: datetime, date_to: datetime, dim: str) -> list[dict]:
|
||||
if dim not in ("event", "app_ver", "oem"):
|
||||
raise ValueError(f"invalid dim: {dim!r}")
|
||||
deltas = _in_range_deltas(db, date_from, date_to)
|
||||
by_key: dict[str, list[dict]] = defaultdict(list)
|
||||
for d in deltas:
|
||||
by_key[d.get(dim) or "(unknown)"].append(d)
|
||||
rows = [{"key": key, **_rates(_sum_deltas(items))} for key, items in by_key.items()]
|
||||
rows.sort(key=lambda r: (r["report_success_rate"] is None, r["report_success_rate"] or 0.0))
|
||||
return rows
|
||||
@@ -51,7 +51,27 @@ def _yuan_to_cents(v: object) -> int | None:
|
||||
|
||||
def _ts_to_dt(ts: object) -> datetime | None:
|
||||
"""秒级时间戳 → tz-aware UTC datetime(绝对时刻,前端按北京展示)。"""
|
||||
if not ts:
|
||||
if ts is None:
|
||||
return None
|
||||
if isinstance(ts, datetime):
|
||||
return ts if ts.tzinfo else ts.replace(tzinfo=_BJ_TZ).astimezone(timezone.utc)
|
||||
s = str(ts).strip()
|
||||
if not s or s.lower() == "null":
|
||||
return None
|
||||
try:
|
||||
seconds = float(Decimal(s))
|
||||
except (InvalidOperation, ValueError):
|
||||
return None
|
||||
if seconds == 0:
|
||||
return None
|
||||
# 美团文档是秒级时间戳,这里顺手兼容毫秒/微秒,避免上游格式变化导致时间再次落空。
|
||||
if abs(seconds) > 10_000_000_000_000:
|
||||
seconds /= 1_000_000
|
||||
elif abs(seconds) > 10_000_000_000:
|
||||
seconds /= 1_000
|
||||
try:
|
||||
return datetime.fromtimestamp(seconds, tz=timezone.utc)
|
||||
except (OverflowError, OSError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
@@ -83,10 +103,6 @@ def _pick(row: dict[str, Any], *keys: str) -> Any:
|
||||
if key in row and row[key] is not None:
|
||||
return row[key]
|
||||
return None
|
||||
try:
|
||||
return datetime.fromtimestamp(int(ts), tz=timezone.utc)
|
||||
except (ValueError, OSError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
# ───────────── 群 ─────────────
|
||||
|
||||
@@ -11,6 +11,7 @@ from zoneinfo import ZoneInfo
|
||||
from sqlalchemy import Select, asc, case, desc, func, or_, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.admin.repositories.stats import COMPARE_START_EVENT, COUPON_START_EVENT
|
||||
from app.core import rewards
|
||||
from app.core.config import settings
|
||||
from app.models.ad_feed_reward import AdFeedRewardRecord
|
||||
@@ -18,6 +19,7 @@ from app.models.ad_reward import AdRewardRecord
|
||||
from app.models.admin import AdminAuditLog
|
||||
from app.models.analytics_event import AnalyticsEvent
|
||||
from app.models.comparison import ComparisonRecord
|
||||
from app.models.coupon_state import CouponPromptEngagement
|
||||
from app.models.device import DeviceLiveness
|
||||
from app.models.feedback import Feedback
|
||||
from app.models.onboarding import OnboardingCompletion
|
||||
@@ -31,6 +33,9 @@ from app.models.wallet import (
|
||||
WithdrawOrder,
|
||||
)
|
||||
|
||||
# 「最近活跃」计入的行为事件(与大盘 DAU/留存活跃口径一致:开始比价 + 开始领券)
|
||||
_ACTIVE_EVENTS = (COMPARE_START_EVENT, COUPON_START_EVENT)
|
||||
|
||||
# 折算成可提现现金时,非广告金币来源的排除集(广告单独统计、人工调整不算"赚取")
|
||||
_NON_TASK_BIZ_TYPES = ("reward_video", "feed_ad_reward", "admin_grant", "admin_deduct")
|
||||
|
||||
@@ -82,6 +87,86 @@ def offset_paginate(
|
||||
return items, next_cursor, total
|
||||
|
||||
|
||||
def _last_active_parts():
|
||||
"""「最近活跃」的两个按 user_id 预聚合派生表(最近开始比价/领券事件、最近领券发起)。
|
||||
|
||||
活跃口径与大盘 DAU/留存一致(2026-07-05 产品定:进入 App≈登录 last_login_at +
|
||||
发起比价 real_compare_start + 发起领券 real_coupon_start/claim_started)。
|
||||
用 LEFT JOIN 预聚合而非相关标量子查询:后者在 PG 上对 users 每行各跑一个 SubPlan
|
||||
(排序键、range 筛选、offset_paginate 的 count 三处叠加),埋点表大了会拖垮列表接口;
|
||||
预聚合借 analytics_event.event 索引只扫两类 start 事件,每次查询聚合一次。
|
||||
"""
|
||||
ev_agg = (
|
||||
select(
|
||||
AnalyticsEvent.user_id.label("user_id"),
|
||||
func.max(AnalyticsEvent.created_at).label("last_at"),
|
||||
)
|
||||
.where(
|
||||
AnalyticsEvent.user_id.is_not(None),
|
||||
AnalyticsEvent.event.in_(_ACTIVE_EVENTS),
|
||||
)
|
||||
.group_by(AnalyticsEvent.user_id)
|
||||
.subquery()
|
||||
)
|
||||
eng_agg = (
|
||||
select(
|
||||
CouponPromptEngagement.user_id.label("user_id"),
|
||||
func.max(CouponPromptEngagement.created_at).label("last_at"),
|
||||
)
|
||||
.where(
|
||||
CouponPromptEngagement.user_id.is_not(None),
|
||||
CouponPromptEngagement.engage_type == "claim_started",
|
||||
)
|
||||
.group_by(CouponPromptEngagement.user_id)
|
||||
.subquery()
|
||||
)
|
||||
return ev_agg, eng_agg
|
||||
|
||||
|
||||
def _norm_utc(dt: datetime | None) -> datetime | None:
|
||||
"""naive 视为 UTC 补 tzinfo(SQLite 读回 naive、PG 读回 aware,混着 max() 会 TypeError)。"""
|
||||
if dt is None:
|
||||
return None
|
||||
return dt if dt.tzinfo is not None else dt.replace(tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _attach_last_active(db: Session, users: list[User]) -> None:
|
||||
"""给本页用户瞬态挂 last_active_at(非 DB 列,供 AdminUserListItem from_attributes 读)。
|
||||
|
||||
口径同 [_last_active_expr];按本页 user_id 批量两次 GROUP BY,防 N+1。
|
||||
"""
|
||||
uids = [u.id for u in users]
|
||||
if not uids:
|
||||
return
|
||||
ev_map = dict(
|
||||
db.execute(
|
||||
select(AnalyticsEvent.user_id, func.max(AnalyticsEvent.created_at))
|
||||
.where(
|
||||
AnalyticsEvent.user_id.in_(uids),
|
||||
AnalyticsEvent.event.in_(_ACTIVE_EVENTS),
|
||||
)
|
||||
.group_by(AnalyticsEvent.user_id)
|
||||
).all()
|
||||
)
|
||||
eng_map = dict(
|
||||
db.execute(
|
||||
select(CouponPromptEngagement.user_id, func.max(CouponPromptEngagement.created_at))
|
||||
.where(
|
||||
CouponPromptEngagement.user_id.in_(uids),
|
||||
CouponPromptEngagement.engage_type == "claim_started",
|
||||
)
|
||||
.group_by(CouponPromptEngagement.user_id)
|
||||
).all()
|
||||
)
|
||||
for u in users:
|
||||
candidates = [
|
||||
_norm_utc(u.last_login_at),
|
||||
_norm_utc(ev_map.get(u.id)),
|
||||
_norm_utc(eng_map.get(u.id)),
|
||||
]
|
||||
u.last_active_at = max((c for c in candidates if c is not None), default=None)
|
||||
|
||||
|
||||
def list_users(
|
||||
db: Session,
|
||||
*,
|
||||
@@ -93,16 +178,34 @@ def list_users(
|
||||
created_to: datetime | None = None,
|
||||
last_login_from: datetime | None = None,
|
||||
last_login_to: datetime | None = None,
|
||||
last_active_from: datetime | None = None,
|
||||
last_active_to: datetime | None = None,
|
||||
sort_by: str = "id",
|
||||
sort_order: str = "desc",
|
||||
limit: int = 20,
|
||||
cursor: int | None = None,
|
||||
) -> tuple[list[User], int | None, int]:
|
||||
"""用户列表(admin 全量)。支持手机号前缀 / 渠道 / 状态 / 昵称模糊 / 注册·最近登录时间范围筛选,
|
||||
按 id·注册时间·最近登录排序。**offset 分页**(cursor=offset):任意列排序下游标语义统一,
|
||||
"""用户列表(admin 全量)。支持手机号前缀 / 渠道 / 状态 / 昵称模糊 / 注册·最近登录·最近活跃
|
||||
时间范围筛选,按 id·注册时间·最近登录·最近活跃排序;每页附带计算列 last_active_at
|
||||
(口径见 [_last_active_expr])。**offset 分页**(cursor=offset):任意列排序下游标语义统一,
|
||||
代价是翻页期间数据变动可能错位一条——admin 低频场景可接受(同 [list_all_withdraw_orders])。
|
||||
日期入参统一转 tz-aware UTC 比较(列为 timestamptz,见 _as_utc)。"""
|
||||
stmt = select(User)
|
||||
# 最近活跃 = max(最近登录, 最近行为事件, 最近领券发起)。PG 用 GREATEST;SQLite 标量 max()
|
||||
# 任一参数 NULL 即返回 NULL,故 LEFT JOIN 未命中侧 coalesce 到 last_login_at 兜底
|
||||
# (注册即登录,该列恒非空)。派生表 1:1(按 user_id 聚合),outerjoin 不会放大行数,
|
||||
# offset_paginate 的 count 不受影响。
|
||||
ev_agg, eng_agg = _last_active_parts()
|
||||
greatest = func.greatest if db.get_bind().dialect.name == "postgresql" else func.max
|
||||
last_active = greatest(
|
||||
User.last_login_at,
|
||||
func.coalesce(ev_agg.c.last_at, User.last_login_at),
|
||||
func.coalesce(eng_agg.c.last_at, User.last_login_at),
|
||||
)
|
||||
stmt = (
|
||||
select(User)
|
||||
.outerjoin(ev_agg, ev_agg.c.user_id == User.id)
|
||||
.outerjoin(eng_agg, eng_agg.c.user_id == User.id)
|
||||
)
|
||||
if phone:
|
||||
stmt = stmt.where(User.phone.like(f"{phone}%")) # 前缀匹配
|
||||
if register_channel:
|
||||
@@ -119,16 +222,25 @@ def list_users(
|
||||
stmt = stmt.where(User.last_login_at >= _as_utc(last_login_from))
|
||||
if last_login_to is not None:
|
||||
stmt = stmt.where(User.last_login_at <= _as_utc(last_login_to))
|
||||
if last_active_from is not None:
|
||||
stmt = stmt.where(last_active >= _as_utc(last_active_from))
|
||||
if last_active_to is not None:
|
||||
stmt = stmt.where(last_active <= _as_utc(last_active_to))
|
||||
|
||||
sort_cols = {
|
||||
"id": User.id,
|
||||
"created_at": User.created_at,
|
||||
"last_login_at": User.last_login_at,
|
||||
"last_active_at": last_active,
|
||||
}
|
||||
sort_col = sort_cols.get(sort_by, User.id)
|
||||
order_fn = asc if sort_order == "asc" else desc
|
||||
id_order = asc(User.id) if sort_order == "asc" else desc(User.id)
|
||||
return offset_paginate(db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor)
|
||||
items, next_cursor, total = offset_paginate(
|
||||
db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor
|
||||
)
|
||||
_attach_last_active(db, items)
|
||||
return items, next_cursor, total
|
||||
|
||||
|
||||
def _attach_user_info(db: Session, records: list[ComparisonRecord | Feedback | PriceReport]) -> None:
|
||||
|
||||
+167
-23
@@ -5,17 +5,23 @@ user.last_login_at / comparison_record.status / withdraw_order.status)要加索
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import Counter
|
||||
from datetime import date, datetime, time, timedelta, timezone
|
||||
from decimal import Decimal, InvalidOperation
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy import case, func, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.admin.repositories.coupon_data import _percentile
|
||||
from app.models.ad_feed_reward import AdFeedRewardRecord
|
||||
from app.models.ad_reward import AdRewardRecord
|
||||
from app.models.analytics_event import AnalyticsEvent
|
||||
from app.models.comparison import ComparisonRecord
|
||||
from app.models.coupon_state import CouponPromptEngagement
|
||||
from app.models.coupon_state import (
|
||||
CouponClaimRecord,
|
||||
CouponPromptEngagement,
|
||||
CouponSession,
|
||||
)
|
||||
from app.models.cps_order import CpsOrder
|
||||
from app.models.feedback import Feedback
|
||||
from app.models.savings import SavingsRecord
|
||||
@@ -24,11 +30,17 @@ from app.models.user import User
|
||||
from app.models.wallet import CoinTransaction, WithdrawOrder
|
||||
|
||||
_BEIJING = timezone(timedelta(hours=8))
|
||||
COUPON_REWARD_BIZ_TYPES = ("reward_video", "ad_reward", "coupon", "coupon_reward")
|
||||
REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward")
|
||||
# 领券/比价奖励金币的真实来源是信息流广告发奖(ad_feed_reward_record,按 feed_scene 分场景);
|
||||
# coin_transaction 里只有扁平的 feed_ad_reward、biz_type 不分 coupon/comparison,故这俩桶历史从未
|
||||
# 被写入,仅留作未来兜底,实际金额在下方按 feed_scene 汇总 ad_feed_reward_record 得出。reward_video/
|
||||
# ad_reward 是激励视频,单独成桶、不再混进领券奖励(历史误并会把激励视频金币双计进领券)。
|
||||
COUPON_REWARD_BIZ_TYPES = ("coupon", "coupon_reward")
|
||||
COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward")
|
||||
EXCLUDED_REWARD_BIZ_TYPES = ("invite_inviter", "invite_invitee", "admin_grant")
|
||||
UNCLASSIFIED_FEED_BIZ_TYPES = ("feed_ad_reward",)
|
||||
REGULAR_TASK_EXCLUDED_BIZ_TYPES = (
|
||||
*REWARD_VIDEO_BIZ_TYPES,
|
||||
*COUPON_REWARD_BIZ_TYPES,
|
||||
*COMPARISON_REWARD_BIZ_TYPES,
|
||||
*EXCLUDED_REWARD_BIZ_TYPES,
|
||||
@@ -299,12 +311,12 @@ def dashboard_overview(
|
||||
period_from=period_from,
|
||||
period_to=period_to,
|
||||
)
|
||||
period_retained_new_user_ids = period_new_user_ids & period_active_user_ids
|
||||
period_retention_rate = (
|
||||
round(len(period_retained_new_user_ids) / len(period_new_user_ids), 4)
|
||||
if period_new_user_ids
|
||||
else None
|
||||
)
|
||||
# 留存口径(2026-07-05 产品改):次日留存——窗口内每天 D,取 **D-1 日(前日)新增**用户,
|
||||
# 统计其 D 日活跃(登录/开始比价/开始领券)比例,逐日累加。默认窗口=昨日单天,即
|
||||
# 「前日新增用户的昨日留存」。原口径(窗口内新增∩窗口内活跃)在单日窗口下≈100% 无意义
|
||||
# (注册即登录,当天新增必然当天活跃)。逐日 cohort 在下方 trend 循环内顺带累计。
|
||||
retention_cohort_total = 0
|
||||
retention_retained_total = 0
|
||||
trend_points: list[dict] = []
|
||||
for cur_date in _date_range(period_from, period_to):
|
||||
day_start_utc, day_end_utc, day_start_local, day_end_local = _period_bounds(
|
||||
@@ -314,18 +326,26 @@ def dashboard_overview(
|
||||
ComparisonRecord.created_at >= day_start_local,
|
||||
ComparisonRecord.created_at < day_end_local,
|
||||
)
|
||||
daily_active_user_ids = _period_active_user_ids(
|
||||
db,
|
||||
start_utc=day_start_utc,
|
||||
end_utc=day_end_utc,
|
||||
period_from=cur_date,
|
||||
period_to=cur_date,
|
||||
)
|
||||
# 次日留存:cohort = 前一日(D-1)新增用户,留存 = 其中当日(D)活跃者(口径见上)。
|
||||
cohort_ids = _user_id_set(
|
||||
select(User.id).where(
|
||||
User.created_at >= day_start_utc - timedelta(days=1),
|
||||
User.created_at < day_end_utc - timedelta(days=1),
|
||||
)
|
||||
)
|
||||
retention_cohort_total += len(cohort_ids)
|
||||
retention_retained_total += len(cohort_ids & daily_active_user_ids)
|
||||
trend_points.append(
|
||||
{
|
||||
"date": cur_date,
|
||||
"active_users": len(
|
||||
_period_active_user_ids(
|
||||
db,
|
||||
start_utc=day_start_utc,
|
||||
end_utc=day_end_utc,
|
||||
period_from=cur_date,
|
||||
period_to=cur_date,
|
||||
)
|
||||
),
|
||||
"active_users": len(daily_active_user_ids),
|
||||
"new_users": _count(
|
||||
User,
|
||||
User.created_at >= day_start_utc,
|
||||
@@ -334,6 +354,11 @@ def dashboard_overview(
|
||||
"comparisons": _count(ComparisonRecord, *daily_comparison_conds),
|
||||
}
|
||||
)
|
||||
period_retention_rate = (
|
||||
round(retention_retained_total / retention_cohort_total, 4)
|
||||
if retention_cohort_total
|
||||
else None
|
||||
)
|
||||
|
||||
period_coin_conds = (
|
||||
CoinTransaction.created_at >= start_local,
|
||||
@@ -343,7 +368,7 @@ def dashboard_overview(
|
||||
period_reward_video_coin_total = _sum(
|
||||
CoinTransaction.amount,
|
||||
*period_coin_conds,
|
||||
CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
|
||||
CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES),
|
||||
)
|
||||
period_feed_ad_coin_total = _sum(
|
||||
CoinTransaction.amount,
|
||||
@@ -365,15 +390,29 @@ def dashboard_overview(
|
||||
*period_coin_conds,
|
||||
CoinTransaction.biz_type.like("task_%"),
|
||||
)
|
||||
# 领券/比价奖励金币 = biz_type 桶(历史空,兜底)+ 该场景信息流广告实发金币
|
||||
# (ad_feed_reward_record.feed_scene,granted;reward_date 是北京日期串,与 period 同自然日窗口)。
|
||||
period_coupon_reward_coin_total = _sum(
|
||||
CoinTransaction.amount,
|
||||
*period_coin_conds,
|
||||
CoinTransaction.biz_type.in_(COUPON_REWARD_BIZ_TYPES),
|
||||
) + _sum(
|
||||
AdFeedRewardRecord.coin,
|
||||
AdFeedRewardRecord.status == "granted",
|
||||
AdFeedRewardRecord.feed_scene == "coupon",
|
||||
AdFeedRewardRecord.reward_date >= period_from.isoformat(),
|
||||
AdFeedRewardRecord.reward_date <= period_to.isoformat(),
|
||||
)
|
||||
period_comparison_reward_coin_total = _sum(
|
||||
CoinTransaction.amount,
|
||||
*period_coin_conds,
|
||||
CoinTransaction.biz_type.in_(COMPARISON_REWARD_BIZ_TYPES),
|
||||
) + _sum(
|
||||
AdFeedRewardRecord.coin,
|
||||
AdFeedRewardRecord.status == "granted",
|
||||
AdFeedRewardRecord.feed_scene == "comparison",
|
||||
AdFeedRewardRecord.reward_date >= period_from.isoformat(),
|
||||
AdFeedRewardRecord.reward_date <= period_to.isoformat(),
|
||||
)
|
||||
period_regular_task_coin_total = _sum(
|
||||
CoinTransaction.amount,
|
||||
@@ -417,6 +456,98 @@ def dashboard_overview(
|
||||
else None
|
||||
)
|
||||
|
||||
# ===== 领券核心数据(2026-07-05 产品新增)=====
|
||||
# 数据源:coupon_session(一次领券一行,started_date 北京自然日)+ coupon_claim_record
|
||||
# (一券/点位一天一条终态,claim_date 北京自然日)。点位与 session 不按 trace_id 关联——
|
||||
# record_claims 更新路径不覆盖 trace_id(同设备同券同日重跑归第一次的 trace),按
|
||||
# (device_id, 自然日) 桶关联才可靠;同桶多次发起共享同一份点位终态。
|
||||
period_coupon_sessions = db.execute(
|
||||
select(
|
||||
CouponSession.device_id,
|
||||
CouponSession.started_date,
|
||||
CouponSession.status,
|
||||
CouponSession.elapsed_ms,
|
||||
).where(
|
||||
CouponSession.started_date >= period_from,
|
||||
CouponSession.started_date <= period_to,
|
||||
# 只统计正式环境,同「领券数据」页默认口径(防 debug 包调试数据串台;
|
||||
# 命中 ix_coupon_session_date_env)。点位表无 app_env 列,但点位指标只经
|
||||
# 下方 prod session 触达的 (device, 日) 桶进入统计,随之收敛到 prod。
|
||||
CouponSession.app_env == "prod",
|
||||
)
|
||||
).all()
|
||||
coupon_started = len(period_coupon_sessions)
|
||||
coupon_completed_elapsed = sorted(
|
||||
s.elapsed_ms
|
||||
for s in period_coupon_sessions
|
||||
if s.status == "completed" and s.elapsed_ms is not None
|
||||
)
|
||||
# 点位桶:(device, 日) → (点位总数, 成功点位数)。成功口径与「我的」页累计领券一致
|
||||
# (sum_claimed_count,2026-06-15 产品定):success + already_claimed(已领过=持有券)都算成功。
|
||||
point_buckets: dict[tuple[str, date], tuple[int, int]] = {
|
||||
(dev, d): (int(total), int(succ or 0))
|
||||
for dev, d, total, succ in db.execute(
|
||||
select(
|
||||
CouponClaimRecord.device_id,
|
||||
CouponClaimRecord.claim_date,
|
||||
func.count(),
|
||||
func.sum(
|
||||
case(
|
||||
(CouponClaimRecord.status.in_(("success", "already_claimed")), 1),
|
||||
else_=0,
|
||||
)
|
||||
),
|
||||
)
|
||||
.where(
|
||||
CouponClaimRecord.claim_date >= period_from,
|
||||
# 上界放宽一天:跨零点场次(23:5x 发起)的点位 claim_date 落在发起日+1,
|
||||
# 桶只经下方 session 触达的键参与计数,放宽不会引入无关数据。
|
||||
CouponClaimRecord.claim_date <= period_to + timedelta(days=1),
|
||||
)
|
||||
.group_by(CouponClaimRecord.device_id, CouponClaimRecord.claim_date)
|
||||
).all()
|
||||
}
|
||||
# 全部领成功的次数:completed 且其 (device, 日) 桶内点位全部成功(桶为空不算)。
|
||||
coupon_all_success = 0
|
||||
completed_bucket_totals: list[int] = []
|
||||
session_bucket_keys: set[tuple[str, date]] = set()
|
||||
for s in period_coupon_sessions:
|
||||
key = (s.device_id, s.started_date)
|
||||
if key not in point_buckets:
|
||||
# 跨零点回退:发起日桶不存在(点位终态全部落在次日)时取 (device, 发起日+1)。
|
||||
# 仅在发起日桶完全缺失时回退,避免抢占该设备次日 session 自己的桶。
|
||||
next_key = (s.device_id, s.started_date + timedelta(days=1))
|
||||
if next_key in point_buckets:
|
||||
key = next_key
|
||||
bucket = point_buckets.get(key)
|
||||
if bucket is not None:
|
||||
session_bucket_keys.add(key)
|
||||
if s.status != "completed" or bucket is None:
|
||||
continue
|
||||
total, succ = bucket
|
||||
completed_bucket_totals.append(total)
|
||||
if total > 0 and succ == total:
|
||||
coupon_all_success += 1
|
||||
# 每次发起的应领点位数:取「完成过的领券」实际点位数的众数(done 帧会给所有点位终态,
|
||||
# 完成场的点位数=当前配置的全量点位数;数据自校准,配置改点位数无需改代码)。本期无完成场
|
||||
# 时给不出,点位成功率置空。
|
||||
coupon_points_per_session = (
|
||||
Counter(completed_bucket_totals).most_common(1)[0][0]
|
||||
if completed_bucket_totals
|
||||
else None
|
||||
)
|
||||
# 成功点位数:本期 session 触达过的 (device, 日) 桶内成功点位之和(桶级去重,同桶重试不重复计)。
|
||||
coupon_point_success = sum(point_buckets[k][1] for k in session_bucket_keys)
|
||||
# 点位成功率 = 成功点位数 / (发起数 × 应领点位数):中途退出未跑到的点位不产生记录,
|
||||
# 但发起数×点位数把它们计入分母 → 视为失败,符合产品口径;重试会拉低该率(分母按次数计)。
|
||||
coupon_point_success_rate = (
|
||||
round(
|
||||
min(1.0, coupon_point_success / (coupon_started * coupon_points_per_session)), 4
|
||||
)
|
||||
if coupon_started and coupon_points_per_session
|
||||
else None
|
||||
)
|
||||
|
||||
return {
|
||||
"users": {
|
||||
"total": _count(User),
|
||||
@@ -432,7 +563,7 @@ def dashboard_overview(
|
||||
"reward_video_coin_total": _sum(
|
||||
CoinTransaction.amount,
|
||||
CoinTransaction.amount > 0,
|
||||
CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
|
||||
CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES),
|
||||
),
|
||||
"reward_video_watch_count": _count(
|
||||
AdRewardRecord,
|
||||
@@ -480,11 +611,13 @@ def dashboard_overview(
|
||||
"users": {
|
||||
"new": len(period_new_user_ids),
|
||||
"active": len(period_active_user_ids),
|
||||
"retained_new_users": len(period_retained_new_user_ids),
|
||||
"retained_new_users": retention_retained_total,
|
||||
"retention_cohort": retention_cohort_total,
|
||||
"retention_rate": period_retention_rate,
|
||||
"retention_note": (
|
||||
"口径:登录(last_login_at)+开始比价(real_compare_start)+"
|
||||
"开始领券(real_coupon_start/claim_started),按用户去重"
|
||||
"次日留存:窗口内每天取前一日新增用户,统计其当日活跃"
|
||||
"(登录/开始比价/开始领券,按用户去重)比例,逐日累加;"
|
||||
"默认窗口=昨日,即前日新增用户的昨日留存"
|
||||
),
|
||||
},
|
||||
"comparison": {
|
||||
@@ -495,6 +628,17 @@ def dashboard_overview(
|
||||
"average_duration_ms": period_avg_duration_ms,
|
||||
"average_saved_cents": period_avg_saved_cents,
|
||||
},
|
||||
"coupon": {
|
||||
"started": coupon_started,
|
||||
"all_success": coupon_all_success,
|
||||
"success_rate": (
|
||||
round(coupon_all_success / coupon_started, 4) if coupon_started else None
|
||||
),
|
||||
"point_success": coupon_point_success,
|
||||
"points_per_session": coupon_points_per_session,
|
||||
"point_success_rate": coupon_point_success_rate,
|
||||
"median_elapsed_ms": _percentile(coupon_completed_elapsed, 50),
|
||||
},
|
||||
"coins": {
|
||||
"granted_total": _sum(CoinTransaction.amount, *period_coin_conds),
|
||||
"reward_video_coin_total": period_reward_video_coin_total,
|
||||
|
||||
@@ -91,6 +91,7 @@ def get_ad_revenue_report(
|
||||
daily=[AdRevenueDaily(**d) for d in result["daily"]],
|
||||
hourly=[AdRevenueHourly(**h) for h in result["hourly"]],
|
||||
type_stats={k: AdRevenueTypeStat(**v) for k, v in result["type_stats"].items()},
|
||||
scene_stats={k: AdRevenueTypeStat(**v) for k, v in result["scene_stats"].items()},
|
||||
dau=result["dau"],
|
||||
total=result["total"],
|
||||
truncated=result["truncated"],
|
||||
|
||||
@@ -5,7 +5,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request
|
||||
|
||||
from app.admin.audit import write_audit
|
||||
from app.admin.deps import AdminDb, CurrentAdmin, get_client_ip, require_role
|
||||
from app.admin.permissions import SUPER_ADMIN_ROLE
|
||||
from app.admin.permissions import CUSTOM_ROLE, SUPER_ADMIN_ROLE, sanitize_pages
|
||||
from app.admin.repositories import admin_role as role_repo
|
||||
from app.admin.repositories import admin_user as admin_repo
|
||||
from app.admin.schemas.admin import AdminCreateRequest, AdminUpdateRequest
|
||||
@@ -26,14 +26,22 @@ def _active_super_count(db: AdminDb) -> int:
|
||||
|
||||
|
||||
def _validate_role(db: AdminDb, role: str) -> None:
|
||||
"""角色必须是 super_admin 或 admin_role 表里已存在的角色,否则 400。"""
|
||||
if role == SUPER_ADMIN_ROLE:
|
||||
"""角色必须是 super_admin / custom(自定义) / admin_role 表里已存在的角色,否则 400。"""
|
||||
if role in (SUPER_ADMIN_ROLE, CUSTOM_ROLE):
|
||||
return
|
||||
role_repo.ensure_builtin_roles(db) # 空表(测试/全新库)兜底播种,再校验
|
||||
if role_repo.get_role(db, role) is None:
|
||||
raise HTTPException(status_code=400, detail=f"角色不存在: {role}")
|
||||
|
||||
|
||||
def _clean_override(role: str, pages_override: list[str] | None) -> list[str] | None:
|
||||
"""按最终角色算出该存的 pages_override:custom → 勾选集(过滤非法 key,可空列表);
|
||||
非 custom → None(切回普通角色即清空自定义页)。"""
|
||||
if role == CUSTOM_ROLE:
|
||||
return sanitize_pages(pages_override)
|
||||
return None
|
||||
|
||||
|
||||
@router.get("", response_model=list[AdminOut], summary="管理员列表(含明文密码,super_admin 专属)")
|
||||
def list_admins(db: AdminDb) -> list[AdminOut]:
|
||||
out: list[AdminOut] = []
|
||||
@@ -51,13 +59,16 @@ def create_admin(
|
||||
if admin_repo.get_by_username(db, body.username) is not None:
|
||||
raise HTTPException(status_code=409, detail="用户名已存在")
|
||||
_validate_role(db, body.role)
|
||||
override = _clean_override(body.role, body.pages_override)
|
||||
new = admin_repo.create_admin(
|
||||
db, username=body.username, password=body.password, role=body.role,
|
||||
plain_password=body.password, # UI 建的账号留存明文,供权限管理页复看
|
||||
pages_override=override,
|
||||
)
|
||||
write_audit(
|
||||
db, admin, action="admin.create", target_type="admin", target_id=new.id,
|
||||
detail={"username": new.username, "role": new.role}, ip=get_client_ip(request), commit=True,
|
||||
detail={"username": new.username, "role": new.role, "pages_override": override},
|
||||
ip=get_client_ip(request), commit=True,
|
||||
)
|
||||
return AdminOut.model_validate(new)
|
||||
|
||||
@@ -89,7 +100,8 @@ def update_admin(
|
||||
_validate_role(db, body.role)
|
||||
|
||||
changes: dict = {}
|
||||
if body.role is not None and body.role != target.role:
|
||||
role_changed = body.role is not None and body.role != target.role
|
||||
if role_changed:
|
||||
changes["role"] = {"before": target.role, "after": body.role}
|
||||
target.role = body.role
|
||||
if body.status is not None and body.status != target.status:
|
||||
@@ -99,6 +111,17 @@ def update_admin(
|
||||
changes["password"] = "reset"
|
||||
target.password_hash = hash_password(body.password)
|
||||
target.plain_password = body.password # 同步留存明文,权限管理页复看保持一致
|
||||
|
||||
# 自定义可见页:按「最终角色」(target.role,已应用完角色变更)决定该存什么。
|
||||
# - 切到/维持 custom 且传了 pages_override → 用勾选集;切到 custom 没传 → 清成空列表。
|
||||
# - 切回普通角色 → 清空 override(_clean_override 返回 None)。
|
||||
# - 角色没变、只传 pages_override(编辑现有 custom 用户的勾选)→ 也更新。
|
||||
if role_changed or body.pages_override is not None:
|
||||
new_override = _clean_override(target.role, body.pages_override)
|
||||
if new_override != target.pages_override:
|
||||
changes["pages_override"] = {"before": target.pages_override, "after": new_override}
|
||||
target.pages_override = new_override
|
||||
|
||||
if not changes:
|
||||
raise HTTPException(status_code=400, detail="无任何变更字段")
|
||||
db.commit()
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
"""admin 埋点健康度:埋点成功率 / 上报成功率 总览 + 趋势 + 下钻(只读)。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
|
||||
from app.admin.deps import AdminDb, get_current_admin
|
||||
from app.admin.repositories import analytics_health as repo
|
||||
from app.admin.schemas.analytics_health import (
|
||||
HealthBreakdownRow,
|
||||
HealthMetrics,
|
||||
HealthTrendPoint,
|
||||
)
|
||||
|
||||
router = APIRouter(
|
||||
prefix="/admin/api/analytics-health",
|
||||
tags=["admin-analytics-health"],
|
||||
dependencies=[Depends(get_current_admin)],
|
||||
)
|
||||
|
||||
|
||||
@router.get("/overview", response_model=HealthMetrics, summary="两段成功率总览")
|
||||
def overview(
|
||||
db: AdminDb,
|
||||
date_from: Annotated[datetime, Query()],
|
||||
date_to: Annotated[datetime, Query()],
|
||||
) -> HealthMetrics:
|
||||
return HealthMetrics(**repo.overview(db, date_from, date_to))
|
||||
|
||||
|
||||
@router.get("/trend", response_model=list[HealthTrendPoint], summary="按北京天趋势")
|
||||
def trend(
|
||||
db: AdminDb,
|
||||
date_from: Annotated[datetime, Query()],
|
||||
date_to: Annotated[datetime, Query()],
|
||||
) -> list[HealthTrendPoint]:
|
||||
return [HealthTrendPoint(**p) for p in repo.trend(db, date_from, date_to)]
|
||||
|
||||
|
||||
@router.get("/breakdown", response_model=list[HealthBreakdownRow], summary="按维度下钻")
|
||||
def breakdown(
|
||||
db: AdminDb,
|
||||
date_from: Annotated[datetime, Query()],
|
||||
date_to: Annotated[datetime, Query()],
|
||||
dim: Annotated[str, Query(pattern="^(event|app_ver|oem)$")] = "event",
|
||||
) -> list[HealthBreakdownRow]:
|
||||
return [HealthBreakdownRow(**r) for r in repo.breakdown(db, date_from, date_to, dim)]
|
||||
@@ -6,6 +6,7 @@ import logging
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
|
||||
from app.admin.deps import AdminDb, CurrentAdmin
|
||||
from app.admin.permissions import CUSTOM_ROLE, sanitize_pages
|
||||
from app.admin.repositories import admin_role as role_repo
|
||||
from app.admin.repositories import admin_user as admin_repo
|
||||
from app.admin.schemas.auth import AdminLoginRequest, AdminLoginResponse, AdminOut
|
||||
@@ -19,9 +20,13 @@ router = APIRouter(prefix="/admin/api/auth", tags=["admin-auth"])
|
||||
|
||||
|
||||
def _admin_out_with_pages(admin, db: AdminDb) -> AdminOut: # noqa: ANN001
|
||||
"""AdminOut + 当前角色有效可见页(前端左侧导航按此过滤)。"""
|
||||
"""AdminOut + 有效可见页(前端左侧导航按此过滤)。
|
||||
role=="custom" → 用这个人的 pages_override(按人自定义,过滤悬空 key);其余走角色解析。"""
|
||||
out = AdminOut.model_validate(admin)
|
||||
out.pages = role_repo.effective_pages_of(db, admin.role)
|
||||
if admin.role == CUSTOM_ROLE:
|
||||
out.pages = sanitize_pages(admin.pages_override)
|
||||
else:
|
||||
out.pages = role_repo.effective_pages_of(db, admin.role)
|
||||
return out
|
||||
|
||||
|
||||
|
||||
@@ -19,6 +19,8 @@ from app.admin.schemas.ops_marquee_seed import (
|
||||
OpsMarqueeSeedCreate,
|
||||
OpsMarqueeSeedOut,
|
||||
OpsMarqueeSeedUpdate,
|
||||
OpsRealRecordItem,
|
||||
OpsRealRecordsOut,
|
||||
OpsSavingsFeedPreviewOut,
|
||||
)
|
||||
from app.models.admin import AdminUser
|
||||
@@ -68,6 +70,22 @@ def preview_feed(
|
||||
return OpsSavingsFeedPreviewOut(items=ops_marquee.get_feed(db, limit=limit, mode=mode))
|
||||
|
||||
|
||||
@router.get("/real-records", response_model=OpsRealRecordsOut, summary="分页浏览当前模式下可展示的记录(审核用)")
|
||||
def list_real_records(
|
||||
db: AdminDb,
|
||||
mode: Annotated[str | None, Query(description="mixed/real/seed;不传=当前持久化模式")] = None,
|
||||
offset: Annotated[int, Query(ge=0)] = 0,
|
||||
limit: Annotated[int, Query(ge=1, le=50)] = 8,
|
||||
) -> OpsRealRecordsOut:
|
||||
"""分页列出**当前模式**下可在 app 轮播展示的全部记录(**不去重**):只真实=真实记录;只种子=各启用
|
||||
种子按生成逻辑各出一行;混播=真实+种子。与 app 同口径洗牌+去连簇,固定种子→翻页稳定、能翻遍全部。
|
||||
item.user_id=0 表示种子行。"""
|
||||
if mode is not None and mode not in ops_marquee.FEED_MODES:
|
||||
raise HTTPException(status_code=400, detail="mode 需为 mixed / real / seed")
|
||||
items, total = ops_marquee.list_real_records(db, mode=mode, offset=offset, limit=limit)
|
||||
return OpsRealRecordsOut(items=[OpsRealRecordItem(**it) for it in items], total=total)
|
||||
|
||||
|
||||
# 注:/mode 两个端点须在 /{seed_id} 之前注册,否则 PATCH /mode 会被 /{seed_id} 抢先按 id 解析。
|
||||
@router.get("/mode", summary="首页轮播数据源模式(mixed/real/seed)")
|
||||
def get_feed_mode(db: AdminDb) -> dict:
|
||||
|
||||
@@ -42,7 +42,12 @@ def list_users(
|
||||
created_to: Annotated[datetime | None, Query()] = None,
|
||||
last_login_from: Annotated[datetime | None, Query()] = None,
|
||||
last_login_to: Annotated[datetime | None, Query()] = None,
|
||||
sort_by: Annotated[str, Query(pattern="^(id|created_at|last_login_at)$")] = "id",
|
||||
# 最近活跃(登录/发起比价/发起领券取最大,见 queries._last_active_expr)筛选与排序
|
||||
last_active_from: Annotated[datetime | None, Query()] = None,
|
||||
last_active_to: Annotated[datetime | None, Query()] = None,
|
||||
sort_by: Annotated[
|
||||
str, Query(pattern="^(id|created_at|last_login_at|last_active_at)$")
|
||||
] = "id",
|
||||
sort_order: Annotated[str, Query(pattern="^(asc|desc)$")] = "desc",
|
||||
limit: Annotated[int, Query(ge=1, le=100)] = 20,
|
||||
cursor: Annotated[int | None, Query()] = None,
|
||||
@@ -51,6 +56,7 @@ def list_users(
|
||||
db, phone=phone, register_channel=register_channel, status=status,
|
||||
nickname=nickname, created_from=created_from, created_to=created_to,
|
||||
last_login_from=last_login_from, last_login_to=last_login_to,
|
||||
last_active_from=last_active_from, last_active_to=last_active_to,
|
||||
sort_by=sort_by, sort_order=sort_order, limit=limit, cursor=cursor,
|
||||
)
|
||||
return CursorPage(
|
||||
|
||||
@@ -136,6 +136,11 @@ class AdRevenueReportOut(BaseModel):
|
||||
default_factory=dict,
|
||||
description="按广告类型(ad_type)小计 {ad_type: {impressions, revenue_yuan}};前端取 draw / reward_video 做分类大盘",
|
||||
)
|
||||
scene_stats: dict[str, AdRevenueTypeStat] = Field(
|
||||
default_factory=dict,
|
||||
description="按信息流场景(feed_scene)小计 {comparison/coupon/welfare: {impressions, revenue_yuan}};"
|
||||
"全量统计(不受分页截断),供数据大盘「领券广告 / 比价广告」卡;feed_scene 为空的事件不计入",
|
||||
)
|
||||
dau: int | None = Field(
|
||||
None,
|
||||
description="所选日期区间的去重活跃用户数(口径同数据大盘 period.users.active:登录 + 开始比价 + "
|
||||
|
||||
@@ -13,14 +13,18 @@ class AdminCreateRequest(BaseModel):
|
||||
username: str = Field(..., min_length=3, max_length=64)
|
||||
password: str = Field(..., min_length=8, max_length=72) # bcrypt ≤72 字节
|
||||
role: str = Field("operator", min_length=1, max_length=32)
|
||||
# 仅当 role == "custom":这个人专属可见页 key 列表(逐页勾选结果)。其余角色不传/忽略。
|
||||
pages_override: list[str] | None = None
|
||||
|
||||
|
||||
class AdminUpdateRequest(BaseModel):
|
||||
"""改角色 / 启用禁用 / 重置密码,字段都可选(只改传了的)。"""
|
||||
"""改角色 / 启用禁用 / 重置密码 / 自定义可见页,字段都可选(只改传了的)。"""
|
||||
|
||||
role: str | None = Field(None, min_length=1, max_length=32)
|
||||
status: Literal["active", "disabled"] | None = None
|
||||
password: str | None = Field(None, min_length=8, max_length=72)
|
||||
# 改成/更新「自定义」可见页;role 切回普通角色时后端会清空 override(见路由)。
|
||||
pages_override: list[str] | None = None
|
||||
|
||||
|
||||
class AdminAuditLogOut(BaseModel):
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
"""埋点健康度 admin 响应 schema。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class HealthMetrics(BaseModel):
|
||||
attempted: int
|
||||
drop_capture: int
|
||||
delivered: int
|
||||
drop_undelivered: int
|
||||
track_success_rate: float | None
|
||||
report_success_rate: float | None
|
||||
|
||||
|
||||
class HealthTrendPoint(HealthMetrics):
|
||||
day: str
|
||||
|
||||
|
||||
class HealthBreakdownRow(HealthMetrics):
|
||||
key: str
|
||||
@@ -21,8 +21,11 @@ class AdminOut(BaseModel):
|
||||
created_at: datetime
|
||||
last_login_at: datetime | None = None
|
||||
# 该管理员当前角色的有效可见页(= 左侧导航项 key);仅登录 / /me 填充,列表接口默认空。
|
||||
# 前端据此过滤左侧导航(super_admin = 全部页)。见 app/admin/permissions.py。
|
||||
# 前端据此过滤左侧导航(super_admin = 全部页;role=="custom" = pages_override)。见 permissions.py。
|
||||
pages: list[str] = []
|
||||
# 「自定义」可见页原始勾选集(role=="custom" 时非空);列表接口下发,供权限管理页编辑回填勾选。
|
||||
# 普通角色为 None。from_attributes 自动从 ORM 列取。
|
||||
pages_override: list[str] | None = None
|
||||
# 明文登录密码:仅「管理员账号列表」(super_admin 专属路由)填充,供权限管理页编辑时复看;
|
||||
# 无留存(脚本建的超管 / 旧账号)为 None → 前端不显示。登录 / /me 不下发(保持 None)。
|
||||
password: str | None = None
|
||||
|
||||
@@ -43,7 +43,10 @@ class DashboardComparison(BaseModel):
|
||||
class DashboardPeriodUsers(BaseModel):
|
||||
new: int
|
||||
active: int
|
||||
# 次日留存(2026-07-05 起):retained_new_users = 窗口内逐日「前一日新增且当日活跃」用户数之和,
|
||||
# retention_cohort = 对应的前一日新增基数之和,retention_rate = 两者之比。
|
||||
retained_new_users: int
|
||||
retention_cohort: int = 0
|
||||
retention_rate: float | None = None
|
||||
retention_note: str
|
||||
|
||||
@@ -57,6 +60,24 @@ class DashboardPeriodComparison(BaseModel):
|
||||
average_saved_cents: int | None = None
|
||||
|
||||
|
||||
class DashboardPeriodCoupon(BaseModel):
|
||||
"""领券核心数据(2026-07-05 产品新增)。点位=一张券(coupon_claim_record 一天一条终态);
|
||||
成功口径 success+already_claimed(与「我的」页累计领券一致)。"""
|
||||
|
||||
started: int = 0
|
||||
# 全部领成功的次数:completed 且当日该设备全部点位成功
|
||||
all_success: int = 0
|
||||
success_rate: float | None = None
|
||||
# 本期 session 触达的点位中成功的条数(同设备同日去重)
|
||||
point_success: int = 0
|
||||
# 每次发起的应领点位数(本期完成场实际点位数的众数;无完成场为空)
|
||||
points_per_session: int | None = None
|
||||
# 点位成功率 = point_success / (started × points_per_session);未跑到的点位计入分母视为失败
|
||||
point_success_rate: float | None = None
|
||||
# 耗时中位数(仅 completed 的 elapsed_ms,同「领券数据」页口径)
|
||||
median_elapsed_ms: int | None = None
|
||||
|
||||
|
||||
class DashboardPeriodCoins(BaseModel):
|
||||
granted_total: int
|
||||
reward_video_coin_total: int = 0
|
||||
@@ -85,6 +106,7 @@ class DashboardPeriod(BaseModel):
|
||||
date_to: date
|
||||
users: DashboardPeriodUsers
|
||||
comparison: DashboardPeriodComparison
|
||||
coupon: DashboardPeriodCoupon = DashboardPeriodCoupon()
|
||||
coins: DashboardPeriodCoins
|
||||
cash: DashboardPeriodCash
|
||||
trend: list[DashboardTrendPoint] = []
|
||||
|
||||
@@ -62,3 +62,16 @@ class OpsSavingsFeedPreviewItem(BaseModel):
|
||||
class OpsSavingsFeedPreviewOut(BaseModel):
|
||||
"""运营预览:实际混播出来的 feed(真实记录会插队,与客户端一致)。"""
|
||||
items: list[OpsSavingsFeedPreviewItem]
|
||||
|
||||
|
||||
class OpsRealRecordItem(BaseModel):
|
||||
masked_user: str # 脱敏后展示名(与 app 一致)
|
||||
saved_amount_cents: int # 节省金额(分)
|
||||
created_at: str # 比价记录时间(YYYY-MM-DD HH:MM)
|
||||
user_id: int # 真实 user_id(供运营核对,不下发客户端)
|
||||
|
||||
|
||||
class OpsRealRecordsOut(BaseModel):
|
||||
"""分页浏览「全部可展示的真实记录」(success+省>0,不去重、稳定顺序)。"""
|
||||
items: list[OpsRealRecordItem]
|
||||
total: int # 满足条件的真实记录总数(算页数用)
|
||||
|
||||
@@ -20,6 +20,9 @@ class AdminUserListItem(BaseModel):
|
||||
wechat_nickname: str | None = None
|
||||
created_at: datetime
|
||||
last_login_at: datetime
|
||||
# 最近活跃 = max(最近登录, 最近发起比价, 最近发起领券);列表页由 queries._attach_last_active
|
||||
# 瞬态挂上。其他复用本 schema 的入口(用户 360 等)没挂该属性 → None(前端显示 '-')。
|
||||
last_active_at: datetime | None = None
|
||||
|
||||
|
||||
class AdminUserOverview(BaseModel):
|
||||
|
||||
+16
-1
@@ -6,13 +6,18 @@ POST /api/v1/analytics/events — 批量接收新手引导(及后续)埋点,appe
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter, Request
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Request
|
||||
|
||||
from app.api.deps import DbSession
|
||||
from app.repositories import analytics as analytics_repo
|
||||
from app.repositories import analytics_selfstat as selfstat_repo
|
||||
from app.schemas.analytics import AnalyticsBatchIn, AnalyticsIngestOut
|
||||
from app.schemas.analytics_selfstat import SelfStatBatchIn, SelfStatIngestOut
|
||||
|
||||
router = APIRouter(prefix="/api/v1/analytics", tags=["analytics"])
|
||||
logger = logging.getLogger("shagua.analytics")
|
||||
|
||||
|
||||
def _client_ip(request: Request) -> str:
|
||||
@@ -29,3 +34,13 @@ def ingest_events(
|
||||
) -> AnalyticsIngestOut:
|
||||
n = analytics_repo.record_batch(db, batch, client_ip=_client_ip(request))
|
||||
return AnalyticsIngestOut(received=n)
|
||||
|
||||
|
||||
@router.post("/selfstat", response_model=SelfStatIngestOut, summary="上报自报计数快照")
|
||||
def ingest_selfstat(batch: SelfStatBatchIn, db: DbSession) -> SelfStatIngestOut:
|
||||
try:
|
||||
snap_id = selfstat_repo.record_selfstat(db, batch)
|
||||
except Exception: # noqa: BLE001 — 计数链路要稳,落库失败不裸奔 500,记日志回明确错误
|
||||
logger.exception("selfstat ingest failed device=%s epoch=%s", batch.device_id, batch.epoch_id)
|
||||
raise HTTPException(status_code=503, detail="selfstat ingest failed") from None
|
||||
return SelfStatIngestOut(snapshot_id=snap_id)
|
||||
|
||||
@@ -7,6 +7,10 @@ from app.models.ad_watch_log import AdWatchLog # noqa: F401
|
||||
from app.models.admin import AdminAuditLog, AdminUser # noqa: F401
|
||||
from app.models.admin_role import AdminRole # noqa: F401
|
||||
from app.models.analytics_event import AnalyticsEvent # noqa: F401
|
||||
from app.models.analytics_selfstat import ( # noqa: F401
|
||||
AnalyticsSelfStat,
|
||||
AnalyticsSelfStatEvent,
|
||||
)
|
||||
from app.models.app_config import AppConfig # noqa: F401
|
||||
from app.models.comparison import ComparisonRecord # noqa: F401
|
||||
from app.models.cps_activity import CpsActivity # noqa: F401
|
||||
|
||||
+4
-1
@@ -28,8 +28,11 @@ class AdminUser(Base):
|
||||
# 明文登录密码:仅「后台 UI 创建/重置」的管理员留存,供超管在权限管理页复看转交。
|
||||
# 脚本/起后台时建的超管账号不写(为 None → 前端「不显示密码」)。⚠️ 内部工具便利取舍,见 create/list。
|
||||
plain_password: Mapped[str | None] = mapped_column(String(128), nullable=True)
|
||||
# super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘)
|
||||
# super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘)/ custom(按人自定义)
|
||||
role: Mapped[str] = mapped_column(String(20), nullable=False, default="operator")
|
||||
# 「自定义」权限:仅当 role == "custom" 时有效,存这个人专属的可见页 key 列表(不共享给他人)。
|
||||
# 非 custom 用户为 None → 可见页跟随角色。登录/`/me` 下发有效页时,非空即优先用它(见 auth._admin_out_with_pages)。
|
||||
pages_override: Mapped[list | None] = mapped_column(_JSON, nullable=True)
|
||||
# active / disabled
|
||||
status: Mapped[str] = mapped_column(String(20), nullable=False, default="active")
|
||||
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
"""埋点/上报成功率自报计数快照表(append-only)。
|
||||
|
||||
客户端周期上报「自 epoch 起算的累计计数」;服务端只存原始快照,查询时在 Python 侧差分聚合
|
||||
(见 app/admin/repositories/analytics_health.py)。与既有 analytics_event 表完全独立。
|
||||
|
||||
- analytics_selfstat :一快照一行(快照头 + 设备维度 + 设备级诊断量)
|
||||
- analytics_selfstat_event :一 event 一行(四类累计计数),外键指向快照头
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, String, func
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.db.base import Base
|
||||
|
||||
|
||||
class AnalyticsSelfStat(Base):
|
||||
__tablename__ = "analytics_selfstat"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
device_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
|
||||
epoch_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
|
||||
# 设备维度(每设备固定,下钻用)
|
||||
app_ver: Mapped[str | None] = mapped_column(String(32), nullable=True)
|
||||
oem: Mapped[str | None] = mapped_column(String(32), nullable=True)
|
||||
os: Mapped[str | None] = mapped_column(String(32), nullable=True)
|
||||
# 设备级诊断量(累计;queue_depth 是瞬时 gauge)
|
||||
batches_attempted: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
|
||||
batches_ok: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
|
||||
batches_fail: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
|
||||
retries: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
|
||||
queue_depth: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
||||
sent_at: Mapped[int | None] = mapped_column(BigInteger, nullable=True) # 端上报时刻 epoch ms
|
||||
# 服务端接收时间(权威,用于时间分桶与分区排序)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
|
||||
)
|
||||
|
||||
def __repr__(self) -> str: # pragma: no cover
|
||||
return f"<AnalyticsSelfStat id={self.id} device={self.device_id} epoch={self.epoch_id}>"
|
||||
|
||||
|
||||
class AnalyticsSelfStatEvent(Base):
|
||||
__tablename__ = "analytics_selfstat_event"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
snapshot_id: Mapped[int] = mapped_column(
|
||||
ForeignKey("analytics_selfstat.id"), index=True, nullable=False
|
||||
)
|
||||
event: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
|
||||
attempted: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
|
||||
drop_capture: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
|
||||
delivered: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
|
||||
drop_undelivered: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
|
||||
|
||||
def __repr__(self) -> str: # pragma: no cover
|
||||
return f"<AnalyticsSelfStatEvent snap={self.snapshot_id} event={self.event}>"
|
||||
@@ -176,10 +176,19 @@ def grant_feed_reward(
|
||||
)
|
||||
return _commit_record(db, rec, client_event_id)
|
||||
|
||||
# 按点位场景拆流水文案(2026-07):比价等候期看的广告 vs 领券时看的广告,在收益明细里分开显示。
|
||||
# feed_scene=comparison→比价奖励 / coupon→领券奖励;其它(welfare/空/旧端不带)维持通用「信息流广告奖励」。
|
||||
# 客户端按此 biz_type 直显固定文案(见 CoinHistoryViewModel.coinTitle),故 remark 只作后台留痕/兜底。
|
||||
if feed_scene == "comparison":
|
||||
reward_biz, reward_remark = "feed_ad_reward_comparison", "比价奖励"
|
||||
elif feed_scene == "coupon":
|
||||
reward_biz, reward_remark = "feed_ad_reward_coupon", "领券奖励"
|
||||
else:
|
||||
reward_biz, reward_remark = "feed_ad_reward", "信息流广告奖励"
|
||||
crud_wallet.grant_coins(
|
||||
db, user_id, coin,
|
||||
biz_type="feed_ad_reward", ref_id=client_event_id,
|
||||
remark="信息流广告奖励",
|
||||
biz_type=reward_biz, ref_id=client_event_id,
|
||||
remark=reward_remark,
|
||||
)
|
||||
rec = AdFeedRewardRecord(
|
||||
client_event_id=client_event_id,
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
"""自报计数快照落库。一次事务:插 1 条快照头 + N 条 event 行,返回快照 id。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.analytics_selfstat import AnalyticsSelfStat, AnalyticsSelfStatEvent
|
||||
from app.schemas.analytics_selfstat import SelfStatBatchIn
|
||||
|
||||
|
||||
def record_selfstat(db: Session, batch: SelfStatBatchIn) -> int:
|
||||
snap = AnalyticsSelfStat(
|
||||
device_id=batch.device_id,
|
||||
epoch_id=batch.epoch_id,
|
||||
app_ver=batch.app_ver,
|
||||
oem=batch.oem,
|
||||
os=batch.os,
|
||||
batches_attempted=batch.batches_attempted,
|
||||
batches_ok=batch.batches_ok,
|
||||
batches_fail=batch.batches_fail,
|
||||
retries=batch.retries,
|
||||
queue_depth=batch.queue_depth,
|
||||
sent_at=batch.sent_at,
|
||||
)
|
||||
db.add(snap)
|
||||
db.flush() # 拿到 snap.id
|
||||
db.add_all([
|
||||
AnalyticsSelfStatEvent(
|
||||
snapshot_id=snap.id,
|
||||
event=e.event,
|
||||
attempted=e.attempted,
|
||||
drop_capture=e.drop_capture,
|
||||
delivered=e.delivered,
|
||||
drop_undelivered=e.drop_undelivered,
|
||||
)
|
||||
for e in batch.events
|
||||
])
|
||||
db.commit()
|
||||
return snap.id
|
||||
@@ -217,12 +217,28 @@ def _recent_real_rows(db: Session) -> list[tuple[int, int, str | None]]:
|
||||
return out
|
||||
|
||||
|
||||
def _shuffle_declustered(rows: list, rng: random.Random | None = None) -> list:
|
||||
"""洗牌 + 「去连簇」:先洗牌,再贪心重排让相邻两条尽量不是同一 user_id(元素 [0] 即 user_id)。
|
||||
rng=None → 用全局 _rng(feed 每次新随机);传入 rng(如固定种子 Random)→ 排列确定(admin 稳定分页)。
|
||||
减少同一用户连续出现;只有少数几个用户时 best-effort。"""
|
||||
rng = rng or _rng
|
||||
pool = list(rows)
|
||||
rng.shuffle(pool)
|
||||
result: list = []
|
||||
while pool:
|
||||
prev_uid = result[-1][0] if result else None
|
||||
# 优先挑与上一条不同 user 的;挑不到(只剩同 user)才取第一个
|
||||
idx = next((i for i, r in enumerate(pool) if r[0] != prev_uid), 0)
|
||||
result.append(pool.pop(idx))
|
||||
return result
|
||||
|
||||
|
||||
def get_feed(db: Session, limit: int = 8, mode: str | None = None) -> list[dict]:
|
||||
"""返回最多 limit 条 {masked_user, saved_amount_cents, time(HH:MM:SS 北京)}。
|
||||
|
||||
mode:显式传入(admin 预览指定模式)则用它、**不改持久化配置**;不传(客户端 /savings-feed)读
|
||||
持久化的 marquee_feed_mode;非法值一律回退到持久化模式。
|
||||
真实条:success 且 0 < saved ≤ 上限,按 user 去重(同一用户只取最新一条,避免单人刷屏)。
|
||||
真实条:success 且 0 < saved ≤ 上限,**不按 user 去重**(打乱 + 去连簇:相邻尽量不同用户、减少单人连刷)后取前 limit。
|
||||
不足用启用的种子补齐——**公平随机抽取** need 个(而非固定取前 N),让所有种子都有机会露出;
|
||||
种子用户名留空则随机合成(避开撞名),金额取**长尾随机**(小额居多、偶尔大额,更像真实分布)。
|
||||
真实 + 种子仍不满 limit → 内置合成条**补满**,保证轮播既不空也不稀疏。
|
||||
@@ -235,19 +251,13 @@ def get_feed(db: Session, limit: int = 8, mode: str | None = None) -> list[dict]
|
||||
items: list[dict] = []
|
||||
used_names: set[str] = set()
|
||||
|
||||
# 真实条(mixed / real):取较多近期记录(带 ~30s 缓存)后按 user 去重;金额超上限的异常值已在查询剔除。
|
||||
# 真实条(mixed / real):**不按 user 去重**——打乱 + 去连簇(相邻尽量不同用户、减少单人连刷)后取前
|
||||
# limit;金额超上限的异常值已在查询剔除。同一用户可多条露出,但被打散、尽量不连续。
|
||||
if mode != "seed":
|
||||
rows = _recent_real_rows(db)
|
||||
seen_users: set[int] = set()
|
||||
for uid, sc, nick in rows:
|
||||
if uid in seen_users:
|
||||
continue
|
||||
seen_users.add(uid)
|
||||
for uid, sc, nick in _shuffle_declustered(_recent_real_rows(db))[:limit]:
|
||||
name = _mask_real(nick, uid)
|
||||
used_names.add(name) # 真实名按昵称/id 稳定;偶发撞名可接受
|
||||
used_names.add(name) # 同一用户可多条,名字重复无害(used_names 仅供种子避重)
|
||||
items.append({"masked_user": name, "saved_amount_cents": int(sc)})
|
||||
if len(items) >= limit:
|
||||
break
|
||||
|
||||
# 种子补位 + 合成兜底(mixed / seed):mixed 下补真实不足的部分,seed 下全量用种子/合成。
|
||||
# real 模式**跳过**——只出真实,不掺任何假数据(真实不足则少于 limit,为 0 时返回空)。
|
||||
@@ -300,6 +310,78 @@ def get_feed(db: Session, limit: int = 8, mode: str | None = None) -> list[dict]
|
||||
return items
|
||||
|
||||
|
||||
# ===== admin 侧:分页浏览「当前模式下可展示的记录」(审核用,不去重) =====
|
||||
_REAL_BROWSE_CAP = 1000 # 真实记录一次最多纳入这么多去洗牌+分页(足够审核;防超大库全量洗牌)
|
||||
_BROWSE_SEED = 20260707 # 固定洗牌种子:同一批数据下排列恒定 → 翻页稳定、能翻遍全部
|
||||
|
||||
|
||||
def _seed_browse_row(seed: OpsMarqueeSeed) -> dict:
|
||||
"""把一条种子按其「生成逻辑」**确定性**生成一行浏览项(名字/金额按 seed.id 派生固定种子 → 翻页稳定)。
|
||||
名字:运营手填的非模板名原样用,否则本地合成;金额:区间内确定性长尾取值。user_id=0(种子无真实用户)。"""
|
||||
r = random.Random(_BROWSE_SEED * 1_000_003 + int(seed.id))
|
||||
fixed = (seed.masked_user or "").strip()
|
||||
name = fixed if (fixed and not fixed.startswith("用户****")) else _synth_name(r)
|
||||
lo = max(0, int(seed.min_cents))
|
||||
hi = max(lo, int(seed.max_cents))
|
||||
amt = lo if hi <= lo else lo + int(round((hi - lo) * (r.random() ** 2.2)))
|
||||
return {"masked_user": name, "saved_amount_cents": amt, "created_at": "", "user_id": 0}
|
||||
|
||||
|
||||
def list_real_records(
|
||||
db: Session, mode: str | None = None, offset: int = 0, limit: int = 8
|
||||
) -> tuple[list[dict], int]:
|
||||
"""分页浏览「**当前模式**下可在 app 轮播展示的全部记录」,供 admin 逐页审核(**不去重**):
|
||||
- 只真实:全部 success+省>0 的真实记录;
|
||||
- 只种子:每条启用种子按生成逻辑各出一行;
|
||||
- 混播:真实 + 种子 全部合在一起。
|
||||
与 app 轮播同口径:先洗牌 + 去连簇(相邻尽量不同 user;种子各自独立、不算同 user);**固定种子** →
|
||||
同批数据下排列恒定,翻页不跳、能翻遍全部。返回 (items, total);item.user_id=0 表示种子。"""
|
||||
mode = mode if mode in FEED_MODES else get_feed_mode(db)
|
||||
# pool: [(cluster_key, item)];cluster_key 供去连簇——真实=user_id、种子=各自唯一负数(互不聚簇)
|
||||
pool: list[tuple[int, dict]] = []
|
||||
if mode != "seed":
|
||||
rows = db.execute(
|
||||
select(
|
||||
ComparisonRecord.user_id,
|
||||
ComparisonRecord.saved_amount_cents,
|
||||
User.nickname,
|
||||
ComparisonRecord.created_at,
|
||||
)
|
||||
.join(User, User.id == ComparisonRecord.user_id)
|
||||
.where(
|
||||
ComparisonRecord.status == "success",
|
||||
ComparisonRecord.saved_amount_cents > 0,
|
||||
ComparisonRecord.saved_amount_cents <= _REAL_MAX_CENTS,
|
||||
)
|
||||
.order_by(ComparisonRecord.created_at.desc())
|
||||
.limit(_REAL_BROWSE_CAP)
|
||||
).all()
|
||||
for uid, sc, nick, ca in rows:
|
||||
pool.append((
|
||||
int(uid),
|
||||
{
|
||||
"masked_user": _mask_real(nick, int(uid)),
|
||||
"saved_amount_cents": int(sc),
|
||||
"created_at": str(ca)[:16] if ca is not None else "",
|
||||
"user_id": int(uid),
|
||||
},
|
||||
))
|
||||
if mode != "real":
|
||||
seeds = (
|
||||
db.execute(select(OpsMarqueeSeed).where(OpsMarqueeSeed.enabled.is_(True)))
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
for i, s in enumerate(seeds):
|
||||
pool.append((-(i + 1), _seed_browse_row(s))) # 每个种子唯一 key → 互不聚簇
|
||||
total = len(pool)
|
||||
# 每次用同一固定种子新建 Random → 同批数据排列恒定(翻页稳定);同时相邻尽量不同 user。
|
||||
ordered = _shuffle_declustered(pool, random.Random(_BROWSE_SEED))
|
||||
off = max(0, offset)
|
||||
items = [item for _key, item in ordered[off : off + limit]]
|
||||
return items, total
|
||||
|
||||
|
||||
# ===== 运营侧:种子 CRUD =====
|
||||
def list_seeds(db: Session) -> list[OpsMarqueeSeed]:
|
||||
return list(
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
"""自报计数上报 schema。字段名对齐客户端 payload(snake_case),累计值语义。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class SelfStatEventIn(BaseModel):
|
||||
event: str = Field(max_length=64)
|
||||
attempted: int = 0
|
||||
drop_capture: int = 0
|
||||
delivered: int = 0
|
||||
drop_undelivered: int = 0
|
||||
|
||||
|
||||
class SelfStatBatchIn(BaseModel):
|
||||
device_id: str = Field(max_length=64)
|
||||
epoch_id: str = Field(max_length=64)
|
||||
sent_at: int | None = None
|
||||
app_ver: str | None = Field(default=None, max_length=32)
|
||||
oem: str | None = Field(default=None, max_length=32)
|
||||
os: str | None = Field(default=None, max_length=32)
|
||||
batches_attempted: int = 0
|
||||
batches_ok: int = 0
|
||||
batches_fail: int = 0
|
||||
retries: int = 0
|
||||
queue_depth: int = 0
|
||||
# 允许空列表:某次快照只上报设备级计数(batches_*/retries/queue_depth)、无 event 细分时也合法
|
||||
# (与 AnalyticsBatchIn 的 min_length=1 有意不同——那是行为事件、必须至少一条)。
|
||||
events: list[SelfStatEventIn] = Field(default_factory=list, max_length=200)
|
||||
|
||||
|
||||
class SelfStatIngestOut(BaseModel):
|
||||
ok: bool = True
|
||||
snapshot_id: int
|
||||
@@ -0,0 +1,166 @@
|
||||
"""收益明细「金币记录」文案验证脚手架(2026-07 文案改版验收用)。
|
||||
|
||||
问题:客户端按 bizType 显示固定文案(路线B,强制覆盖后端 remark),但账号若没有对应
|
||||
bizType 的流水,收益明细页就空着,无从验证。本脚本往指定测试用户塞每种 bizType 各一条
|
||||
流水,让你在手机收益明细页一屏核对全部新文案;验收完 --clean 一键删除,不污染数据。
|
||||
|
||||
⚠️ 仅 dev 库用(APP_ENV=dev 时才允许 --seed/--clean)。所有测试流水 ref_id 前缀 TESTDOC,
|
||||
按前缀精确清理,不会误删真实流水。
|
||||
|
||||
用法(pricebot env 直调,见项目 CLAUDE.md):
|
||||
D:/miniconda/envs/pricebot/python.exe scripts/seed_coinhistory_labels_test.py --show
|
||||
D:/miniconda/envs/pricebot/python.exe scripts/seed_coinhistory_labels_test.py --seed
|
||||
D:/miniconda/envs/pricebot/python.exe scripts/seed_coinhistory_labels_test.py --clean
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.rewards import CN_TZ
|
||||
from datetime import datetime
|
||||
|
||||
from app.db.session import SessionLocal
|
||||
from app.models.user import User
|
||||
from app.models.wallet import CoinAccount, CoinTransaction
|
||||
|
||||
TEST_PHONE = "11111111111"
|
||||
REF_PREFIX = "TESTDOC" # 所有本脚本造的流水都带这个 ref_id 前缀,便于精确清理
|
||||
|
||||
# 每条 = (bizType, 故意写错/留空的 remark, 金币数)。
|
||||
# remark 故意填「错的」→ 若客户端仍显示新文案 = 证明路线B强制覆盖生效(无视后端 remark)。
|
||||
# task_ 一条留空 remark → 走客户端兜底映射。顺序即手机上从新到旧的展示顺序(后塞的在最上)。
|
||||
CASES: list[tuple[str, str, int]] = [
|
||||
("signin", "每日签到 第99天(旧文案,应被覆盖)", 220),
|
||||
("signin_boost", "签到膨胀 第99天", 3000),
|
||||
("reward_video", "看视频奖励金币(旧文案,应被覆盖)", 200),
|
||||
("feed_ad_reward_comparison", "", 50), # 后端新拆:比价场景 → 比价奖励
|
||||
("feed_ad_reward_coupon", "", 50), # 后端新拆:领券场景 → 领券奖励
|
||||
("feed_ad_reward", "信息流广告奖励(welfare/旧数据兜底)", 50),
|
||||
("price_report_reward", "上报更低价审核通过(旧文案,应被覆盖)", 1000),
|
||||
("feedback_reward", "意见反馈被采纳(旧文案,应被覆盖)", 10000),
|
||||
("task_enable_notification", "", 750), # remark 留空 → 客户端「打开消息提醒奖励」
|
||||
# 下两条现实中不会进金币记录(invite 发现金进邀请钱包 / compare_milestone 后端死代码不发钱),
|
||||
# 仅用于验证「杀掉好友比价奖励 + 兜底改任务奖励」:两条都应显示「任务奖励」(remark 被强制无视)。
|
||||
("invite", "好友比价奖励(旧文案,应被杀→任务奖励)", 200),
|
||||
("compare_milestone", "", 120),
|
||||
]
|
||||
|
||||
|
||||
def _client_coin_title(biz_type: str, remark: str | None) -> str:
|
||||
"""复刻 CoinHistoryViewModel.coinTitle 的最新逻辑(路线B),用于 --show 预览。
|
||||
|
||||
⚠️ 必须与客户端保持一致;客户端改了这里也要同步,否则预览会骗人。
|
||||
"""
|
||||
fixed = {
|
||||
"exchange_out": "金币兑换现金",
|
||||
"signin": "每日签到奖励",
|
||||
"signin_boost": "签到膨胀奖励",
|
||||
"reward_video": "看视频赚金币",
|
||||
"ad_reward": "看视频赚金币",
|
||||
"feed_ad_reward_comparison": "比价奖励",
|
||||
"feed_ad_reward_coupon": "领券奖励",
|
||||
"feed_ad_reward": "信息流广告奖励",
|
||||
"price_report_reward": "爆料奖励",
|
||||
"feedback_reward": "反馈奖励",
|
||||
"invite": "任务奖励", # 杀掉"好友比价奖励",归兜底
|
||||
}
|
||||
if biz_type in fixed:
|
||||
return fixed[biz_type]
|
||||
if remark:
|
||||
return remark
|
||||
if biz_type == "task_enable_notification":
|
||||
return "打开消息提醒奖励"
|
||||
return "任务奖励"
|
||||
|
||||
|
||||
def _get_user(db) -> User:
|
||||
u = db.query(User).filter(User.phone == TEST_PHONE).first()
|
||||
if not u:
|
||||
print(f"✗ 库里没有测试号 {TEST_PHONE} —— 先在手机上用这个号登录一次再跑本脚本。")
|
||||
sys.exit(1)
|
||||
return u
|
||||
|
||||
|
||||
def cmd_show() -> None:
|
||||
"""只打印:每种 bizType 经客户端映射后会显示成什么(不写库)。"""
|
||||
print("bizType 造流水后,收益明细页预期显示的文案:\n")
|
||||
print(f" {'bizType':32} {'后端remark(故意填的)':32} → 手机显示")
|
||||
print(" " + "-" * 90)
|
||||
for biz, remark, _coin in CASES:
|
||||
shown = _client_coin_title(biz, remark or None)
|
||||
rk = (remark or "(空)")
|
||||
print(f" {biz:32} {rk:32} → {shown}")
|
||||
print("\n注:remark 列是故意填的『旧/错』文案;'手机显示'若为新文案 = 强制覆盖生效。")
|
||||
print("invite / compare_milestone 已从客户端映射删除:invite 有 remark 故显示原样,")
|
||||
print("compare_milestone remark 空故落兜底『奖励』—— 两者都不再有专属新文案(符合『去掉』)。")
|
||||
|
||||
|
||||
def cmd_seed() -> None:
|
||||
if settings.APP_ENV != "dev":
|
||||
print(f"✗ 拒绝:APP_ENV={settings.APP_ENV},本脚本只在 dev 库造测试数据。")
|
||||
sys.exit(1)
|
||||
db = SessionLocal()
|
||||
u = _get_user(db)
|
||||
acc = db.query(CoinAccount).filter(CoinAccount.user_id == u.id).first()
|
||||
if acc is None:
|
||||
acc = CoinAccount(user_id=u.id, coin_balance=0, cash_balance_cents=0)
|
||||
db.add(acc)
|
||||
db.flush()
|
||||
now = datetime.now(CN_TZ).replace(tzinfo=None)
|
||||
made = 0
|
||||
for i, (biz, remark, coin) in enumerate(CASES):
|
||||
ref = f"{REF_PREFIX}:{biz}:{i}"
|
||||
exists = db.query(CoinTransaction).filter(CoinTransaction.ref_id == ref).first()
|
||||
if exists:
|
||||
continue
|
||||
acc.coin_balance += coin
|
||||
db.add(CoinTransaction(
|
||||
user_id=u.id, amount=coin, balance_after=acc.coin_balance,
|
||||
biz_type=biz, ref_id=ref, remark=remark or None, created_at=now,
|
||||
))
|
||||
made += 1
|
||||
db.commit()
|
||||
print(f"✓ 已给 user_id={u.id}({TEST_PHONE})造 {made} 条测试流水,当前金币余额 {acc.coin_balance}。")
|
||||
print(" → 打开手机 App「收益明细 / 金币记录」下拉刷新,逐条核对文案。")
|
||||
print(" → 验收完跑 --clean 删除这些测试流水。")
|
||||
|
||||
|
||||
def cmd_clean() -> None:
|
||||
if settings.APP_ENV != "dev":
|
||||
print(f"✗ 拒绝:APP_ENV={settings.APP_ENV}。")
|
||||
sys.exit(1)
|
||||
db = SessionLocal()
|
||||
u = _get_user(db)
|
||||
rows = db.query(CoinTransaction).filter(
|
||||
CoinTransaction.user_id == u.id,
|
||||
CoinTransaction.ref_id.like(f"{REF_PREFIX}:%"),
|
||||
).all()
|
||||
total = sum(r.amount for r in rows)
|
||||
for r in rows:
|
||||
db.delete(r)
|
||||
acc = db.query(CoinAccount).filter(CoinAccount.user_id == u.id).first()
|
||||
if acc is not None:
|
||||
acc.coin_balance -= total # 把造流水时加的余额扣回,还原
|
||||
db.commit()
|
||||
print(f"✓ 已删除 {len(rows)} 条 TESTDOC 测试流水,余额回扣 {total},当前 {acc.coin_balance if acc else 0}。")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser(description="收益明细金币文案验证脚手架")
|
||||
g = ap.add_mutually_exclusive_group(required=True)
|
||||
g.add_argument("--show", action="store_true", help="只打印每种 bizType 的预期显示文案,不写库")
|
||||
g.add_argument("--seed", action="store_true", help="往测试号造每种 bizType 各一条流水")
|
||||
g.add_argument("--clean", action="store_true", help="删除本脚本造的所有测试流水")
|
||||
args = ap.parse_args()
|
||||
if args.show:
|
||||
cmd_show()
|
||||
elif args.seed:
|
||||
cmd_seed()
|
||||
elif args.clean:
|
||||
cmd_clean()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -320,3 +320,92 @@ def test_read_apis_require_auth(admin_client: TestClient) -> None:
|
||||
"/admin/api/feedbacks",
|
||||
]:
|
||||
assert admin_client.get(path).status_code == 401, path
|
||||
|
||||
|
||||
def test_period_comparison_reward_coin_from_feed_scene(
|
||||
admin_client: TestClient, admin_token: str
|
||||
) -> None:
|
||||
"""比价奖励金币口径:按 ad_feed_reward_record.feed_scene='comparison' 的实发金币
|
||||
(status=granted)汇总,而非查从不写入的 biz_type 桶(修复大盘该卡恒 0)。
|
||||
too_short(未发奖)不计。"""
|
||||
from app.models.ad_feed_reward import AdFeedRewardRecord
|
||||
|
||||
d = "2021-06-15" # 独立历史日,隔离其它用例数据
|
||||
db = SessionLocal()
|
||||
try:
|
||||
uid = user_repo.upsert_user_for_login(db, phone="13800008801", register_channel="sms").id
|
||||
db.add(AdFeedRewardRecord(
|
||||
client_event_id="cmp-fs-granted", user_id=uid, reward_date=d,
|
||||
ecpm_raw="0", coin=123, feed_scene="comparison", status="granted",
|
||||
))
|
||||
db.add(AdFeedRewardRecord(
|
||||
client_event_id="cmp-fs-tooshort", user_id=uid, reward_date=d,
|
||||
ecpm_raw="0", coin=99, feed_scene="comparison", status="too_short",
|
||||
))
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
r = admin_client.get(
|
||||
"/admin/api/stats/overview", params={"date_from": d, "date_to": d},
|
||||
headers=_auth(admin_token),
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["period"]["coins"]["comparison_reward_coin_total"] == 123
|
||||
|
||||
|
||||
def test_period_coupon_reward_coin_from_feed_scene(
|
||||
admin_client: TestClient, admin_token: str
|
||||
) -> None:
|
||||
"""领券奖励金币口径:按 ad_feed_reward_record.feed_scene='coupon' 的实发金币汇总。"""
|
||||
from app.models.ad_feed_reward import AdFeedRewardRecord
|
||||
|
||||
d = "2021-06-16"
|
||||
db = SessionLocal()
|
||||
try:
|
||||
uid = user_repo.upsert_user_for_login(db, phone="13800008802", register_channel="sms").id
|
||||
db.add(AdFeedRewardRecord(
|
||||
client_event_id="cpn-fs-granted", user_id=uid, reward_date=d,
|
||||
ecpm_raw="0", coin=456, feed_scene="coupon", status="granted",
|
||||
))
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
r = admin_client.get(
|
||||
"/admin/api/stats/overview", params={"date_from": d, "date_to": d},
|
||||
headers=_auth(admin_token),
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["period"]["coins"]["coupon_reward_coin_total"] == 456
|
||||
|
||||
|
||||
def test_period_coupon_reward_excludes_reward_video(
|
||||
admin_client: TestClient, admin_token: str
|
||||
) -> None:
|
||||
"""领券奖励金币不再把激励视频金币算进来(reward_video/ad_reward 从领券桶拆出);
|
||||
激励视频仍单独计入 reward_video_coin_total。"""
|
||||
from datetime import datetime
|
||||
|
||||
from app.models.wallet import CoinTransaction
|
||||
|
||||
d = "2021-06-17"
|
||||
db = SessionLocal()
|
||||
try:
|
||||
uid = user_repo.upsert_user_for_login(db, phone="13800008803", register_channel="sms").id
|
||||
db.add(CoinTransaction(
|
||||
user_id=uid, amount=50, balance_after=50, biz_type="reward_video",
|
||||
ref_id="rv-split-1", created_at=datetime(2021, 6, 17, 12, 0, 0),
|
||||
))
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
r = admin_client.get(
|
||||
"/admin/api/stats/overview", params={"date_from": d, "date_to": d},
|
||||
headers=_auth(admin_token),
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
coins = r.json()["period"]["coins"]
|
||||
assert coins["coupon_reward_coin_total"] == 0 # 激励视频不计入领券奖励
|
||||
assert coins["reward_video_coin_total"] == 50 # 仍计入激励视频卡
|
||||
|
||||
@@ -150,3 +150,80 @@ def test_create_admin_rejects_unknown_role(admin_client, super_token) -> None:
|
||||
headers=_auth(super_token),
|
||||
)
|
||||
assert r.status_code == 400
|
||||
|
||||
|
||||
def _login_pages(username: str, password: str = "pass1234") -> list[str]:
|
||||
"""以某账号登录,取下发的有效可见页(左侧导航过滤依据)。"""
|
||||
c = TestClient(admin_app)
|
||||
return c.post(
|
||||
"/admin/api/auth/login", json={"username": username, "password": password}
|
||||
).json()["admin"]["pages"]
|
||||
|
||||
|
||||
def test_custom_role_create_pages_take_effect(admin_client, super_token) -> None:
|
||||
# 建「自定义」账号:role=custom + 勾选页(含一个非法 key,应被过滤)
|
||||
r = admin_client.post(
|
||||
"/admin/api/admins",
|
||||
json={
|
||||
"username": "cust_user", "password": "pass1234", "role": "custom",
|
||||
"pages_override": ["dashboard", "withdraws", "xx-bad"],
|
||||
},
|
||||
headers=_auth(super_token),
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
# 登录下发的 pages == 勾选集(非法 key 过滤),真正驱动左侧导航
|
||||
assert set(_login_pages("cust_user")) == {"dashboard", "withdraws"}
|
||||
# 账号列表回显原始勾选集(供编辑回填),同样已过滤
|
||||
admins = {a["username"]: a for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()}
|
||||
assert set(admins["cust_user"]["pages_override"]) == {"dashboard", "withdraws"}
|
||||
assert admins["cust_user"]["role"] == "custom"
|
||||
|
||||
|
||||
def test_custom_role_update_and_switch_back_clears_override(admin_client, super_token) -> None:
|
||||
admin_client.post(
|
||||
"/admin/api/admins",
|
||||
json={
|
||||
"username": "cust_sw", "password": "pass1234", "role": "custom",
|
||||
"pages_override": ["dashboard"],
|
||||
},
|
||||
headers=_auth(super_token),
|
||||
)
|
||||
aid = next(
|
||||
a["id"] for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()
|
||||
if a["username"] == "cust_sw"
|
||||
)
|
||||
# 只改勾选集(角色不变)→ 生效
|
||||
u = admin_client.patch(
|
||||
f"/admin/api/admins/{aid}", json={"pages_override": ["dashboard", "feedbacks"]},
|
||||
headers=_auth(super_token),
|
||||
)
|
||||
assert u.status_code == 200, u.text
|
||||
assert set(_login_pages("cust_sw")) == {"dashboard", "feedbacks"}
|
||||
# 切回普通角色 operator → override 清空,pages 跟随角色(运营页集,且不含 admins)
|
||||
u2 = admin_client.patch(
|
||||
f"/admin/api/admins/{aid}", json={"role": "operator"}, headers=_auth(super_token)
|
||||
)
|
||||
assert u2.status_code == 200, u2.text
|
||||
admins = {a["username"]: a for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()}
|
||||
assert admins["cust_sw"]["pages_override"] is None
|
||||
pages = set(_login_pages("cust_sw"))
|
||||
assert "feedbacks" in pages and "admins" not in pages # 运营口径,自定义页已不生效
|
||||
|
||||
|
||||
def test_switch_operator_to_custom_sets_override(admin_client, super_token) -> None:
|
||||
admin_client.post(
|
||||
"/admin/api/admins",
|
||||
json={"username": "op2cust", "password": "pass1234", "role": "operator"},
|
||||
headers=_auth(super_token),
|
||||
)
|
||||
aid = next(
|
||||
a["id"] for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()
|
||||
if a["username"] == "op2cust"
|
||||
)
|
||||
u = admin_client.patch(
|
||||
f"/admin/api/admins/{aid}",
|
||||
json={"role": "custom", "pages_override": ["config"]},
|
||||
headers=_auth(super_token),
|
||||
)
|
||||
assert u.status_code == 200, u.text
|
||||
assert set(_login_pages("op2cust")) == {"config"}
|
||||
|
||||
@@ -0,0 +1,160 @@
|
||||
"""埋点健康度聚合:纯差分函数 + admin 端点。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from app.admin.repositories.analytics_health import _cn_day, _rates, diff_snapshots
|
||||
|
||||
|
||||
def _row(device, epoch, event, ts, **cum) -> dict:
|
||||
base = {"attempted": 0, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0}
|
||||
base.update(cum)
|
||||
return {
|
||||
"device_id": device, "epoch_id": epoch, "event": event,
|
||||
"created_at": datetime(2026, 7, 1, ts, 0, tzinfo=timezone.utc),
|
||||
"app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
|
||||
**base,
|
||||
}
|
||||
|
||||
|
||||
def test_diff_first_row_is_full_cumulative() -> None:
|
||||
rows = [_row("d", "e", "video_play", 1, attempted=100, delivered=90)]
|
||||
out = diff_snapshots(rows)
|
||||
assert len(out) == 1
|
||||
assert out[0]["d_attempted"] == 100
|
||||
assert out[0]["d_delivered"] == 90
|
||||
|
||||
|
||||
def test_diff_consecutive_delta() -> None:
|
||||
rows = [
|
||||
_row("d", "e", "video_play", 1, attempted=100, delivered=90),
|
||||
_row("d", "e", "video_play", 2, attempted=150, delivered=140),
|
||||
]
|
||||
out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"])
|
||||
assert out[1]["d_attempted"] == 50
|
||||
assert out[1]["d_delivered"] == 50
|
||||
|
||||
|
||||
def test_diff_epoch_reset_new_partition() -> None:
|
||||
rows = [
|
||||
_row("d", "e1", "video_play", 1, attempted=100),
|
||||
_row("d", "e2", "video_play", 2, attempted=5),
|
||||
]
|
||||
out = {(r["epoch_id"]): r["d_attempted"] for r in diff_snapshots(rows)}
|
||||
assert out["e1"] == 100
|
||||
assert out["e2"] == 5
|
||||
|
||||
|
||||
def test_diff_clamps_negative_on_reorder() -> None:
|
||||
rows = [
|
||||
_row("d", "e", "video_play", 1, attempted=100),
|
||||
_row("d", "e", "video_play", 2, attempted=80),
|
||||
]
|
||||
out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"])
|
||||
assert out[1]["d_attempted"] == 0
|
||||
|
||||
|
||||
def test_rates_computes_both_formulas() -> None:
|
||||
out = _rates({"attempted": 150, "drop_capture": 0, "delivered": 140, "drop_undelivered": 10})
|
||||
assert out["track_success_rate"] == 1.0
|
||||
assert out["report_success_rate"] == 140 / 150
|
||||
|
||||
|
||||
def test_rates_zero_denominator_yields_none() -> None:
|
||||
out = _rates({"attempted": 0, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0})
|
||||
assert out["track_success_rate"] is None
|
||||
assert out["report_success_rate"] is None
|
||||
|
||||
|
||||
def test_cn_day_beijing_boundary() -> None:
|
||||
# UTC 15:59 → 北京 23:59 同日;UTC 16:00 → 北京 次日 00:00
|
||||
assert _cn_day(datetime(2026, 7, 1, 15, 59, tzinfo=timezone.utc)) == "2026-07-01"
|
||||
assert _cn_day(datetime(2026, 7, 1, 16, 0, tzinfo=timezone.utc)) == "2026-07-02"
|
||||
|
||||
|
||||
def test_diff_same_timestamp_ordered_by_id() -> None:
|
||||
ts = datetime(2026, 7, 1, 1, 0, tzinfo=timezone.utc)
|
||||
# 相同 created_at,乱序传入(id=2 在前);应按 id 排序 → id=1(cum100) 在前
|
||||
rows = [
|
||||
{"id": 2, "device_id": "d", "epoch_id": "e", "event": "vp", "created_at": ts,
|
||||
"app_ver": "v", "oem": "o", "os": "s",
|
||||
"attempted": 150, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0},
|
||||
{"id": 1, "device_id": "d", "epoch_id": "e", "event": "vp", "created_at": ts,
|
||||
"app_ver": "v", "oem": "o", "os": "s",
|
||||
"attempted": 100, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0},
|
||||
]
|
||||
out = diff_snapshots(rows)
|
||||
assert out[0]["d_attempted"] == 100 # id=1 sorts first → its cumulative
|
||||
assert out[1]["d_attempted"] == 50 # id=2 second → 150-100
|
||||
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.admin.main import admin_app
|
||||
from app.admin.repositories import admin_user as admin_repo
|
||||
from app.db.session import SessionLocal
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def admin_client() -> TestClient:
|
||||
return TestClient(admin_app)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def admin_token() -> str:
|
||||
db = SessionLocal()
|
||||
try:
|
||||
if admin_repo.get_by_username(db, "health_admin") is None:
|
||||
admin_repo.create_admin(db, username="health_admin", password="pw", role="super_admin")
|
||||
finally:
|
||||
db.close()
|
||||
c = TestClient(admin_app)
|
||||
r = c.post("/admin/api/auth/login", json={"username": "health_admin", "password": "pw"})
|
||||
return r.json()["access_token"]
|
||||
|
||||
|
||||
def _auth(t: str) -> dict:
|
||||
return {"Authorization": f"Bearer {t}"}
|
||||
|
||||
|
||||
def _seed_two_snapshots(client: TestClient) -> None:
|
||||
for attempted, delivered in ((100, 90), (150, 140)):
|
||||
client.post("/api/v1/analytics/selfstat", json={
|
||||
"device_id": "d-health", "epoch_id": "e-health",
|
||||
"app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
|
||||
"events": [{"event": "video_play", "attempted": attempted,
|
||||
"drop_capture": 0, "delivered": delivered, "drop_undelivered": 0}],
|
||||
})
|
||||
|
||||
|
||||
def test_health_overview_requires_auth(admin_client: TestClient) -> None:
|
||||
r = admin_client.get("/admin/api/analytics-health/overview",
|
||||
params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z"})
|
||||
assert r.status_code == 401
|
||||
|
||||
|
||||
def test_health_overview(client: TestClient, admin_client: TestClient, admin_token: str) -> None:
|
||||
_seed_two_snapshots(client) # 播种走公开 app(ingest 端点在 app,不在 admin_app)
|
||||
r = admin_client.get(
|
||||
"/admin/api/analytics-health/overview",
|
||||
params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z"},
|
||||
headers=_auth(admin_token),
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
data = r.json()
|
||||
# 两快照累计 100→150,差分后 attempted 总 150(首快照 100 + 增量 50)
|
||||
assert data["attempted"] == 150
|
||||
assert data["track_success_rate"] == 1.0
|
||||
|
||||
|
||||
def test_health_breakdown(client: TestClient, admin_client: TestClient, admin_token: str) -> None:
|
||||
_seed_two_snapshots(client) # 播种走公开 app
|
||||
r = admin_client.get(
|
||||
"/admin/api/analytics-health/breakdown",
|
||||
params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z", "dim": "event"},
|
||||
headers=_auth(admin_token),
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
rows = r.json()
|
||||
assert any(row["key"] == "video_play" for row in rows)
|
||||
@@ -0,0 +1,43 @@
|
||||
"""自报计数上报端点测试。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
def _payload(**over) -> dict:
|
||||
base = {
|
||||
"device_id": "dev-1", "epoch_id": "ep-1", "sent_at": 1700000000000,
|
||||
"app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
|
||||
"batches_attempted": 10, "batches_ok": 9, "batches_fail": 1,
|
||||
"retries": 1, "queue_depth": 2,
|
||||
"events": [
|
||||
{"event": "video_play", "attempted": 100, "drop_capture": 1,
|
||||
"delivered": 95, "drop_undelivered": 2},
|
||||
],
|
||||
}
|
||||
base.update(over)
|
||||
return base
|
||||
|
||||
|
||||
def test_selfstat_ingest_ok(client: TestClient) -> None:
|
||||
r = client.post("/api/v1/analytics/selfstat", json=_payload())
|
||||
assert r.status_code == 200, r.text
|
||||
body = r.json()
|
||||
assert body["ok"] is True
|
||||
assert isinstance(body["snapshot_id"], int)
|
||||
|
||||
|
||||
def test_selfstat_ingest_empty_events(client: TestClient) -> None:
|
||||
r = client.post("/api/v1/analytics/selfstat", json=_payload(events=[]))
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["ok"] is True
|
||||
|
||||
|
||||
def test_selfstat_ingest_db_error_returns_503(client: TestClient, monkeypatch) -> None:
|
||||
"""落库异常时端点回 503(计数链路稳定性守卫),不裸奔 500。"""
|
||||
def _boom(*_args, **_kwargs):
|
||||
raise RuntimeError("db gone")
|
||||
|
||||
monkeypatch.setattr("app.repositories.analytics_selfstat.record_selfstat", _boom)
|
||||
r = client.post("/api/v1/analytics/selfstat", json=_payload())
|
||||
assert r.status_code == 503, r.text
|
||||
Reference in New Issue
Block a user