Ghost
0e149c83e7
Co-authored-by: guke <guke@autohome.com.cn> Co-authored-by: 陈世睿 <2839904623@qq.com> Reviewed-on: #119 Co-authored-by: Ghost <> Co-committed-by: Ghost <>
795 lines
30 KiB
Python
795 lines
30 KiB
Python
"""admin CPS 数据访问:群/活动 CRUD + 美团订单拉单入库(对账) + 按群统计聚合。
|
|
|
|
美团 query_order 返回的金额是「元」字符串、时间是秒级时间戳,这里统一转「分」+ tz-aware。
|
|
统计在 Python 侧聚合(订单量级小、admin 低频),逻辑清晰、跨 PG/SQLite 无 SQL 方言坑。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from decimal import Decimal, InvalidOperation
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import desc, func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.admin.repositories.queries import _as_utc, offset_paginate
|
|
from app.integrations import jd_union, meituan
|
|
from app.repositories import cps_link as cps_link_repo
|
|
from app.models.cps_activity import CpsActivity
|
|
from app.models.cps_group import CpsGroup
|
|
from app.models.cps_link import CpsClick, CpsLink
|
|
from app.models.cps_order import CpsOrder
|
|
from app.models.cps_wx_user import CpsWxUser
|
|
|
|
# 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账
|
|
_INVALID_STATUS = {"4", "5"}
|
|
_SETTLED_STATUS = "6"
|
|
_JD_INVALID_CODES = {
|
|
"2", "3", "4", "5", "6", "7", "8", "9", "11", "13", "14", "19", "20", "21",
|
|
"22", "23", "25", "26", "27", "28", "29", "30", "31", "34", "35", "36",
|
|
}
|
|
_JD_UNPAID_CODES = {"15"}
|
|
|
|
# CPS 点击时序按北京时区分桶(运营看的是北京时间)
|
|
_BJ_TZ = timezone(timedelta(hours=8))
|
|
|
|
|
|
# ───────────── 单位换算 ─────────────
|
|
def _yuan_to_cents(v: object) -> int | None:
|
|
"""「元」字符串/数 → 分。None / 空 / 字面 "null" → None。"""
|
|
if v is None:
|
|
return None
|
|
s = str(v).strip()
|
|
if not s or s.lower() == "null":
|
|
return None
|
|
try:
|
|
return int((Decimal(s) * 100).to_integral_value())
|
|
except (InvalidOperation, ValueError):
|
|
return None
|
|
|
|
|
|
def _ts_to_dt(ts: object) -> datetime | None:
|
|
"""秒级时间戳 → tz-aware UTC datetime(绝对时刻,前端按北京展示)。"""
|
|
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
|
|
|
|
|
|
def _jd_dt_to_utc(value: object) -> datetime | None:
|
|
"""京东时间字符串(北京时间) → UTC aware datetime。"""
|
|
if value is None:
|
|
return None
|
|
s = str(value).strip()
|
|
if not s:
|
|
return None
|
|
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
|
|
try:
|
|
dt = datetime.strptime(s, fmt)
|
|
return dt.replace(tzinfo=_BJ_TZ).astimezone(timezone.utc)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def _text(value: object) -> str | None:
|
|
if value is None:
|
|
return None
|
|
s = str(value).strip()
|
|
return s or None
|
|
|
|
|
|
def _pick(row: dict[str, Any], *keys: str) -> Any:
|
|
for key in keys:
|
|
if key in row and row[key] is not None:
|
|
return row[key]
|
|
return None
|
|
|
|
|
|
# ───────────── 群 ─────────────
|
|
def get_group(db: Session, group_id: int) -> CpsGroup | None:
|
|
return db.get(CpsGroup, group_id)
|
|
|
|
|
|
def get_group_by_sid(db: Session, sid: str) -> CpsGroup | None:
|
|
return db.execute(select(CpsGroup).where(CpsGroup.sid == sid)).scalar_one_or_none()
|
|
|
|
|
|
def list_groups(
|
|
db: Session, *, keyword: str | None = None, status: str | None = None,
|
|
limit: int = 20, cursor: int | None = None,
|
|
) -> tuple[list[CpsGroup], int | None, int]:
|
|
stmt = select(CpsGroup)
|
|
if keyword and keyword.strip():
|
|
kw = f"%{keyword.strip()}%"
|
|
stmt = stmt.where(CpsGroup.name.ilike(kw) | CpsGroup.sid.ilike(kw))
|
|
if status:
|
|
stmt = stmt.where(CpsGroup.status == status)
|
|
return offset_paginate(db, stmt, (desc(CpsGroup.id),), limit=limit, cursor=cursor)
|
|
|
|
|
|
def create_group(
|
|
db: Session, *, name: str, platforms: list[str], sid: str | None = None,
|
|
member_count: int | None = None, remark: str | None = None, commit: bool = True,
|
|
) -> CpsGroup:
|
|
"""建群。含 meituan 才有 sid(留空自动 g<id>);纯淘宝/京东 sid=None(那俩不支持 sid)。"""
|
|
has_meituan = "meituan" in (platforms or [])
|
|
# 含美团:用填的 sid,或临时占位待回填 g<id>;不含美团:sid 恒 None
|
|
seed_sid = (sid or f"tmp{uuid4().hex[:20]}") if has_meituan else None
|
|
group = CpsGroup(
|
|
name=name,
|
|
platforms=list(platforms or []),
|
|
sid=seed_sid,
|
|
member_count=member_count,
|
|
remark=remark,
|
|
)
|
|
db.add(group)
|
|
db.flush()
|
|
if has_meituan and not sid:
|
|
group.sid = f"g{group.id}"
|
|
db.flush()
|
|
if commit:
|
|
db.commit()
|
|
db.refresh(group)
|
|
return group
|
|
|
|
|
|
def update_group(
|
|
db: Session, group: CpsGroup, *, name: str | None = None,
|
|
platforms: list[str] | None = None,
|
|
member_count: int | None = None, status: str | None = None,
|
|
remark: str | None = None, commit: bool = True,
|
|
) -> CpsGroup:
|
|
if name is not None:
|
|
group.name = name
|
|
if platforms is not None:
|
|
group.platforms = list(platforms)
|
|
if member_count is not None:
|
|
group.member_count = member_count
|
|
if status is not None:
|
|
group.status = status
|
|
if remark is not None:
|
|
group.remark = remark
|
|
if commit:
|
|
db.commit()
|
|
db.refresh(group)
|
|
else:
|
|
db.flush()
|
|
return group
|
|
|
|
|
|
def delete_group(db: Session, group: CpsGroup, *, commit: bool = True) -> None:
|
|
"""硬删群。不级联删已生成的 cps_link(已发链接继续可用);该群历史点击/订单按 sid 仍在库。"""
|
|
db.delete(group)
|
|
if commit:
|
|
db.commit()
|
|
else:
|
|
db.flush()
|
|
|
|
|
|
# ───────────── 活动 ─────────────
|
|
def get_activity(db: Session, activity_id: int) -> CpsActivity | None:
|
|
return db.get(CpsActivity, activity_id)
|
|
|
|
|
|
def list_activities(
|
|
db: Session, *, platform: str | None = None, status: str | None = None,
|
|
limit: int = 20, cursor: int | None = None,
|
|
) -> tuple[list[CpsActivity], int | None, int]:
|
|
stmt = select(CpsActivity)
|
|
if platform:
|
|
stmt = stmt.where(CpsActivity.platform == platform)
|
|
if status:
|
|
stmt = stmt.where(CpsActivity.status == status)
|
|
return offset_paginate(db, stmt, (desc(CpsActivity.id),), limit=limit, cursor=cursor)
|
|
|
|
|
|
def list_activity_images(db: Session) -> list[str]:
|
|
"""已用过的活动落地页图(distinct 非空 image_url),供新建活动复用选择。"""
|
|
rows = (
|
|
db.execute(
|
|
select(CpsActivity.image_url)
|
|
.where(CpsActivity.image_url.is_not(None))
|
|
.distinct()
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
return [r for r in rows if r]
|
|
|
|
|
|
def create_activity(
|
|
db: Session, *, name: str, platform: str = "meituan", act_id: str | None = None,
|
|
product_view_sign: str | None = None, payload: str | None = None,
|
|
image_url: str | None = None, remark: str | None = None, commit: bool = True,
|
|
) -> CpsActivity:
|
|
activity = CpsActivity(
|
|
name=name, platform=platform, act_id=act_id,
|
|
product_view_sign=product_view_sign, payload=payload,
|
|
image_url=image_url, remark=remark,
|
|
)
|
|
db.add(activity)
|
|
if commit:
|
|
db.commit()
|
|
db.refresh(activity)
|
|
else:
|
|
db.flush()
|
|
return activity
|
|
|
|
|
|
def update_activity(
|
|
db: Session, activity: CpsActivity, *, name: str | None = None,
|
|
platform: str | None = None, act_id: str | None = None,
|
|
product_view_sign: str | None = None, payload: str | None = None,
|
|
image_url: str | None = None, remark: str | None = None,
|
|
status: str | None = None, commit: bool = True,
|
|
) -> CpsActivity:
|
|
if name is not None:
|
|
activity.name = name
|
|
if platform is not None:
|
|
activity.platform = platform
|
|
if act_id is not None:
|
|
activity.act_id = act_id
|
|
if product_view_sign is not None:
|
|
activity.product_view_sign = product_view_sign
|
|
if payload is not None:
|
|
activity.payload = payload
|
|
if image_url is not None:
|
|
activity.image_url = image_url
|
|
if remark is not None:
|
|
activity.remark = remark
|
|
if status is not None:
|
|
activity.status = status
|
|
if commit:
|
|
db.commit()
|
|
db.refresh(activity)
|
|
else:
|
|
db.flush()
|
|
return activity
|
|
|
|
|
|
def delete_activity(db: Session, activity: CpsActivity, *, commit: bool = True) -> None:
|
|
"""硬删活动。不级联删 cps_link(已发链接继续可用;淘宝落地页缺活动时取图走兜底默认图)。"""
|
|
db.delete(activity)
|
|
if commit:
|
|
db.commit()
|
|
else:
|
|
db.flush()
|
|
|
|
|
|
# ───────────── 订单(对账) ─────────────
|
|
def _map_order_fields(r: dict) -> dict:
|
|
"""美团 query_order 单条 dataList → CpsOrder 字段(金额转分、时间转 datetime)。"""
|
|
pn = r.get("productName")
|
|
if pn and len(pn) > 500:
|
|
pn = pn[:500]
|
|
return {
|
|
"sid": (r.get("sid") or None),
|
|
"act_id": str(r["actId"]) if r.get("actId") is not None else None,
|
|
"biz_line": r.get("businessLine"),
|
|
"trade_type": r.get("tradeType"),
|
|
"pay_price_cents": _yuan_to_cents(r.get("payPrice")),
|
|
"commission_cents": _yuan_to_cents(r.get("profit")),
|
|
"commission_rate": str(r["commissionRate"]) if r.get("commissionRate") is not None else None,
|
|
"refund_price_cents": _yuan_to_cents(r.get("refundPrice")),
|
|
"refund_profit_cents": _yuan_to_cents(r.get("refundProfit")),
|
|
"mt_status": str(r["status"]) if r.get("status") is not None else None,
|
|
"invalid_reason": (r.get("invalidReason") or None),
|
|
"product_name": pn or None,
|
|
"pay_time": _ts_to_dt(r.get("payTime")),
|
|
"mt_update_time": _ts_to_dt(r.get("updateTime")),
|
|
"raw": r,
|
|
}
|
|
|
|
|
|
def _jd_order_key(r: dict[str, Any]) -> str | None:
|
|
row_id = _text(_pick(r, "id", "rowId", "orderRowId"))
|
|
if row_id:
|
|
return f"jd:{row_id}"
|
|
order_id = _text(_pick(r, "orderId", "parentOrderId"))
|
|
sku_id = _text(_pick(r, "skuId"))
|
|
if order_id and sku_id:
|
|
return f"jd:{order_id}:{sku_id}"
|
|
if order_id:
|
|
return f"jd:{order_id}"
|
|
return None
|
|
|
|
|
|
def _map_jd_order_fields(r: dict[str, Any]) -> dict:
|
|
"""京东 order.row.query 单条订单行 → CpsOrder 字段。"""
|
|
sku_name = _text(_pick(r, "skuName", "goodsName", "productName"))
|
|
if sku_name and len(sku_name) > 500:
|
|
sku_name = sku_name[:500]
|
|
valid_code = _text(_pick(r, "validCode", "valid_code"))
|
|
actual_fee = _yuan_to_cents(_pick(r, "actualFee", "actual_fee"))
|
|
estimate_fee = _yuan_to_cents(_pick(r, "estimateFee", "estimate_fee"))
|
|
commission = actual_fee if actual_fee not in (None, 0) else estimate_fee
|
|
order_time = _jd_dt_to_utc(_pick(r, "orderTime", "order_time"))
|
|
return {
|
|
"platform": "jd",
|
|
"external_order_id": _text(_pick(r, "orderId", "parentOrderId")),
|
|
"external_row_id": _text(_pick(r, "id", "rowId", "orderRowId")),
|
|
"sid": _text(_pick(r, "subUnionId", "sub_union_id")),
|
|
"act_id": None,
|
|
"biz_line": None,
|
|
"trade_type": None,
|
|
"pay_price_cents": _yuan_to_cents(
|
|
_pick(r, "actualCosPrice", "estimateCosPrice", "price")
|
|
),
|
|
"commission_cents": commission,
|
|
"commission_rate": _text(_pick(r, "commissionRate", "commission_rate")),
|
|
"refund_price_cents": None,
|
|
"refund_profit_cents": None,
|
|
"estimated_commission_cents": estimate_fee,
|
|
"actual_commission_cents": actual_fee,
|
|
"mt_status": None,
|
|
"jd_valid_code": valid_code,
|
|
"invalid_reason": None if _is_jd_valid_code(valid_code) else f"validCode={valid_code}",
|
|
"product_name": sku_name,
|
|
"settle_month": _text(_pick(r, "payMonth", "settleMonth", "pay_month")),
|
|
"site_id": _text(_pick(r, "siteId", "site_id")),
|
|
"position_id": _text(_pick(r, "positionId", "position_id")),
|
|
"pid": _text(_pick(r, "pid")),
|
|
"sub_union_id": _text(_pick(r, "subUnionId", "sub_union_id")),
|
|
"pay_time": order_time,
|
|
"mt_update_time": _jd_dt_to_utc(_pick(r, "modifyTime", "updateTime", "modify_time"))
|
|
or order_time,
|
|
"raw": r,
|
|
}
|
|
|
|
|
|
def _is_jd_valid_code(valid_code: str | None) -> bool:
|
|
code = str(valid_code).strip() if valid_code is not None else ""
|
|
return bool(code and code not in _JD_INVALID_CODES and code not in _JD_UNPAID_CODES)
|
|
|
|
|
|
def is_jd_order_valid(order: CpsOrder) -> bool:
|
|
return _is_jd_valid_code(order.jd_valid_code)
|
|
|
|
|
|
def effective_commission_cents(order: CpsOrder) -> int:
|
|
if order.platform == "jd":
|
|
if order.actual_commission_cents not in (None, 0):
|
|
return order.actual_commission_cents or 0
|
|
if order.estimated_commission_cents is not None:
|
|
return order.estimated_commission_cents or 0
|
|
return order.commission_cents or 0
|
|
|
|
|
|
def reconcile_orders(
|
|
db: Session, *, start_time: int, end_time: int,
|
|
query_time_type: int = 1, sid: str | None = None, max_pages: int = 200,
|
|
) -> dict:
|
|
"""调美团 query_order 分页拉单 → 按 order_id upsert。返回 {fetched, inserted, updated, pages}。
|
|
|
|
订单状态会随时间变(付款→完成→结算/退款),重复拉同一单则更新。max_pages 防异常死循环。
|
|
"""
|
|
fetched = inserted = updated = pages = 0
|
|
page = 1
|
|
while page <= max_pages:
|
|
resp = meituan.query_order(
|
|
sid=sid, start_time=start_time, end_time=end_time,
|
|
query_time_type=query_time_type, page=page, limit=100,
|
|
)
|
|
rows = ((resp.get("data") or {}).get("dataList")) or []
|
|
if not rows:
|
|
break
|
|
pages += 1
|
|
for r in rows:
|
|
order_id = str(r.get("orderId") or "").strip()
|
|
if not order_id:
|
|
continue
|
|
fetched += 1
|
|
fields = _map_order_fields(r)
|
|
fields.setdefault("platform", "meituan")
|
|
fields.setdefault("external_order_id", order_id)
|
|
fields.setdefault("external_row_id", None)
|
|
fields.setdefault("estimated_commission_cents", fields.get("commission_cents"))
|
|
fields.setdefault("actual_commission_cents", None)
|
|
existing = db.execute(
|
|
select(CpsOrder).where(CpsOrder.order_id == order_id)
|
|
).scalar_one_or_none()
|
|
if existing is None:
|
|
db.add(CpsOrder(order_id=order_id, **fields))
|
|
inserted += 1
|
|
else:
|
|
for k, v in fields.items():
|
|
setattr(existing, k, v)
|
|
updated += 1
|
|
if len(rows) < 100:
|
|
break
|
|
page += 1
|
|
db.commit()
|
|
return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
|
|
|
|
|
|
def reconcile_jd_orders(
|
|
db: Session, *, start_time: datetime, end_time: datetime,
|
|
query_time_type: int = 3, max_pages: int = 100,
|
|
) -> dict:
|
|
"""调京东 order.row.query 拉单 → 按订单行 upsert。
|
|
|
|
京东单次查询窗口最多 1 小时,这里按北京自然时间切窗并逐页拉取。
|
|
"""
|
|
fetched = inserted = updated = pages = 0
|
|
cur = start_time
|
|
while cur < end_time:
|
|
win_end = min(cur + timedelta(hours=1), end_time)
|
|
page = 1
|
|
while page <= max_pages:
|
|
resp = jd_union.query_order_rows(
|
|
start_time=cur,
|
|
end_time=win_end,
|
|
query_time_type=query_time_type,
|
|
page_index=page,
|
|
page_size=200,
|
|
)
|
|
rows = resp.get("rows") or []
|
|
has_more = bool(resp.get("has_more"))
|
|
if not rows:
|
|
break
|
|
pages += 1
|
|
for r in rows:
|
|
order_id = _jd_order_key(r)
|
|
if not order_id:
|
|
continue
|
|
fetched += 1
|
|
fields = _map_jd_order_fields(r)
|
|
existing = db.execute(
|
|
select(CpsOrder).where(CpsOrder.order_id == order_id)
|
|
).scalar_one_or_none()
|
|
if existing is None:
|
|
db.add(CpsOrder(order_id=order_id, **fields))
|
|
inserted += 1
|
|
else:
|
|
for k, v in fields.items():
|
|
setattr(existing, k, v)
|
|
updated += 1
|
|
if not has_more or len(rows) < 200:
|
|
break
|
|
page += 1
|
|
cur = win_end
|
|
db.commit()
|
|
return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
|
|
|
|
|
|
def list_orders(
|
|
db: Session, *, sid: str | None = None, mt_status: str | None = None,
|
|
limit: int = 20, cursor: int | None = None,
|
|
) -> tuple[list[CpsOrder], int | None, int]:
|
|
stmt = select(CpsOrder)
|
|
if sid:
|
|
stmt = stmt.where(CpsOrder.sid == sid)
|
|
if mt_status:
|
|
stmt = stmt.where(CpsOrder.mt_status == mt_status)
|
|
return offset_paginate(db, stmt, (desc(CpsOrder.id),), limit=limit, cursor=cursor)
|
|
|
|
|
|
# ───────────── 统计(按群聚合) ─────────────
|
|
def group_stats(
|
|
db: Session, *, date_from: datetime | None = None, date_to: datetime | None = None,
|
|
) -> list[dict]:
|
|
"""按 sid 聚合订单 + join 群信息。已建但本期无单的活跃群也列出(全 0)。
|
|
未归群的 sid(历史/其它来源)单独成行 group_id=None。按预估佣金降序。
|
|
"""
|
|
ostmt = select(CpsOrder)
|
|
if date_from is not None:
|
|
ostmt = ostmt.where(CpsOrder.pay_time >= _as_utc(date_from))
|
|
if date_to is not None:
|
|
ostmt = ostmt.where(CpsOrder.pay_time <= _as_utc(date_to))
|
|
orders_by_sid: dict[str | None, list[CpsOrder]] = {}
|
|
for o in db.execute(ostmt).scalars().all():
|
|
orders_by_sid.setdefault(o.sid, []).append(o)
|
|
|
|
clicks = cps_link_repo.click_stats_by_group(db, date_from=date_from, date_to=date_to)
|
|
groups = list(db.execute(select(CpsGroup)).scalars().all())
|
|
|
|
def _order_agg(items: list[CpsOrder]) -> dict:
|
|
valid = [o for o in items if o.mt_status not in _INVALID_STATUS]
|
|
settled = [o for o in items if o.mt_status == _SETTLED_STATUS]
|
|
canceled = [o for o in items if o.mt_status in _INVALID_STATUS]
|
|
return {
|
|
"order_count": len(valid),
|
|
"settled_count": len(settled),
|
|
"canceled_count": len(canceled),
|
|
"gmv_cents": sum(o.pay_price_cents or 0 for o in valid),
|
|
"est_commission_cents": sum(o.commission_cents or 0 for o in valid),
|
|
"settled_commission_cents": sum(o.commission_cents or 0 for o in settled),
|
|
}
|
|
|
|
# 淘宝/京东无法对账 → 对账字段全 None(前端显示 "-")
|
|
_no_recon = {
|
|
"order_count": None, "settled_count": None, "canceled_count": None,
|
|
"gmv_cents": None, "est_commission_cents": None, "settled_commission_cents": None,
|
|
}
|
|
|
|
rows: list[dict] = []
|
|
seen_sids: set[str] = set()
|
|
for g in groups:
|
|
if g.status != "active":
|
|
continue
|
|
c = clicks.get(g.id) or {}
|
|
row = {
|
|
"group_id": g.id, "sid": g.sid, "name": g.name,
|
|
"platforms": list(g.platforms or []), "member_count": g.member_count,
|
|
"click_pv": c.get("pv", 0), "click_uv": c.get("uv", 0),
|
|
"copy_pv": c.get("copy_pv", 0), "copy_uv": c.get("copy_uv", 0),
|
|
}
|
|
# 对账只对美团群(有 sid);淘宝/京东 → None
|
|
if "meituan" in (g.platforms or []) and g.sid:
|
|
row.update(_order_agg(orders_by_sid.get(g.sid, [])))
|
|
seen_sids.add(g.sid)
|
|
else:
|
|
row.update(_no_recon)
|
|
rows.append(row)
|
|
|
|
# 未归群的历史美团 sid(如 wonderableai):单列,有对账无点击
|
|
for sid, items in orders_by_sid.items():
|
|
if sid is None or sid in seen_sids:
|
|
continue
|
|
row = {
|
|
"group_id": None, "sid": sid, "name": sid, "platforms": ["meituan"],
|
|
"member_count": None, "click_pv": 0, "click_uv": 0, "copy_pv": 0, "copy_uv": 0,
|
|
}
|
|
row.update(_order_agg(items))
|
|
rows.append(row)
|
|
|
|
# 佣金降序(None 当 0),次按点击
|
|
rows.sort(key=lambda x: ((x["est_commission_cents"] or 0), x["click_pv"]), reverse=True)
|
|
return rows
|
|
|
|
|
|
def group_click_timeseries(
|
|
db: Session, *, group_id: int, granularity: str,
|
|
start: datetime, end: datetime,
|
|
) -> list[dict]:
|
|
"""该群点击时序(北京时区分桶,granularity=day|hour)。
|
|
|
|
每桶:click_pv/uv(visit)+copy_pv/uv;UV=distinct(ip,ua)。**生成连续桶(无数据补 0)**,
|
|
让折线图 x 轴按所选范围铺满、不断裂(CPS 数据稀疏,补零是关键)。
|
|
"""
|
|
rows = db.execute(
|
|
select(CpsClick.clicked_at, CpsClick.event_type, CpsClick.ip, CpsClick.ua)
|
|
.where(CpsClick.group_id == group_id)
|
|
.where(CpsClick.clicked_at >= _as_utc(start))
|
|
.where(CpsClick.clicked_at <= _as_utc(end))
|
|
).all()
|
|
|
|
agg: dict = {} # bucket_key -> {vp, vu(set), cp, cu(set)}
|
|
for clicked_at, event_type, ip, ua in rows:
|
|
# naive(本地 SQLite 读回无 tz)按 UTC 兜底,再转北京——与 queries._as_utc 约定一致,
|
|
# 否则 astimezone 会把 UTC 瞬刻当系统本地时区解释,分桶整体偏移。
|
|
ca = clicked_at if clicked_at.tzinfo else clicked_at.replace(tzinfo=timezone.utc)
|
|
bj = ca.astimezone(_BJ_TZ)
|
|
key = bj.strftime("%Y-%m-%d") if granularity == "day" else bj.hour
|
|
b = agg.setdefault(key, {"vp": 0, "vu": set(), "cp": 0, "cu": set()})
|
|
uid = f"{ip or ''}|{ua or ''}"
|
|
if event_type == "copy":
|
|
b["cp"] += 1
|
|
b["cu"].add(uid)
|
|
else:
|
|
b["vp"] += 1
|
|
b["vu"].add(uid)
|
|
|
|
def _point(label: str, b: dict | None) -> dict:
|
|
return {
|
|
"label": label,
|
|
"click_pv": b["vp"] if b else 0,
|
|
"click_uv": len(b["vu"]) if b else 0,
|
|
"copy_pv": b["cp"] if b else 0,
|
|
"copy_uv": len(b["cu"]) if b else 0,
|
|
}
|
|
|
|
points: list[dict] = []
|
|
if granularity == "day":
|
|
cur = start.astimezone(_BJ_TZ).date()
|
|
last = end.astimezone(_BJ_TZ).date()
|
|
while cur <= last:
|
|
points.append(_point(cur.strftime("%m-%d"), agg.get(cur.strftime("%Y-%m-%d"))))
|
|
cur += timedelta(days=1)
|
|
else:
|
|
for h in range(24):
|
|
points.append(_point(f"{h:02d}:00", agg.get(h)))
|
|
return points
|
|
|
|
|
|
def group_order_daily(
|
|
db: Session, *, sid: str, start: datetime, end: datetime,
|
|
) -> dict[str, dict]:
|
|
"""美团群订单按天(北京时区,key=YYYY-MM-DD)聚合。口径与 group_stats._order_agg 一致:
|
|
取消(4)/风控(5)不计有效;结算只数 status=6。退款佣金取当天全部单的 refund_profit_cents 之和。
|
|
返回 {date: {order_count, canceled_count, gmv_cents, est_commission_cents,
|
|
refund_profit_cents, settled_commission_cents}};无单的天不在 dict 里(端点侧补零)。
|
|
"""
|
|
rows = (
|
|
db.execute(
|
|
select(CpsOrder)
|
|
.where(CpsOrder.sid == sid)
|
|
.where(CpsOrder.pay_time >= _as_utc(start))
|
|
.where(CpsOrder.pay_time <= _as_utc(end))
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
by_day: dict[str, list] = {}
|
|
for o in rows:
|
|
if not o.pay_time:
|
|
continue
|
|
pt = o.pay_time if o.pay_time.tzinfo else o.pay_time.replace(tzinfo=timezone.utc)
|
|
d = pt.astimezone(_BJ_TZ).strftime("%Y-%m-%d")
|
|
by_day.setdefault(d, []).append(o)
|
|
|
|
out: dict[str, dict] = {}
|
|
for d, items in by_day.items():
|
|
valid = [o for o in items if o.mt_status not in _INVALID_STATUS]
|
|
settled = [o for o in items if o.mt_status == _SETTLED_STATUS]
|
|
canceled = [o for o in items if o.mt_status in _INVALID_STATUS]
|
|
out[d] = {
|
|
"order_count": len(valid),
|
|
"canceled_count": len(canceled),
|
|
"gmv_cents": sum(o.pay_price_cents or 0 for o in valid),
|
|
"est_commission_cents": sum(o.commission_cents or 0 for o in valid),
|
|
"refund_profit_cents": sum(o.refund_profit_cents or 0 for o in items),
|
|
"settled_commission_cents": sum(o.commission_cents or 0 for o in settled),
|
|
}
|
|
return out
|
|
|
|
|
|
def group_wx_users(db: Session, *, group_id: int, limit: int = 200) -> list[dict]:
|
|
"""该群来过的微信用户:在本群有带 openid 点击的所有用户 + 本群 visit/copy 次数 + 画像。
|
|
|
|
用「本群点击」判定归属,而非 cps_wx_user.first_group_id —— 同一用户会在多个群活动,
|
|
first_group_id 只记首次授权群,会漏掉"先在别处授权、cookie 已存、又来本群"的用户。
|
|
copy=本群点「复制口令」次数(领券意向),visit=进落地页次数。按本群首次点击倒序。
|
|
昵称/头像来自授权画像;美团/京东群只 302 跳转、无 userinfo 触发点,故多为空。
|
|
"""
|
|
rows = db.execute(
|
|
select(CpsClick.openid, CpsClick.event_type, func.count())
|
|
.where(CpsClick.group_id == group_id)
|
|
.where(CpsClick.openid.is_not(None))
|
|
.group_by(CpsClick.openid, CpsClick.event_type)
|
|
).all()
|
|
if not rows:
|
|
return []
|
|
stat: dict[str, dict] = {}
|
|
for openid, event_type, cnt in rows:
|
|
s = stat.setdefault(openid, {"visit": 0, "copy": 0})
|
|
if event_type == "copy":
|
|
s["copy"] = cnt
|
|
else:
|
|
s["visit"] = cnt
|
|
openids = list(stat.keys())
|
|
# 本群每个 openid 的首次点击时间(排序 + 展示"首次进入")
|
|
first_seen = dict(
|
|
db.execute(
|
|
select(CpsClick.openid, func.min(CpsClick.clicked_at))
|
|
.where(CpsClick.group_id == group_id)
|
|
.where(CpsClick.openid.is_not(None))
|
|
.group_by(CpsClick.openid)
|
|
).all()
|
|
)
|
|
# 关联画像(未授权 userinfo 的昵称/头像为空)
|
|
users = {
|
|
u.openid: u
|
|
for u in db.execute(select(CpsWxUser).where(CpsWxUser.openid.in_(openids))).scalars().all()
|
|
}
|
|
result = [
|
|
{
|
|
"openid": openid,
|
|
"nickname": users[openid].nickname if openid in users else None,
|
|
"headimgurl": users[openid].headimgurl if openid in users else None,
|
|
"first_seen": first_seen.get(openid),
|
|
"visit_count": stat[openid]["visit"],
|
|
"copy_count": stat[openid]["copy"],
|
|
}
|
|
for openid in openids
|
|
]
|
|
result.sort(key=lambda x: x["first_seen"], reverse=True)
|
|
return result[:limit]
|
|
|
|
|
|
def group_day_users(
|
|
db: Session, *, group_id: int, start: datetime, end: datetime, limit: int = 200,
|
|
) -> list[dict]:
|
|
"""该群某天(北京)以用户为单位的领券/点击 + 每人 visit 过的券。
|
|
|
|
时间窗为半开区间 [start, end)(end=次日 00:00),避免午夜双计。只统计 openid 非空
|
|
(可归属到人)的点击 —— 匿名点击(美团/京东 302 多为匿名)不计入。券名 = 该点击 link
|
|
对应活动名;活动被硬删则兜底 活动#{id}。copy=领券次数、visit=点击次数;coupons 仅
|
|
取 visit 事件按活动分组、按次数倒序(合计 = visit_count)。排序:领券 desc、再点击 desc。
|
|
与 group_wx_users 同风格(Python 侧聚合,跨 PG/SQLite 无方言坑)。
|
|
|
|
注:每日明细行的 click_pv/copy_pv 计全部点击(含匿名、UV 按 ip,ua);本函数只计 openid
|
|
用户,故各用户求和 <= 当天行总数,二者口径不同、不必相等。
|
|
"""
|
|
rows = db.execute(
|
|
select(CpsClick.openid, CpsClick.event_type, CpsClick.link_id)
|
|
.where(CpsClick.group_id == group_id)
|
|
.where(CpsClick.clicked_at >= _as_utc(start))
|
|
.where(CpsClick.clicked_at < _as_utc(end))
|
|
.where(CpsClick.openid.is_not(None))
|
|
).all()
|
|
if not rows:
|
|
return []
|
|
|
|
# link_id -> activity_id -> 券名(活动名)
|
|
link_ids = {r.link_id for r in rows}
|
|
link_to_act = dict(
|
|
db.execute(
|
|
select(CpsLink.id, CpsLink.activity_id).where(CpsLink.id.in_(link_ids))
|
|
).all()
|
|
)
|
|
act_ids = {aid for aid in link_to_act.values() if aid is not None}
|
|
act_name = (
|
|
dict(
|
|
db.execute(
|
|
select(CpsActivity.id, CpsActivity.name).where(CpsActivity.id.in_(act_ids))
|
|
).all()
|
|
)
|
|
if act_ids
|
|
else {}
|
|
)
|
|
|
|
def _coupon_name(link_id: int) -> str:
|
|
aid = link_to_act.get(link_id)
|
|
if aid is None:
|
|
return f"链接#{link_id}"
|
|
return act_name.get(aid) or f"活动#{aid}"
|
|
|
|
stat: dict[str, dict] = {}
|
|
for openid, event_type, link_id in rows:
|
|
s = stat.setdefault(openid, {"copy": 0, "visit": 0, "coupons": {}})
|
|
if event_type == "copy":
|
|
s["copy"] += 1
|
|
else:
|
|
s["visit"] += 1
|
|
name = _coupon_name(link_id)
|
|
s["coupons"][name] = s["coupons"].get(name, 0) + 1
|
|
|
|
openids = list(stat.keys())
|
|
users = {
|
|
u.openid: u
|
|
for u in db.execute(
|
|
select(CpsWxUser).where(CpsWxUser.openid.in_(openids))
|
|
).scalars().all()
|
|
}
|
|
|
|
result = [
|
|
{
|
|
"openid": openid,
|
|
"nickname": users[openid].nickname if openid in users else None,
|
|
"headimgurl": users[openid].headimgurl if openid in users else None,
|
|
"copy_count": s["copy"],
|
|
"visit_count": s["visit"],
|
|
"coupons": [
|
|
{"name": name, "count": cnt}
|
|
# 次数倒序;同次数按券名升序兜底,保证 PG 无 ORDER BY 行序下输出稳定
|
|
for name, cnt in sorted(
|
|
s["coupons"].items(), key=lambda kv: (-kv[1], kv[0])
|
|
)
|
|
],
|
|
}
|
|
for openid, s in stat.items()
|
|
]
|
|
result.sort(key=lambda x: (x["copy_count"], x["visit_count"]), reverse=True)
|
|
return result[:limit]
|