商城公域引流：从零到一搭建自动化引流系统

一、系统架构与核心组件

本系统基于开源技术栈构建，包含数据采集、内容生成、渠道分发三个核心模块。系统架构图如下：

所需环境与工具：

• 服务器：Linux (Ubuntu 20.04 LTS)，2核4G配置
• 编程语言：Python 3.8+
• 数据库：MySQL 8.0
• 任务调度：Celery + Redis

二、环境配置与依赖安装

2.1 服务器环境初始化

登录服务器后执行以下命令：

``` sudo apt update sudo apt install python3-pip python3-venv git nginx mysql-server -y sudo systemctl start mysql sudo systemctl enable mysql ```

创建项目目录并设置虚拟环境：

``` mkdir -p /opt/traffic_system cd /opt/traffic_system python3 -m venv venv source venv/bin/activate ```

2.2 Python依赖包安装

创建requirements.txt文件：

``` requests==2.28.1 selenium==4.8.0 beautifulsoup4==4.11.1 pymysql==1.0.2 celery==5.2.7 redis==4.5.1 schedule==1.2.0 pandas==1.5.3 openpyxl==3.0.10 ```

安装依赖：

``` pip install -r requirements.txt ```

三、数据库设计与配置

3.1 数据库初始化

登录MySQL创建数据库：

``` mysql -u root -p CREATE DATABASE traffic_system CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; CREATE USER 'traffic_user'@'localhost' IDENTIFIED BY 'YourPassword123!'; GRANT ALL PRIVILEGES ON traffic_system. TO 'traffic_user'@'localhost'; FLUSH PRIVILEGES; EXIT; ```

3.2 数据表结构

创建database.py文件：

``` import pymysql def create_tables(): conn = pymysql.connect( host='localhost', user='traffic_user', password='YourPassword123!', database='traffic_system', charset='utf8mb4' ) sql_statements = [ """ CREATE TABLE IF NOT EXISTS content_sources ( id INT AUTO_INCREMENT PRIMARY KEY, platform_name VARCHAR(50) NOT NULL, source_url VARCHAR(500) NOT NULL, category VARCHAR(50), crawl_interval INT DEFAULT 3600, is_active BOOLEAN DEFAULT TRUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """, """ CREATE TABLE IF NOT EXISTS crawled_content ( id INT AUTO_INCREMENT PRIMARY KEY, source_id INT, title VARCHAR(200) NOT NULL, content TEXT, original_url VARCHAR(500), publish_time DATETIME, keywords TEXT, processed_content TEXT, status ENUM('raw', 'processed', 'published') DEFAULT 'raw', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (source_id) REFERENCES content_sources(id) ) """, """ CREATE TABLE IF NOT EXISTS publishing_log ( id INT AUTO_INCREMENT PRIMARY KEY, content_id INT, target_platform VARCHAR(50) NOT NULL, publish_time DATETIME NOT NULL, publish_url VARCHAR(500), status ENUM('success', 'failed') NOT NULL, error_message TEXT, FOREIGN KEY (content_id) REFERENCES crawled_content(id) ) """ ] with conn.cursor() as cursor: for sql in sql_statements: cursor.execute(sql) conn.commit() conn.close() if __name__ == "__main__": create_tables() ```

四、数据采集模块实现

4.1 基础爬虫配置

创建crawler.py文件：

``` import requests from bs4 import BeautifulSoup import pymysql from datetime import datetime import time class ContentCrawler: def __init__(self): self.db_conn = pymysql.connect( host='localhost', user='traffic_user', password='YourPassword123!', database='traffic_system', charset='utf8mb4' ) self.headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' } def crawl_zhihu_hot(self): """采集知乎热榜""" url = "https://www.zhihu.com/billboard" try: response = requests.get(url, headers=self.headers, timeout=10) soup = BeautifulSoup(response.text, 'html.parser') hot_items = soup.select('.HotList-item') for item in hot_items[:20]: 取前20条 title_elem = item.select_one('.HotList-itemTitle') if title_elem: title = title_elem.text.strip() 保存到数据库 with self.db_conn.cursor() as cursor: sql = """ INSERT INTO crawled_content (source_id, title, original_url, publish_time, status) VALUES (%s, %s, %s, %s, %s) """ cursor.execute(sql, ( 1, 知乎source_id title, "https://www.zhihu.com", datetime.now(), 'raw' )) self.db_conn.commit() except Exception as e: print(f"知乎爬取失败: {str(e)}") def crawl_weibo_hot(self): """采集微博热搜""" url = "https://s.weibo.com/top/summary" try: response = requests.get(url, headers=self.headers, timeout=10) soup = BeautifulSoup(response.text, 'html.parser') hot_items = soup.select('.td-02 a') for item in hot_items[:20]: title = item.text.strip() if title and not title.startswith(''): with self.db_conn.cursor() as cursor: sql = """ INSERT INTO crawled_content (source_id, title, original_url, publish_time, status) VALUES (%s, %s, %s, %s, %s) """ cursor.execute(sql, ( 2, 微博source_id title, "https://s.weibo.com", datetime.now(), 'raw' )) self.db_conn.commit() except Exception as e: print(f"微博爬取失败: {str(e)}") def run_all(self): """执行所有爬虫任务""" self.crawl_zhihu_hot() time.sleep(2) 避免请求过快 self.crawl_weibo_hot() self.db_conn.close() ```

4.2 内容源配置

创建sources.json配置文件：

``` { "sources": [ { "id": 1, "platform_name": "知乎", "source_url": "https://www.zhihu.com/billboard", "category": "热点", "crawl_interval": 1800, "is_active": true }, { "id": 2, "platform_name": "微博", "source_url": "https://s.weibo.com/top/summary", "category": "热搜", "crawl_interval": 1800, "is_active": true }, { "id": 3, "platform_name": "豆瓣小组", "source_url": "https://www.douban.com/group/explore", "category": "讨论", "crawl_interval": 3600, "is_active": true } ] } ```

五、内容处理与优化

5.1 内容清洗与格式化

创建processor.py文件：

``` import re import jieba import jieba.analyse from datetime import datetime class ContentProcessor: def __init__(self): 加载停用词 with open('stopwords.txt', 'r', encoding='utf-8') as f: self.stopwords = set([line.strip() for line in f]) 加载行业关键词 self.industry_keywords = ['电商', '商城', '购物', '促销', '优惠', '折扣'] def clean_content(self, text): """清洗文本内容""" if not text: return "" 移除HTML标签 text = re.sub(r'<[^>]+>', '', text) 移除特殊字符 text = re.sub(r'[^\w\u4e00-\u9fff\s]', '', text) 移除多余空格 text = re.sub(r'\s+', ' ', text).strip() return text def extract_keywords(self, text, top_k=5): """提取关键词""" keywords = jieba.analyse.extract_tags( text, topK=top_k, allowPOS=('n', 'vn', 'v', 'a') ) 过滤停用词 keywords = [kw for kw in keywords if kw not in self.stopwords] return keywords def generate_related_content(self, title, keywords): """生成相关内容""" 基础模板 templates = [ f"最近{title}很火，我们商城正好有相关商品在促销，点击查看详情→", f"关于{title}，我们整理了相关的优质商品，现在购买享专属优惠→", f"{keywords[0]}相关的好物推荐，限时特价中，不要错过→" ] import random template = random.choice(templates) 添加商城链接（需要替换为实际链接） final_content = f"{template} https://your-mall.com/product?keywords={keywords[0]}" return final_content def process_batch(self): """批量处理未处理的内容""" import pymysql conn = pymysql.connect( host='localhost', user='traffic_user', password='YourPassword123!', database='traffic_system', charset='utf8mb4' ) with conn.cursor() as cursor: 获取未处理的内容 cursor.execute(""" SELECT id, title, content FROM crawled_content WHERE status = 'raw' LIMIT 50 """) rows = cursor.fetchall() for row in rows: content_id, title, content = row 清洗内容 cleaned_content = self.clean_content(content or title) 提取关键词 keywords = self.extract_keywords(cleaned_content) 生成相关内容 related_content = self.generate_related_content(title, keywords) 更新数据库 update_sql = """ UPDATE crawled_content SET processed_content = %s, keywords = %s, status = 'processed' WHERE id = %s """ cursor.execute(update_sql, ( related_content, ','.join(keywords), content_id )) conn.commit() conn.close() ```

六、渠道分发自动化

6.1 微博自动发布

创建weibo_publisher.py：

``` import requests import json from datetime import datetime class WeiboPublisher: def __init__(self): 微博API配置（需要先申请开发者权限） self.app_key = "YOUR_APP_KEY" self.app_secret = "YOUR_APP_SECRET" self.access_token = "YOUR_ACCESS_TOKEN" self.api_url = "https://api.weibo.com/2/statuses/share.json" def publish(self, content, image_path=None): """发布微博""" payload = { 'access_token': self.access_token, 'status': content } files = {} if image_path: files['pic'] = open(image_path, 'rb') try: response = requests.post( self.api_url, data=payload, files=files if files else None ) result = response.json() if 'id' in result: return { 'success': True, 'url': f"https://weibo.com/{result['id']}", 'publish_time': datetime.now() } else: return { 'success': False, 'error': result.get('error', '未知错误') } except Exception as e: return { 'success': False, 'error': str(e) } def publish_from_database(self): """从数据库获取内容并发布""" import pymysql conn = pymysql.connect( host='localhost', user='traffic_user', password='YourPassword123!', database='traffic_system', charset='utf8mb4' ) with conn.cursor() as cursor: 获取待发布的内容 cursor.execute(""" SELECT id, processed_content FROM crawled_content WHERE status = 'processed' ORDER BY publish_time DESC LIMIT 1 """) row = cursor.fetchone() if row: content_id, content = row 发布到微博 result = self.publish(content) 记录发布日志 log_sql = """ INSERT INTO publishing_log (content_id, target_platform, publish_time, publish_url, status, error_message) VALUES (%s, %s, %s, %s, %s, %s) """ if result['success']: cursor.execute(log_sql, ( content_id, 'weibo', result['publish_time'], result['url'], 'success', None )) 更新内容状态 update_sql = """ UPDATE crawled_content SET status = 'published' WHERE id = %s """ cursor.execute(update_sql, (content_id,)) else: cursor.execute(log_sql, ( content_id, 'weibo', datetime.now(), None, 'failed', result['error'] )) conn.commit() conn.close() ```

6.2 定时任务配置

创建scheduler.py：

``` import schedule import time from crawler import ContentCrawler from processor import ContentProcessor from weibo_publisher import WeiboPublisher def crawl_job(): print("开始执行爬虫任务...") crawler = ContentCrawler() crawler.run_all() print("爬虫任务完成") def process_job(): print("开始处理内容...") processor = ContentProcessor() processor.process_batch() print("内容处理完成") def publish_job(): print("开始发布内容...") publisher = WeiboPublisher() publisher.publish_from_database() print("内容发布完成") 设置定时任务 schedule.every(30).minutes.do(crawl_job) 每30分钟采集一次 schedule.every(45).minutes.do(process_job) 每45分钟处理一次 schedule.every(60).minutes.do(publish_job) 每60分钟发布一次 print("定时任务系统已启动...") while True: schedule.run_pending() time.sleep(60) ```

七、监控与优化

7.1 效果监控脚本

创建monitor.py：

``` import pymysql from datetime import datetime, timedelta class PerformanceMonitor: def __init__(self): self.conn = pymysql.connect( host='localhost', user='traffic_user', password='YourPassword123!', database='traffic_system', charset='utf8mb4' ) def get_daily_stats(self): """获取每日统计数据""" with self.conn.cursor() as cursor: 今日数据 today = datetime.now().date() cursor.execute(""" SELECT COUNT() as total_published, SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) as success_count, SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) as failed_count FROM publishing_log WHERE DATE(publish_time) = %s """, (today,)) stats = cursor.fetchone() return { 'date': today.strftime('%Y-%m-%d'), 'total_published': stats[0] or 0, 'success_count': stats[1] or 0, 'failed_count': stats[2] or 0, 'success_rate': (stats[1] / stats[0] 100) if stats[0] > 0 else 0 } def get_top_keywords(self, days=7): """获取热门关键词""" with self.conn.cursor() as cursor: start_date = datetime.now() - timedelta(days=days) cursor.execute(""" SELECT keywords, COUNT() as count FROM crawled_content WHERE created_at >= %s AND keywords IS NOT NULL GROUP BY keywords ORDER BY count DESC LIMIT 10 """, (start