一、环境准备与依赖安装
在开始构建流量追踪系统之前，需要确保服务器或本地环境已经安装了Python运行环境。本系统基于Flask框架开发，轻量且易于部署，适合快速上手。
1. 基础环境检查
首先检查Python版本，建议使用Python 3.8及以上版本以确保依赖库的兼容性。在终端执行以下命令：
```bash
python3 --version
```
如果未安装Python，请使用系统对应的包管理器进行安装，例如在Ubuntu环境下执行：
```bash
sudo apt-get update
sudo apt-get install python3 python3-pip
```
2. 创建项目目录结构
为了保持项目整洁，我们需要创建一个独立的目录来存放代码和日志文件。执行以下命令：
```bash
mkdir traffic_tracker
cd traffic_tracker
mkdir logs
```
这里创建了一个名为logs的子目录，专门用于存储后续产生的流量数据文件。
3. 安装核心依赖库
本系统主要依赖Flask作为Web框架，并使用Gunicorn作为生产环境的服务器。为了管理时间，我们还会用到python-dateutil。直接执行安装命令：
```bash
pip3 install flask gunicorn python-dateutil
```
如果安装速度较慢，可以使用国内镜像源加速：
```bash
pip3 install flask gunicorn python-dateutil -i https://pypi.tuna.tsinghua.edu.cn/simple
```
二、系统配置与核心代码实现
系统的主要逻辑是：接收带有UTM参数的推广链接请求 -> 解析参数并记录到日志文件 -> 将用户重定向到目标落地页。以下是完整的代码实现。
1. 配置文件编写 (config.py)
创建一个config.py文件，将系统可配置项抽离出来，方便后续维护。请直接复制以下内容：
```python
config.py
目标落地页地址，所有推广流量最终都会跳转到这个页面
TARGET_URL = "https://www.example.com/landing-page"
日志文件存储路径
LOG_FILE_PATH = "./logs/traffic_data.csv"
日志文件表头，第一次运行时会自动写入
CSV_HEADERS = ["timestamp", "ip_address", "utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content", "user_agent"]
服务监听端口
PORT = 5000
```
注意：请将TARGET_URL修改为您实际业务的落地页地址。
2. 核心逻辑实现 (app.py)
在同级目录下创建app.py，这是系统的入口文件。代码包含参数解析、CSV文件写入和重定向逻辑。
```python
app.py
import os
import csv
import time
from datetime import datetime
from flask import Flask, request, redirect
from config import TARGET_URL, LOG_FILE_PATH, CSV_HEADERS
app = Flask(__name__)
def ensure_log_file_exists():
"""检查并初始化日志文件，确保表头存在"""
if not os.path.exists(LOG_FILE_PATH):
with open(LOG_FILE_PATH, mode='w', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(CSV_HEADERS)
def log_traffic_data(data):
"""将流量数据追加写入CSV文件"""
try:
'a'模式表示追加写入，newline=''防止在Windows下出现空行
with open(LOG_FILE_PATH, mode='a', encoding='utf-8', newline='') as f:
writer = csv.writer(f)
writer.writerow(data)
except Exception as e:
print(f"日志写入失败: {e}")
@app.route('/')
def track_and_redirect():
1. 获取客户端IP
ip_address = request.headers.get('X-Forwarded-For', request.remote_addr)
2. 获取UTM参数，如果不存在则设为 'unknown'
utm_source = request.args.get('utm_source', 'unknown')
utm_medium = request.args.get('utm_medium', 'unknown')
utm_campaign = request.args.get('utm_campaign', 'unknown')
utm_term = request.args.get('utm_term', '')
utm_content = request.args.get('utm_content', '')
3. 获取User-Agent
user_agent = request.headers.get('User-Agent', '')
4. 获取当前时间戳
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
5. 组装数据行
row_data = [
timestamp,
ip_address,
utm_source,
utm_medium,
utm_campaign,
utm_term,
utm_content,
user_agent
]
6. 写入日志
ensure_log_file_exists()
log_traffic_data(row_data)
7. 执行302重定向到目标页面
return redirect(TARGET_URL)
if __name__ == '__main__':
确保日志目录存在
os.makedirs(os.path.dirname(LOG_FILE_PATH), exist_ok=True)
ensure_log_file_exists()
app.run(host='0.0.0.0', port=5000)
```
三、服务启动与测试验证
代码编写完成后，我们需要启动服务并进行模拟测试，确保数据能够被正确记录。
1. 本地测试启动
在项目根目录下执行以下命令启动服务：
```bash
python3 app.py
```
终端显示Running on http://0.0.0.0:5000即表示启动成功。
2. 模拟流量请求测试
打开新的终端窗口，使用curl命令模拟带有推广参数的访问请求：
```bash
curl "http://127.0.0.1:5000/?utm_source=google&utm_medium=cpc&utm_campaign=spring_sale"
```
执行后，检查logs/traffic_data.csv文件内容。可以使用cat命令查看：
```bash
cat logs/traffic_data.csv
```
您应该能看到一行包含刚才传入的时间戳、IP、google、cpc等信息的记录。如果看到HTTP 302响应跳转到了配置的TARGET_URL，说明逻辑正常。
四、生产环境部署配置
在本地测试通过后，如果需要部署到服务器供外部访问，建议使用Gunicorn作为WSGI服务器，它能提供更高的并发性能和稳定性。
1. 使用Gunicorn启动
直接使用以下命令启动，-w指定工作进程数（建议设置为CPU核心数的2-4倍），-b指定绑定地址：
```bash
gunicorn -w 4 -b 0.0.0.0:5000 app:app
```
2. 配置后台运行与自动重启
为了保证服务关闭后依然运行，并实现崩溃自动重启，建议结合supervisor进行管理。首先安装supervisor：
```bash
sudo apt-get install supervisor
```
创建配置文件/etc/supervisor/conf.d/traffic_tracker.conf：
```ini
[program:traffic_tracker]
command=/usr/local/bin/gunicorn -w 4 -b 0.0.0.0:5000 app:app
directory=/root/traffic_tracker
user=root
autostart=true
autorestart=true
stderr_logfile=/var/log/traffic_tracker.err.log
stdout_logfile=/var/log/traffic_tracker.out.log
```
注意：请将directory修改为您实际的项目路径。保存后执行以下命令生效：
```bash
sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start traffic_tracker
```
五、数据清洗与分析脚本
随着流量数据的积累，CSV文件会变大。为了方便查看各渠道的推广效果，我们可以编写一个简单的Python脚本进行数据统计。
创建analyze.py：
```python
analyze.py
import csv
from collections import Counter
LOG_FILE = "./logs/traffic_data.csv"
def analyze_sources():
data = []
if not os.path.exists(LOG_FILE):
print("日志文件不存在")
return
with open(LOG_FILE, mode='r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
data.append(row)
if not data:
print("暂无数据")
return
统计 utm_source 来源分布
sources = [row['utm_source'] for row in data]
source_counts = Counter(sources)
print(f"总访问量: {len(data)}")
print("\n 来源渠道统计 ")
for source, count in source_counts.most_common():
print(f"{source}: {count} 次 ({round(count/len(data)100, 2)}%)")
统计 utm_campaign 活动分布
campaigns = [row['utm_campaign'] for row in data]
campaign_counts = Counter(campaigns)
print("\n 推广活动统计 ")
for campaign, count in campaign_counts.most_common():
print(f"{campaign}: {count} 次")
if __name__ == "__main__":
import os
analyze_sources()
```
执行分析脚本即可快速查看流量汇总：
```bash
python3 analyze.py
```
通过以上步骤，您已经完成了一套从流量采集、存储到分析的完整闭环系统。在实际推广中，只需将推广链接中的URL替换为该服务器的地址，并附带相应的UTM参数即可。例如：http://your-server-ip:5000/?utm_source=wechat&utm_medium=post。
