from urllib.parse import unquote from backend.access_log import _log_request from backend.visit_stats import record_activity_report def _sparse_page_activity_log_cum(cum: int) -> bool: """无服务端状态:累计秒为 2、10、20(前 20s 内与前端档位一致)或 40、80、160…(40·2^k)时打访问日志。""" if cum in (2, 10, 20): return True if cum < 40 or cum % 40: return False q = cum // 40 return q > 0 and (q & (q - 1)) == 0 def client_activity_report(activity_body=None): d = activity_body if isinstance(activity_body, dict) else {} p = str(d.get("page_path") or "")[:512].strip() try: cum = int(d.get("total_active_sec")) dlt = int(d.get("delta_active_sec")) if not p or cum < 1 or dlt < 0: return {"ok": True} except (TypeError, ValueError): return {"ok": True} path_only = p.split("?", 1)[0].split("#", 1)[0].strip() page_key = path_only.rstrip("/").split("/")[-1] or path_only if not page_key: return {"ok": True} if "?" in p: qs = p.split("?", 1)[1].split("#", 1)[0] log_path = f"{page_key}?{unquote(qs)}" if qs else page_key else: log_path = page_key raw_os = d.get("client_os") client_os = str(raw_os).strip() if raw_os is not None else None record_activity_report(page_key, dlt, cum, client_os) if _sparse_page_activity_log_cum(cum): _log_request( "📄 页面活跃", f"path={log_path!r} total_sec={cum} delta_sec={dlt}", ) return {"ok": True}