流量统计脚本怎么做？

wen 实用脚本 2026-06-06 45

流量统计脚本怎么做？从零到一，手把手教你搭建高转化追踪系统

目录导读

流量统计脚本的核心价值：为什么你需要自己写而不是直接用现成工具？
底层逻辑拆解：用户访问数据的采集与存储原理
代码实战（附完整示例）：Python + JavaScript 双端实现方案
常见问题与避坑指南：数据偏差、跨域、隐私合规怎么解？
QA问答：开发者最困惑的5个真实问题

流量统计脚本的核心价值

很多人认为“流量统计=装个Google Analytics或百度统计”，但在实际场景中，自建脚本能解决三大痛点：

流量统计脚本怎么做？

数据主权：第三方工具会记录用户IP、设备指纹，甚至与广告平台共享（例如Google Analytics默认与Google Ads联动）,自建脚本可彻底脱敏。
定制化维度：比如统计“用户点击某个按钮后停留3秒才算有效流量”，或记录“页面滚动深度>70%”的访客——第三方工具无法实现这类业务逻辑。
实时性与成本：自建脚本每天处理10万次请求只需1台轻量服务器,而同等规模使用付费SaaS月费可能超500元。

底层逻辑拆解：用户访问数据的生命周期

一个完整流量统计脚本包含三个环节：

1 数据采集（前端埋点）

通过JavaScript在用户浏览器中捕获：

// 基础采集项
const metrics = {
  page_url: window.location.href,    // 当前页面完整地址
  referrer: document.referrer,       // 从哪个链接跳转而来
  user_agent: navigator.userAgent,   // 浏览器型号、操作系统
  screen_resolution: `${screen.width}x${screen.height}`, // 屏幕分辨率
  timestamp: Date.now(),             // 访问时间戳
  ua_hash: simpleHash(navigator.userAgent) // 用户标识（非cookie）
};

2 数据传输（HTTP请求）

通过Image Beacon或Fetch API异步发送数据（避免阻塞页面加载）：

// 使用1x1像素GIF发送（兼容性最好）
new Image().src = `https://your-server.com/track?${new URLSearchParams(metrics)}`;
// 或使用sendBeacon（页面卸载前仍能发送）
navigator.sendBeacon('/track', JSON.stringify(metrics));

3 数据存储（服务端处理）

服务端（以Python Flask为例）接收并写入数据库：

from flask import Flask, request
import sqlite3
app = Flask(__name__)
@app.route('/track')
def track():
    page_url = request.args.get('page_url')
    user_agent = request.args.get('user_agent')
    timestamp = request.args.get('timestamp')
    # 写入数据库（生产环境建议使用PostgreSQL/ClickHouse）
    conn = sqlite3.connect('stats.db')
    c = conn.cursor()
    c.execute("INSERT INTO visits VALUES (?, ?, ?)", 
              (page_url, user_agent, timestamp))
    conn.commit()
    conn.close()
    return 'ok'

代码实战：一套可直接运行的流量统计脚本

1 前端脚本（完整版，支持防刷）

// 文件名: tracker.js
(function() {
  'use strict';
  // 配置项
  const CONFIG = {
    endpoint: 'https://your-server.com/track', // 替换为实际服务器地址
    sampleRate: 1, // 采样率，1表示100%
    excludedPaths: ['/admin', '/static'] // 忽略的路径
  };
  // 数据采集
  function getMetrics() {
    return {
      url: window.location.href,
      ref: document.referrer || '',
      res: `${screen.width}x${screen.height}`,
      ua: navigator.userAgent,
      ts: Date.now(),
      vid: getVisitorId() // 基于UA+屏幕计算（不存cookie）
    };
  }
  // 发送数据
  function sendData() {
    if (Math.random() > CONFIG.sampleRate) return;
    const currentPath = window.location.pathname;
    if (CONFIG.excludedPaths.some(p => currentPath.startsWith(p))) return;
    const payload = getMetrics();
    try {
      const img = new Image();
      img.src = CONFIG.endpoint + '?' + new URLSearchParams(payload);
    } catch(e) {
      // fallback到fetch
      fetch(CONFIG.endpoint, {method: 'POST', body: JSON.stringify(payload)});
    }
  }
  // 页面加载时触发
  if (document.readyState === 'complete') {
    sendData();
  } else {
    window.addEventListener('load', sendData);
  }
})();

2 服务端脚本（Python + SQLite，支持高并发）

# 文件名: server.py
from flask import Flask, request, Response
import sqlite3
import datetime
app = Flask(__name__)
# 初始化数据库
def init_db():
    conn = sqlite3.connect('website_stats.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS visits
                 (id INTEGER PRIMARY KEY,
                  url TEXT,
                  referrer TEXT,
                  screen_res TEXT,
                  user_agent TEXT,
                  visit_time TIMESTAMP,
                  visitor_id TEXT)''')
    conn.commit()
    conn.close()
@app.route('/track')
def track_visit():
    # 获取参数
    url = request.args.get('url', '')
    ref = request.args.get('ref', '')
    res = request.args.get('res', '')
    ua = request.args.get('ua', '')
    ts = request.args.get('ts', int(datetime.datetime.now().timestamp()))
    vid = request.args.get('vid', '')
    # 数据清洗（示例：过滤明显异常UA）
    if 'bot' in ua.lower() or 'crawler' in ua.lower():
        return Response('ignored', status=200)
    # 入库
    try:
        conn = sqlite3.connect('website_stats.db')
        c = conn.cursor()
        c.execute(
            "INSERT INTO visits (url, referrer, screen_res, user_agent, visit_time, visitor_id) VALUES (?,?,?,?,?,?)",
            (url, ref, res, ua, datetime.datetime.fromtimestamp(int(ts)), vid)
        )
        conn.commit()
    except Exception as e:
        print(f"DB error: {e}")
        return Response('error', status=500)
    finally:
        conn.close()
    # 返回1x1像素透明GIF（兼容广告拦截器）
    pixel = b'\x47\x49\x46\x38\x39\x61\x01\x00\x01\x00\x80\x00\x00\xff\xff\xff\x00\x00\x00\x21\xf9\x04\x00\x00\x00\x00\x00\x2c\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x02\x44\x01\x00\x3b'
    return Response(pixel, mimetype='image/gif')
if __name__ == '__main__':
    init_db()
    app.run(host='0.0.0.0', port=5000)

3 部署关键点

跨域问题：只需在服务端添加Access-Control-Allow-Origin: *（因为前端使用Image对象发送,天然不受跨域限制）
HTTPS强制：使用Let's Encrypt免费证书，避免浏览器拦截混合内容
日志压缩：启用Nginx Gzip，传输前对JSON体进行压缩

常见问题与避坑指南

1 数据缺失问题

现象：统计数量比实际PV少20% 原因：移动端浏览器在页面关闭时可能放弃请求 解决方案：

改用navigator.sendBeacon（在页面卸载时仍能100%发送）
或增加“心跳机制”：每30秒发送一次当前页面停留数据

2 广告拦截器屏蔽

当你的/track路径包含“analytics”“tracking”等关键词时，uBlock Origin等扩展会直接拦截。 应对策略：

将路径伪装为静态资源，例如/images/stat.gif?data=xxx
使用随机路径：每24小时自动更换一次端点（需服务端兼容）

3 GDPR与隐私合规

如果你面向欧盟用户,必须注意：

禁止收集未脱敏的IP地址（需用hash处理，如SHA256+盐值）
在页面显式告知：“本网站使用自建统计系统，仅记录页面访问行为,不存储个人身份信息”

QA问答：开发者最困惑的5个真实问题

Q1：自建脚本和百度统计/GA并存会冲突吗？ 不会，自建脚本与第三方工具是独立运行的两套系统，但需注意：不要在自建脚本中再次调用其他统计代码的变量（如_gaq）,否则可能引发性能问题。

Q2：如何防止恶意刷量（比如每小时10万次请求）？

同一IP+同一visitor_id的请求，每秒只记录一次（用Redis做限流）
检测User-Agent是否包含“node-fetch”“python-requests”等关键词
对短时间内大量来自同一个referrer的请求进行降权

Q3：我的网站是静态页（GitHub Pages），能部署自建脚本吗？ 可以，你只需要一个能运行Python/Node.js的服务器（比如AWS免费层、阿里云轻量云）作为服务端，前端JS直接托管在静态服务器上，通过fetch或Image发送数据到该服务端即可。

Q4：统计脚本影响页面加载速度怎么办？

使用async加载：<script async src="tracker.js"></script>
脚本体积必须小于2KB（Demo代码已满足）
服务端响应时间应<50ms（SQLite写入可以进一步优化为批量插入）

Q5：数据库写入速度跟不上高并发怎么办？

改用ClickHouse等列式数据库（写入速度是MySQL的10倍）
或使用消息队列：将数据先写入Redis列表，后台进程批量写入数据库

示例如下：

# 使用Redis缓存写入
r.lpush('stats_queue', json.dumps(payload))
# 每5秒一次性取100条写入DB
while True:
  data = [json.loads(item) for item in r.lrange('stats_queue', 0, 99)]
  r.ltrim('stats_queue', 100, -1)
  cursor.executemany("INSERT INTO visits VALUES (...)", data)
  time.sleep(5)