TikTokDownloader隐藏玩法：用Web API模式打造自动化数据看板（Python实战）
·
TikTokDownloader隐藏玩法：用Web API模式打造自动化数据看板（Python实战）
如果你还在用TikTokDownloader的图形界面一个个手动下载视频，那可能只发挥了它10%的潜力。这个开源工具真正的宝藏，藏在那个看似不起眼的“Web API模式”里。想象一下，每天早晨打开电脑，一份最新的热门视频趋势报告已经躺在你的邮箱里；或者你的数据看板实时刷新着竞品账号的互动数据变化——这些都不是幻想，而是通过TikTokDownloader的API接口就能实现的自动化场景。
对于数据分析师和开发者来说，TikTokDownloader提供的Web API接口就像一把瑞士军刀，它把复杂的数据采集过程封装成了简单的HTTP请求。这意味着你可以用Python脚本、定时任务、甚至与其他数据分析工具无缝集成，构建出完全定制化的数据管道。今天，我们就来深入挖掘这个工具的API潜力，看看如何用它打造一个真正实用的自动化数据看板。
1. 从零搭建Web API环境：不只是启动服务那么简单
很多人以为启动Web API模式就是运行一个命令，然后访问localhost:5555就完事了。但实际上，要让API服务稳定、安全地运行在自动化环境中，需要考虑的细节要多得多。
1.1 服务部署的几种姿势
首先，你需要决定如何部署这个服务。最简单的方式当然是直接运行：
python main.py
然后在交互菜单中选择“Web API模式”。但这种方式不适合自动化场景，因为你需要手动操作。更好的方法是使用配置文件中的run_command参数：
{
"run_command": "8",
"root": "/data/tiktok_downloader",
"storage_format": "sql"
}
这样配置后，程序启动时会自动进入Web API模式。但还有更优雅的方案——使用Docker容器化部署。我在实际项目中经常使用下面的Docker Compose配置：
version: '3.8'
services:
tiktok-downloader:
image: joeanamier/tiktok-downloader:latest
container_name: tiktok-downloader-api
ports:
- "5555:5555"
volumes:
- ./volume:/app/Volume
- ./config:/app/config
environment:
- TZ=Asia/Shanghai
restart: unless-stopped
command: ["--api-only"]
提示：使用Docker部署时，记得将本地的
settings.json配置文件挂载到容器内，否则每次重启容器配置都会丢失。
1.2 安全配置与访问控制
公开的API服务如果没有适当的安全措施，可能会带来风险。TikTokDownloader的Web API默认没有身份验证，这在本地测试时没问题，但如果部署在服务器上就需要额外处理。
我通常会在前面加一层Nginx反向代理，并配置基本的HTTP认证：
server {
listen 80;
server_name api.yourdomain.com;
location / {
auth_basic "Restricted Access";
auth_basic_user_file /etc/nginx/.htpasswd;
proxy_pass http://localhost:5555;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
}
对于更复杂的场景，你还可以在应用中添加API密钥验证。虽然TikTokDownloader本身不支持，但可以通过中间件的方式实现。下面是一个简单的Flask中间件示例：
from flask import Flask, request, jsonify
from functools import wraps
app = Flask(__name__)
API_KEYS = {
"your_api_key_here": "data_team"
}
def require_api_key(f):
@wraps(f)
def decorated_function(*args, **kwargs):
api_key = request.headers.get('X-API-Key')
if api_key not in API_KEYS:
return jsonify({"error": "Invalid API key"}), 403
return f(*args, **kwargs)
return decorated_function
@app.route('/douyin/comment', methods=['POST'])
@require_api_key
def proxy_to_tiktokdownloader():
# 这里转发请求到本地的TikTokDownloader API
pass
1.3 性能优化与监控
当API服务开始处理大量请求时，性能监控就变得很重要。我习惯使用prometheus-client来暴露一些关键指标：
from prometheus_client import start_http_server, Counter, Histogram
import time
# 定义指标
REQUEST_COUNT = Counter('tiktok_api_requests_total', 'Total API requests')
REQUEST_LATENCY = Histogram('tiktok_api_request_latency_seconds', 'API request latency')
def track_request(func):
def wrapper(*args, **kwargs):
REQUEST_COUNT.inc()
start_time = time.time()
result = func(*args, **kwargs)
latency = time.time() - start_time
REQUEST_LATENCY.observe(latency)
return result
return wrapper
把这些装饰器应用到你的API路由上，就能实时监控请求量和响应时间。当发现性能瓶颈时，可以考虑调整TikTokDownloader的配置参数，比如增加chunk大小或调整max_retry次数。
2. 热榜数据自动化采集：构建你的内容趋势雷达
热榜数据是了解平台内容风向最直接的窗口。TikTokDownloader的/douyin/hot接口可以获取抖音的热榜数据，但如何让这些数据真正产生价值，需要一些技巧。
2.1 定时采集与数据存储
首先，我们创建一个定时采集脚本。我推荐使用schedule库配合sqlite3进行数据存储：
import schedule
import time
import sqlite3
from datetime import datetime
import httpx
def create_hot_table():
"""创建热榜数据表"""
conn = sqlite3.connect('tiktok_data.db')
c = conn.cursor()
c.execute('''
CREATE TABLE IF NOT EXISTS hotlist (
id INTEGER PRIMARY KEY AUTOINCREMENT,
rank INTEGER,
title TEXT,
hot_value INTEGER,
word_cover TEXT,
collect_time TIMESTAMP,
hot_type TEXT
)
''')
conn.commit()
conn.close()
def fetch_hotlist_data(hot_type='hot'):
"""获取热榜数据并存入数据库"""
try:
# 调用TikTokDownloader API
api_url = "http://localhost:5555/douyin/hot"
data = {"type": hot_type, "pages": 1}
response = httpx.post(api_url, json=data, timeout=30)
hot_data = response.json()
if hot_data.get("status") == "success":
conn = sqlite3.connect('tiktok_data.db')
c = conn.cursor()
collect_time = datetime.now()
for item in hot_data.get("data", []):
c.execute('''
INSERT INTO hotlist (rank, title, hot_value, word_cover, collect_time, hot_type)
VALUES (?, ?, ?, ?, ?, ?)
''', (
item.get("rank"),
item.get("title"),
item.get("hot_value"),
item.get("word_cover"),
collect_time,
hot_type
))
conn.commit()
conn.close()
print(f"[{collect_time}] {hot_type}热榜数据采集完成")
else:
print(f"API返回错误: {hot_data}")
except Exception as e:
print(f"采集热榜数据失败: {e}")
# 设置定时任务
schedule.every(30).minutes.do(fetch_hotlist_data, hot_type='hot') # 总榜每30分钟
schedule.every(2).hours.do(fetch_hotlist_data, hot_type='entertainment') # 娱乐榜每2小时
schedule.every(3).hours.do(fetch_hotlist_data, hot_type='social') # 社会榜每3小时
if __name__ == "__main__":
create_hot_table()
print("热榜数据采集服务启动...")
while True:
schedule.run_pending()
time.sleep(60)
这个脚本会定期采集不同类别的热榜数据，并存储到SQLite数据库中。有了历史数据，我们就可以进行趋势分析了。
2.2 热词趋势分析与可视化
单纯存储数据还不够，我们需要从中提取洞察。使用Pandas进行数据分析是个不错的选择：
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.font_manager import FontProperties
import jieba
from wordcloud import WordCloud
# 设置中文字体
font_path = "/path/to/your/chinese_font.ttf"
font_prop = FontProperties(fname=font_path)
def analyze_hot_trends(days=7):
"""分析最近N天的热榜趋势"""
conn = sqlite3.connect('tiktok_data.db')
# 读取最近7天的数据
query = f"""
SELECT title, hot_value, collect_time, hot_type
FROM hotlist
WHERE collect_time >= datetime('now', '-{days} days')
ORDER BY collect_time DESC
"""
df = pd.read_sql_query(query, conn)
conn.close()
if df.empty:
print("没有找到足够的数据")
return
# 1. 热榜类型分布
type_distribution = df['hot_type'].value_counts()
# 2. 热度变化趋势
df['date'] = pd.to_datetime(df['collect_time']).dt.date
daily_avg_hot = df.groupby('date')['hot_value'].mean()
# 3. 高频词汇分析
all_titles = ' '.join(df['title'].dropna().tolist())
word_freq = {}
for word in jieba.cut(all_titles):
if len(word) > 1: # 只保留长度大于1的词
word_freq[word] = word_freq.get(word, 0) + 1
# 生成词云
wordcloud = WordCloud(
font_path=font_path,
width=800,
height=400,
background_color='white'
).generate_from_frequencies(word_freq)
# 可视化
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# 热榜类型分布饼图
axes[0, 0].pie(type_distribution.values, labels=type_distribution.index, autopct='%1.1f%%')
axes[0, 0].set_title('热榜类型分布', fontproperties=font_prop)
# 日均热度变化折线图
axes[0, 1].plot(daily_avg_hot.index.astype(str), daily_avg_hot.values, marker='o')
axes[0, 1].set_title('日均热度变化趋势', fontproperties=font_prop)
axes[0, 1].set_xlabel('日期', fontproperties=font_prop)
axes[0, 1].set_ylabel('平均热度', fontproperties=font_prop)
axes[0, 1].tick_params(axis='x', rotation=45)
# 词云
axes[1, 0].imshow(wordcloud, interpolation='bilinear')
axes[1, 0].axis('off')
axes[1, 0].set_title('热榜关键词云', fontproperties=font_prop)
# 高频词条形图（前10）
top_words = pd.Series(word_freq).nlargest(10)
axes[1, 1].barh(range(len(top_words)), top_words.values)
axes[1, 1].set_yticks(range(len(top_words)))
axes[1, 1].set_yticklabels(top_words.index, fontproperties=font_prop)
axes[1, 1].set_title('高频词汇Top 10', fontproperties=font_prop)
axes[1, 1].set_xlabel('出现次数', fontproperties=font_prop)
plt.tight_layout()
plt.savefig('hot_trends_analysis.png', dpi=300, bbox_inches='tight')
plt.show()
return df, word_freq
# 运行分析
df_result, word_freq_result = analyze_hot_trends(days=7)
这段代码会生成一个包含四个子图的分析报告：热榜类型分布、热度变化趋势、关键词云和高频词排名。这样的可视化报告比原始数据直观得多。
2.3 实时预警系统
对于运营和内容团队来说，及时发现突发热点很重要。我们可以基于热榜数据构建一个简单的预警系统：
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import numpy as np
def detect_hotspot_anomaly(current_data, historical_window=24):
"""检测热点异常（突然飙升的话题）"""
conn = sqlite3.connect('tiktok_data.db')
# 获取当前热榜
current_hot_items = []
for item in current_data:
current_hot_items.append({
'title': item['title'],
'hot_value': item['hot_value'],
'rank': item['rank']
})
# 获取历史数据（过去24小时）
query = f"""
SELECT title, AVG(hot_value) as avg_hot, MAX(hot_value) as max_hot
FROM hotlist
WHERE collect_time >= datetime('now', '-{historical_window} hours')
GROUP BY title
"""
historical_stats = pd.read_sql_query(query, conn)
conn.close()
anomalies = []
for current_item in current_hot_items:
title = current_item['title']
current_hot = current_item['hot_value']
# 查找历史记录
historical_record = historical_stats[historical_stats['title'] == title]
if not historical_record.empty:
avg_hot = historical_record['avg_hot'].values[0]
max_hot = historical_record['max_hot'].values[0]
# 如果当前热度超过历史平均值的2倍且超过历史最大值
if current_hot > avg_hot * 2 and current_hot > max_hot:
increase_ratio = current_hot / avg_hot if avg_hot > 0 else float('inf')
anomalies.append({
'title': title,
'current_hot': current_hot,
'historical_avg': avg_hot,
'increase_ratio': increase_ratio,
'rank': current_item['rank']
})
return anomalies
def send_alert_email(anomalies, recipient_email):
"""发送预警邮件"""
if not anomalies:
return
# 按增长比例排序
anomalies_sorted = sorted(anomalies, key=lambda x: x['increase_ratio'], reverse=True)
# 构建邮件内容
msg = MIMEMultipart('alternative')
msg['Subject'] = f'抖音热榜异常预警 - {len(anomalies)}个话题飙升'
msg['From'] = 'your_email@example.com'
msg['To'] = recipient_email
# HTML内容
html = f"""
<html>
<body>
<h2>抖音热榜异常预警</h2>
<p>检测时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
<p>共发现{len(anomalies)}个异常飙升话题:</p>
<table border="1" cellpadding="5">
<tr>
<th>排名</th>
<th>话题</th>
<th>当前热度</th>
<th>历史平均</th>
<th>增长倍数</th>
</tr>
"""
for anomaly in anomalies_sorted[:10]: # 只显示前10个
html += f"""
<tr>
<td>{anomaly['rank']}</td>
<td>{anomaly['title']}</td>
<td>{anomaly['current_hot']:,.0f}</td>
<td>{anomaly['historical_avg']:,.0f}</td>
<td>{anomaly['increase_ratio']:.1f}x</td>
</tr>
"""
html += """
</table>
<p><small>此邮件由自动化系统发送，请勿直接回复</small></p>
</body>
</html>
"""
msg.attach(MIMEText(html, 'html'))
# 发送邮件（需要配置SMTP服务器）
try:
with smtplib.SMTP('smtp.example.com', 587) as server:
server.starttls()
server.login('your_email@example.com', 'your_password')
server.send_message(msg)
print(f"预警邮件已发送至 {recipient_email}")
except Exception as e:
print(f"发送邮件失败: {e}")
# 在定时任务中添加预警功能
def scheduled_hotlist_with_alert():
"""带预警的定时热榜采集"""
hot_data = fetch_hotlist_data('hot')
anomalies = detect_hotspot_anomaly(hot_data)
if anomalies:
send_alert_email(anomalies, 'your_team@example.com')
return hot_data
这个预警系统会在某个话题热度突然飙升时自动发送邮件通知，帮助团队第一时间抓住热点。
3. 账号数据深度监控：竞品分析与成长追踪
对于运营多个账号的团队或个人来说，监控账号数据变化至关重要。TikTokDownloader的/douyin/account接口可以获取账号的详细数据，我们可以利用这个功能构建账号监控系统。
3.1 多账号数据批量采集
首先，我们需要一个管理多个账号配置的系统。我建议使用YAML配置文件来管理要监控的账号：
# accounts_to_monitor.yaml
accounts:
- name: "品牌官方账号"
url: "https://www.douyin.com/user/品牌ID"
category: "brand"
priority: "high"
alert_threshold: 0.1 # 粉丝变化超过10%时告警
- name: "竞品A"
url: "https://www.douyin.com/user/竞品A_ID"
category: "competitor"
priority: "medium"
alert_threshold: 0.15
- name: "行业KOL"
url: "https://www.douyin.com/user/KOL_ID"
category: "influencer"
priority: "low"
alert_threshold: 0.2
- name: "自家小号"
url: "https://www.douyin.com/user/小号ID"
category: "secondary"
priority: "low"
alert_threshold: 0.25
schedule:
frequency: "daily" # daily, weekly, monthly
time: "02:00" # 每天凌晨2点执行
retention_days: 90 # 数据保留90天
然后创建对应的数据采集脚本：
import yaml
import pandas as pd
from datetime import datetime, timedelta
import httpx
import json
class AccountMonitor:
def __init__(self, config_path='accounts_to_monitor.yaml'):
with open(config_path, 'r', encoding='utf-8') as f:
self.config = yaml.safe_load(f)
self.api_base = "http://localhost:5555"
self.db_path = "account_monitor.db"
self.init_database()
def init_database(self):
"""初始化数据库表结构"""
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
# 账号基本信息表
c.execute('''
CREATE TABLE IF NOT EXISTS accounts (
account_id TEXT PRIMARY KEY,
nickname TEXT,
unique_id TEXT,
signature TEXT,
category TEXT,
created_at TIMESTAMP,
updated_at TIMESTAMP
)
''')
# 账号数据快照表（每次采集的数据）
c.execute('''
CREATE TABLE IF NOT EXISTS account_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
account_id TEXT,
collect_time TIMESTAMP,
follower_count INTEGER,
following_count INTEGER,
total_favorited INTEGER,
aweme_count INTEGER,
likes_count INTEGER,
video_count INTEGER,
digg_count INTEGER,
FOREIGN KEY (account_id) REFERENCES accounts (account_id)
)
''')
# 作品数据表
c.execute('''
CREATE TABLE IF NOT EXISTS videos (
video_id TEXT PRIMARY KEY,
account_id TEXT,
collect_time TIMESTAMP,
desc TEXT,
create_time TIMESTAMP,
duration INTEGER,
play_count INTEGER,
digg_count INTEGER,
comment_count INTEGER,
share_count INTEGER,
download_count INTEGER,
music_title TEXT,
music_author TEXT,
FOREIGN KEY (account_id) REFERENCES accounts (account_id)
)
''')
conn.commit()
conn.close()
def fetch_account_data(self, account_url):
"""通过API获取账号数据"""
try:
api_url = f"{self.api_base}/douyin/account"
data = {"url": account_url, "pages": 1}
response = httpx.post(api_url, json=data, timeout=30)
result = response.json()
if result.get("status") == "success":
return result.get("data", {})
else:
print(f"获取账号数据失败: {result.get('message')}")
return None
except Exception as e:
print(f"API请求异常: {e}")
return None
def fetch_account_videos(self, account_url, max_pages=3):
"""获取账号作品数据"""
try:
# 这里需要调用批量下载账号作品的API
# 注意：实际API端点可能需要调整
api_url = f"{self.api_base}/douyin/account/videos"
data = {
"url": account_url,
"max_pages": max_pages,
"tab": "post" # post: 发布作品, favorite: 喜欢作品
}
response = httpx.post(api_url, json=data, timeout=60)
result = response.json()
if result.get("status") == "success":
return result.get("data", {}).get("videos", [])
else:
print(f"获取作品数据失败: {result.get('message')}")
return []
except Exception as e:
print(f"获取作品数据异常: {e}")
return []
def save_account_snapshot(self, account_data, account_info):
"""保存账号数据快照"""
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
# 保存或更新账号基本信息
c.execute('''
INSERT OR REPLACE INTO accounts
(account_id, nickname, unique_id, signature, category, updated_at)
VALUES (?, ?, ?, ?, ?, ?)
''', (
account_data.get('uid'),
account_data.get('nickname'),
account_data.get('unique_id'),
account_data.get('signature'),
account_info.get('category'),
datetime.now()
))
# 保存数据快照
stats = account_data.get('stats', {})
c.execute('''
INSERT INTO account_snapshots
(account_id, collect_time, follower_count, following_count,
total_favorited, aweme_count, likes_count, video_count, digg_count)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
account_data.get('uid'),
datetime.now(),
stats.get('follower_count', 0),
stats.get('following_count', 0),
stats.get('total_favorited', 0),
stats.get('aweme_count', 0),
account_data.get('likes_count', 0),
account_data.get('video_count', 0),
account_data.get('digg_count', 0)
))
conn.commit()
conn.close()
print(f"账号 {account_data.get('nickname')} 数据已保存")
def save_videos_data(self, videos, account_id):
"""保存作品数据"""
if not videos:
return
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
for video in videos:
try:
c.execute('''
INSERT OR REPLACE INTO videos
(video_id, account_id, collect_time, desc, create_time,
duration, play_count, digg_count, comment_count,
share_count, download_count, music_title, music_author)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
video.get('aweme_id'),
account_id,
datetime.now(),
video.get('desc', '')[:500], # 限制描述长度
datetime.fromtimestamp(video.get('create_time', 0)),
video.get('duration', 0),
video.get('statistics', {}).get('play_count', 0),
video.get('statistics', {}).get('digg_count', 0),
video.get('statistics', {}).get('comment_count', 0),
video.get('statistics', {}).get('share_count', 0),
video.get('statistics', {}).get('download_count', 0),
video.get('music', {}).get('title', ''),
video.get('music', {}).get('author', '')
))
except Exception as e:
print(f"保存作品数据失败: {e}")
continue
conn.commit()
conn.close()
print(f"已保存 {len(videos)} 个作品数据")
def run_monitoring(self):
"""执行监控任务"""
print(f"开始执行账号监控任务，共 {len(self.config['accounts'])} 个账号")
for account in self.config['accounts']:
print(f"处理账号: {account['name']}")
# 获取账号数据
account_data = self.fetch_account_data(account['url'])
if account_data:
self.save_account_snapshot(account_data, account)
# 获取作品数据（可选，根据配置决定）
if self.config.get('collect_videos', False):
videos = self.fetch_account_videos(account['url'])
if videos:
self.save_videos_data(videos, account_data.get('uid'))
# 避免请求过于频繁
time.sleep(2)
print("账号监控任务完成")
def generate_growth_report(self, account_id, days=30):
"""生成账号成长报告"""
conn = sqlite3.connect(self.db_path)
query = f"""
SELECT
DATE(collect_time) as date,
AVG(follower_count) as avg_followers,
AVG(digg_count) as avg_likes,
COUNT(DISTINCT video_id) as new_videos
FROM account_snapshots
LEFT JOIN videos ON account_snapshots.account_id = videos.account_id
AND DATE(videos.collect_time) = DATE(account_snapshots.collect_time)
WHERE account_snapshots.account_id = ?
AND account_snapshots.collect_time >= datetime('now', '-{days} days')
GROUP BY DATE(account_snapshots.collect_time)
ORDER BY date
"""
df = pd.read_sql_query(query, conn, params=(account_id,))
conn.close()
if df.empty:
return None
# 计算增长率
df['follower_growth'] = df['avg_followers'].pct_change() * 100
df['like_growth'] = df['avg_likes'].pct_change() * 100
return df
# 使用示例
if __name__ == "__main__":
monitor = AccountMonitor()
# 执行一次监控
monitor.run_monitoring()
# 生成本品牌账号的成长报告
brand_account_id = "你的品牌账号ID"
growth_report = monitor.generate_growth_report(brand_account_id, days=30)
if growth_report is not None:
print("账号成长报告:")
print(growth_report.tail()) # 显示最近几天的数据
这个账号监控系统可以定期采集指定账号的数据，并存储到数据库中。有了历史数据，我们就可以进行各种分析了。
3.2 竞品对比分析仪表板
有了多个账号的数据后，我们可以创建一个竞品对比分析仪表板。这里使用Pyecharts进行可视化：
from pyecharts import options as opts
from pyecharts.charts import Line, Bar, Radar, Grid
from pyecharts.commons.utils import JsCode
def create_competitor_dashboard(account_ids, days=30):
"""创建竞品对比仪表板"""
conn = sqlite3.connect('account_monitor.db')
# 获取各账号最近30天的数据
all_data = {}
account_names = {}
for account_id in account_ids:
# 获取账号名称
cursor = conn.cursor()
cursor.execute("SELECT nickname FROM accounts WHERE account_id = ?", (account_id,))
result = cursor.fetchone()
account_name = result[0] if result else account_id
account_names[account_id] = account_name
# 获取成长数据
query = f"""
SELECT
DATE(collect_time) as date,
AVG(follower_count) as followers,
AVG(digg_count) as likes,
AVG(aweme_count) as videos
FROM account_snapshots
WHERE account_id = ?
AND collect_time >= datetime('now', '-{days} days')
GROUP BY DATE(collect_time)
ORDER BY date
"""
df = pd.read_sql_query(query, conn, params=(account_id,))
all_data[account_id] = df
conn.close()
# 1. 粉丝增长趋势图
line_followers = (
Line()
.set_global_opts(
title_opts=opts.TitleOpts(title="粉丝增长趋势", subtitle="最近30天"),
tooltip_opts=opts.TooltipOpts(trigger="axis"),
legend_opts=opts.LegendOpts(pos_top="10%"),
xaxis_opts=opts.AxisOpts(
type_="category",
boundary_gap=False,
axislabel_opts=opts.LabelOpts(rotate=45)
),
yaxis_opts=opts.AxisOpts(
type_="value",
axislabel_opts=opts.LabelOpts(formatter="{value}"),
splitline_opts=opts.SplitLineOpts(is_show=True)
)
)
)
# 2. 互动数据对比雷达图
# 计算各账号的平均互动数据
radar_data = []
for account_id in account_ids:
df = all_data[account_id]
if not df.empty:
avg_followers = df['followers'].mean()
avg_likes = df['likes'].mean()
avg_videos = df['videos'].mean()
# 计算互动率（假设）
engagement_rate = (avg_likes / avg_followers * 100) if avg_followers > 0 else 0
radar_data.append({
"name": account_names[account_id],
"value": [
avg_followers / 10000, # 转换为万为单位
avg_likes / 10000,
avg_videos,
engagement_rate,
avg_followers / max(avg_videos, 1) # 单视频平均吸粉
]
})
radar = (
Radar()
.add_schema(
schema=[
opts.RadarIndicatorItem(name="粉丝数(万)", max_=100),
opts.RadarIndicatorItem(name="获赞数(万)", max_=50),
opts.RadarIndicatorItem(name="作品数", max_=100),
opts.RadarIndicatorItem(name="互动率%", max_=10),
opts.RadarIndicatorItem(name="单视频吸粉", max_=1000),
]
)
.set_series_opts(label_opts=opts.LabelOpts(is_show=False))
)
# 为每个账号添加数据到图表
colors = ['#5470c6', '#91cc75', '#fac858', '#ee6666', '#73c0de']
for idx, account_id in enumerate(account_ids):
df = all_data[account_id]
if not df.empty:
color = colors[idx % len(colors)]
# 添加到折线图
line_followers.add_xaxis(df['date'].tolist())
line_followers.add_yaxis(
series_name=account_names[account_id],
y_axis=df['followers'].tolist(),
is_smooth=True,
linestyle_opts=opts.LineStyleOpts(width=3),
itemstyle_opts=opts.ItemStyleOpts(color=color),
label_opts=opts.LabelOpts(is_show=False)
)
# 添加到雷达图
if idx < len(radar_data):
radar.add(
series_name=radar_data[idx]["name"],
data=[radar_data[idx]["value"]],
linestyle_opts=opts.LineStyleOpts(width=2, color=color),
areastyle_opts=opts.AreaStyleOpts(opacity=0.1, color=color)
)
# 3. 最近作品表现条形图
# 获取各账号最近5个作品的数据
conn = sqlite3.connect('account_monitor.db')
recent_videos_data = []
for account_id in account_ids:
query = """
SELECT
v.desc as title,
v.digg_count as likes,
v.comment_count as comments,
v.share_count as shares,
v.play_count as plays,
v.create_time
FROM videos v
WHERE v.account_id = ?
ORDER BY v.create_time DESC
LIMIT 5
"""
df_videos = pd.read_sql_query(query, conn, params=(account_id,))
for _, row in df_videos.iterrows():
# 简化标题
title_short = row['title'][:20] + "..." if len(row['title']) > 20 else row['title']
recent_videos_data.append({
"account": account_names[account_id],
"title": title_short,
"likes": row['likes'],
"comments": row['comments'],
"shares": row['shares'],
"plays": row['plays']
})
conn.close()
# 转换为DataFrame便于分析
df_recent = pd.DataFrame(recent_videos_data)
# 创建作品表现条形图
if not df_recent.empty:
# 按点赞数排序
df_sorted = df_recent.sort_values('likes', ascending=True).tail(10)
bar_videos = (
Bar()
.add_xaxis(df_sorted['title'].tolist())
.add_yaxis(
"点赞数",
df_sorted['likes'].tolist(),
itemstyle_opts=opts.ItemStyleOpts(color="#91cc75"),
label_opts=opts.LabelOpts(
position="right",
formatter=JsCode("function(params){return params.value.toLocaleString();}")
)
)
.reversal_axis()
.set_global_opts(
title_opts=opts.TitleOpts(title="热门作品表现", pos_left="center"),
xaxis_opts=opts.AxisOpts(
axislabel_opts=opts.LabelOpts(
formatter=JsCode("function(value){return value.toLocaleString();}")
)
),
tooltip_opts=opts.TooltipOpts(
trigger="axis",
axis_pointer_type="shadow",
formatter=JsCode("""
function(params){
let data = params[0];
let title = data.name;
let likes = data.value.toLocaleString();
let idx = data.dataIndex;
let comments = %s[idx].toLocaleString();
let shares = %s[idx].toLocaleString();
let plays = %s[idx].toLocaleString();
return title + '<br/>' +
'点赞: ' + likes + '<br/>' +
'评论: ' + comments + '<br/>' +
'分享: ' + shares + '<br/>' +
'播放: ' + plays;
}
""" % (
df_sorted['comments'].tolist(),
df_sorted['shares'].tolist(),
df_sorted['plays'].tolist()
))
)
)
)
# 使用Grid组合图表
grid = (
Grid()
.add(
line_followers,
grid_opts=opts.GridOpts(pos_left="10%", pos_right="55%", pos_top="10%", height="35%")
)
.add(
radar,
grid_opts=opts.GridOpts(pos_left="60%", pos_right="5%", pos_top="10%", height="35%")
)
)
if not df_recent.empty:
grid.add(
bar_videos,
grid_opts=opts.GridOpts(pos_left="10%", pos_right="5%", pos_top="55%", height="40%")
)
# 渲染图表
grid.render("competitor_dashboard.html")
print("竞品分析仪表板已生成: competitor_dashboard.html")
return grid
# 使用示例
account_ids_to_compare = ["账号1_ID", "账号2_ID", "账号3_ID"]
dashboard = create_competitor_dashboard(account_ids_to_compare, days=30)
这个仪表板会生成一个HTML文件，包含三个主要部分：粉丝增长趋势折线图、账号综合能力雷达图和热门作品表现条形图。你可以定期运行这个脚本，然后将生成的HTML文件分享给团队，或者部署到内部服务器上。
3.3 内容策略建议生成
基于收集到的数据，我们还可以尝试生成一些内容策略建议。虽然这需要更复杂的算法，但我们可以先实现一个简单的版本：
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
import numpy as np
def analyze_content_patterns(account_id, days=90):
"""分析账号的内容模式"""
conn = sqlite3.connect('account_monitor.db')
# 获取账号最近的作品数据
query = f"""
SELECT
v.desc,
v.digg_count,
v.comment_count,
v.share_count,
v.play_count,
v.duration,
v.create_time,
v.music_title,
v.music_author
FROM videos v
WHERE v.account_id = ?
AND v.create_time >= datetime('now', '-{days} days')
ORDER BY v.create_time DESC
"""
df = pd.read_sql_query(query, conn, params=(account_id,))
conn.close()
if df.empty or len(df) < 10:
print("作品数据不足，无法进行分析")
return None
# 1. 基础统计分析
stats = {
"total_videos": len(df),
"avg_likes": df['digg_count'].mean(),
"avg_comments": df['comment_count'].mean(),
"avg_shares": df['share_count'].mean(),
"avg_plays": df['play_count'].mean(),
"avg_duration": df['duration'].mean(),
"max_likes": df['digg_count'].max(),
"max_comments": df['comment_count'].max(),
"max_shares": df['share_count'].max(),
"best_performing_video": df.loc[df['digg_count'].idxmax()]['desc'][:50] if not df.empty else ""
}
# 2. 发布时间分析
df['hour'] = pd.to_datetime(df['create_time']).dt.hour
df['weekday'] = pd.to_datetime(df['create_time']).dt.weekday
best_hour = df.groupby('hour')['digg_count'].mean().idxmax()
best_weekday = df.groupby('weekday')['digg_count'].mean().idxmax()
stats['best_posting_hour'] = best_hour
stats['best_posting_weekday'] = best_weekday
# 3. 内容聚类分析（尝试找出成功的内容模式）
try:
# 选择特征
features = df[['digg_count', 'comment_count', 'share_count', 'play_count', 'duration']].fillna(0)
# 标准化
scaler = StandardScaler()
features_scaled = scaler.fit_transform(features)
# 聚类
n_clusters = min(3, len(df) // 5) # 确保每个簇有足够样本
if n_clusters >= 2:
kmeans = KMeans(n_clusters=n_clusters, random_state=42)
df['cluster'] = kmeans.fit_predict(features_scaled)
# 分析每个簇的特征
cluster_stats = []
for cluster_id in range(n_clusters):
cluster_data = df[df['cluster'] == cluster_id]
if len(cluster_data) > 0:
cluster_stats.append({
'cluster': cluster_id,
'size': len(cluster_data),
'avg_likes': cluster_data['digg_count'].mean(),
'avg_comments': cluster_data['comment_count'].mean(),
'avg_shares': cluster_data['share_count'].mean(),
'avg_duration': cluster_data['duration'].mean(),
'sample_titles': cluster_data['desc'].head(3).tolist()
})
# 找出表现最好的簇
cluster_stats_sorted = sorted(cluster_stats, key=lambda x: x['avg_likes'], reverse=True)
best_cluster = cluster_stats_sorted[0] if cluster_stats_sorted else None
stats['content_clusters'] = cluster_stats
stats['best_content_pattern'] = best_cluster
else:
stats['content_clusters'] = []
stats['best_content_pattern'] = None
except Exception as e:
print(f"聚类分析失败: {e}")
stats['content_clusters'] = []
stats['best_content_pattern'] = None
# 4. 生成建议
recommendations = generate_recommendations(stats, df)
return {
"statistics": stats,
"recommendations": recommendations,
"raw_data_sample": df.head(5).to_dict('records')
}
def generate_recommendations(stats, df):
"""基于分析结果生成建议"""
recommendations = []
# 基于互动率的建议
engagement_rate = stats['avg_comments'] / max(stats['avg_likes'], 1)
if engagement_rate < 0.01:
recommendations.append("评论互动率较低，建议在视频中增加提问或引导评论的内容")
elif engagement_rate > 0.05:
recommendations.append("评论互动率很高，继续保持与粉丝的互动")
# 基于分享率的建议
share_rate = stats['avg_shares'] / max(stats['avg_likes'], 1)
if share_rate < 0.005:
recommendations.append("分享率有待提升，考虑制作更有传播价值的内容")
# 基于发布时间
hour_names = {0: "凌晨", 6: "早晨", 12: "中午", 18: "傍晚"}
best_hour = stats.get('best_posting_hour', 12)
closest_hour_key = min(hour_names.keys(), key=lambda x: abs(x - best_hour))
recommendations.append(f"最佳发布时间可能在{hour_names[closest_hour_key]}时段（基于历史数据）")
# 基于内容时长
avg_duration = stats['avg_duration']
if avg_duration < 15:
recommendations.append("视频时长较短，可以尝试制作更深入的内容")
elif avg_duration > 60:
recommendations.append("视频时长较长，考虑制作短视频版本以提高完播率")
# 基于最佳表现内容模式
best_pattern = stats.get('best_content_pattern')
if best_pattern and best_pattern.get('sample_titles'):
recommendations.append(f"表现最好的内容类型可能涉及: {', '.join(best_pattern['sample_titles'][:2])}")
return recommendations
# 使用示例
account_analysis = analyze_content_patterns("你的账号ID", days=90)
if account_analysis:
print("=== 账号分析报告 ===")
print(f"分析时间段: 最近90天")
print(f"作品数量: {account_analysis['statistics']['total_videos']}")
print(f"平均点赞: {account_analysis['statistics']['avg_likes']:.0f}")
print(f"平均评论: {account_analysis['statistics']['avg_comments']:.0f}")
print(f"最高点赞作品: {account_analysis['statistics']['best_performing_video']}")
print("\n=== 内容策略建议 ===")
for i, rec in enumerate(account_analysis['recommendations'], 1):
print(f"{i}. {rec}")
这个分析系统可以帮助你理解什么类型的内容在你的账号上表现最好，什么时间发布效果最佳，从而优化你的内容策略。
4. 评论数据情感分析与用户洞察
评论数据是了解用户反馈的宝贵资源。TikTokDownloader可以采集作品评论数据，我们可以利用这些数据进行情感分析和用户洞察。
4.1 评论数据批量采集与分析
首先，我们需要一个系统来定期采集热门作品的评论数据：
import jieba.analyse
from snownlp import SnowNLP
from collections import Counter
import re
class CommentAnalyzer:
def __init__(self):
self.api_base = "http://localhost:5555"
self.db_path = "comments_analysis.db"
self.init_database()
# 加载停用词
self.stopwords = self.load_stopwords()
def load_stopwords(self):
"""加载中文停用词表"""
stopwords = set()
try:
with open('chinese_stopwords.txt', 'r', encoding='utf-8') as f:
for line in f:
stopwords.add(line.strip())
except FileNotFoundError:
# 如果没有停用词文件，使用一个基本的列表
basic_stopwords = ["的", "了", "在", "是", "我", "有", "和", "就",
"不", "人", "都", "一", "一个", "上", "也", "很",
"到", "说", "要", "去", "你", "会", "着", "没有",
"看", "好", "自己", "这"]
stopwords = set(basic_stopwords)
return stopwords
def init_database(self):
"""初始化评论数据库"""
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
# 作品表
c.execute('''
CREATE TABLE IF NOT EXISTS videos_comments (
video_id TEXT PRIMARY KEY,
video_desc TEXT,
collect_time TIMESTAMP,
total_comments INTEGER,
avg_sentiment REAL
)
''')
# 评论表
c.execute('''
CREATE TABLE IF NOT EXISTS comments (
comment_id TEXT PRIMARY KEY,
video_id TEXT,
user_id TEXT,
nickname TEXT,
content TEXT,
create_time TIMESTAMP,
digg_count INTEGER,
sentiment_score REAL,
keywords TEXT,
collect_time TIMESTAMP,
FOREIGN KEY (video_id) REFERENCES videos_comments (video_id)
)
''')
# 关键词表
c.execute('''
CREATE TABLE IF NOT EXISTS keywords (
keyword TEXT,
video_id TEXT,
frequency INTEGER,
collect_time TIMESTAMP,
PRIMARY KEY (keyword, video_id)
)
''')
conn.commit()
conn.close()
def fetch_comments(self, video_url, max_pages=5):
"""获取作品评论数据"""
try:
api_url = f"{self.api_base}/douyin/comment"
data = {
"url": video_url,
"pages": max_pages
}
response = httpx.post(api_url, json=data, timeout=60)
result = response.json()
if result.get("status") == "success":
return result.get("data", {})
else:
print(f"获取评论数据失败: {result.get('message')}")
return None
except Exception as e:
print(f"API请求异常: {e}")
return None
def analyze_sentiment(self, text):
"""分析文本情感（简单版本）"""
try:
s = SnowNLP(text)
return s.sentiments # 返回0-1之间的情感分数，越接近1越积极
except:
return 0.5 # 中性
def extract_keywords(self, text, top_k=5):
"""提取关键词"""
# 去除特殊字符和空白
text_clean = re.sub(r'[^\w\u4e00-\u9fff]+', ' ', text)
# 使用jieba提取关键词
keywords = jieba.analyse.extract_tags(
text_clean,
topK=top_k,
withWeight=False,
allowPOS=('n', 'vn', 'v', 'a') # 名词、动名词、动词、形容词
)
# 过滤停用词
keywords_filtered = [kw for kw in keywords if kw not in self.stopwords]
return keywords_filtered
def process_comments(self, video_data):
"""处理评论数据"""
if not video_data:
return None
video_id = video_data.get('aweme_id')
video_desc = video_data.get('desc', '')[:200]
comments_list = video_data.get('comments', [])
if not comments_list:
print(f"视频 {video_id} 没有评论数据")
return None
all_comments_processed = []
all_keywords = []
sentiment_scores = []
for comment in comments_list:
content = comment.get('text', '')
if not content or len(content.strip()) < 2:
continue
# 分析情感
sentiment = self.analyze_sentiment(content)
sentiment_scores.append(sentiment)
# 提取关键词
keywords = self.extract_keywords(content, top_k=3)
all_keywords.extend(keywords)
# 处理后的评论数据
processed_comment = {
'comment_id': comment.get('cid', ''),
'user_id': comment.get('user', {}).get('uid', ''),
'nickname': comment.get('user', {}).get('nickname', ''),
'content': content,
'create_time': datetime.fromtimestamp(comment.get('create_time', 0)),
'digg_count': comment.get('digg_count', 0),
'sentiment_score': sentiment,
'keywords': ','.join(keywords)
}
all_comments_processed.append(processed_comment)
# 统计关键词频率
keyword_freq = Counter(all_keywords)
# 计算平均情感
avg_sentiment = sum(sentiment_scores) / len(sentiment_scores) if sentiment_scores else 0.5
return {
'video_id': video_id,
'video_desc': video_desc,
'total_comments': len(all_comments_processed),
'avg_sentiment': avg_sentiment,
'comments': all_comments_processed,
'keyword_freq': dict(keyword_freq.most_common(20)) # 取前20个关键词
}
def save_to_database(self, processed_data):
"""保存处理后的数据到数据库"""
if not processed_data:
return
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
collect_time = datetime.now()
# 保存视频信息
c.execute('''
INSERT OR REPLACE INTO videos_comments
(video_id, video_desc, collect_time, total_comments, avg_sentiment)
VALUES (?, ?, ?, ?, ?)
''', (
processed_data['video_id'],
processed_data['video_desc'],
collect_time,
processed_data['total_comments'],
processed_data['avg_sentiment']
))
# 保存评论
for comment in processed_data['comments']:
c.execute('''
INSERT OR REPLACE INTO comments
(comment_id, video_id, user_id, nickname, content, create_time,
digg_count, sentiment_score, keywords, collect_time)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
comment['comment_id'],
processed_data['video_id'],
comment['user_id'],
comment['nickname'],
comment['content'],
comment['create_time'],
comment['digg_count'],
comment['sentiment_score'],
comment['keywords'],
collect_time
))
# 保存关键词
for keyword, freq in processed_data['keyword_freq'].items():
c.execute('''
INSERT OR REPLACE INTO keywords
(keyword, video_id, frequency, collect_time)
VALUES (?, ?, ?, ?)
''', (
keyword,
processed_data['video_id'],
freq,
collect_time
))
conn.commit()
conn.close()
print(f"视频 {processed_data['video_id']} 的评论数据已保存，共 {processed_data['total_comments']} 条评论")
def analyze_video_comments(self, video_url):
"""分析单个视频的评论"""
print(f"开始分析视频: {video_url}")
# 获取评论数据
raw_data = self.fetch_comments(video_url)
if not raw_data:
return None
# 处理数据
processed_data = self.process_comments(raw_data)
if not processed_data:
return None
# 保存到数据库
self.save_to_database(processed_data)
return processed_data
def generate_sentiment_report(self, video_id=None, days=7):
"""生成情感分析报告"""
conn = sqlite3.connect(self.db_path)
if video_id:
# 单个视频的报告
query = """
SELECT
v.video_desc,
v.avg_sentiment,
v.total_comments,
COUNT(DISTINCT c.user_id) as unique_users,
AVG(c.digg_count) as avg_comment_likes
FROM videos_comments v
LEFT JOIN comments c ON v.video_id = c.video_id
WHERE v.video_id = ?
GROUP BY v.video_id
"""
df_video = pd.read_sql_query(query, conn, params=(video_id,))
# 获取情感分布
query_sentiment = """
SELECT
CASE
WHEN sentiment_score > 0.7 THEN '积极'
WHEN sentiment_score < 0.3 THEN '消极'
ELSE '中性'
END as sentiment_category,
COUNT(*) as count,
AVG(digg_count) as avg_likes
FROM comments
WHERE video_id = ?
GROUP BY sentiment_category
ORDER BY count DESC
"""
df_sentiment = pd.read_sql_query(query_sentiment, conn, params=(video_id,))
# 获取热门关键词
query_keywords = """
SELECT keyword, frequency
FROM keywords
WHERE video_id = ?
ORDER BY frequency DESC
LIMIT 10
"""
df_keywords = pd.read_sql_query(query_keywords, conn, params=(video_id,))
conn.close()
return {
'video_info': df_video.to_dict('records')[0] if not df_video.empty else {},
'sentiment_distribution': df_sentiment.to_dict('records'),
'top_keywords': df_keywords.to_dict('records')
}
else:
# 多个视频的汇总报告
query = f"""
SELECT
v.video_id,
v.video_desc,
v.avg_sentiment,
v.total_comments,
COUNT(DISTINCT c.user_id) as unique_users,
AVG(c.digg_count) as avg_comment_likes,
v.collect_time
FROM videos_comments v
LEFT JOIN comments c ON v.video_id = c.video_id
WHERE v.collect_time >= datetime('now', '-{days} days')
GROUP BY v.video_id
ORDER BY v.total_comments DESC
LIMIT 20
"""
df_videos = pd.read_sql_query(query, conn)
# 总体情感趋势
query_trend = f"""
SELECT
DATE(collect_time) as date,
AVG(avg_sentiment) as daily_avg_sentiment,
SUM(total_comments) as daily_total_comments
FROM videos_comments
WHERE collect_time >= datetime('now', '-{days} days')
GROUP BY DATE(collect_time)
ORDER BY date
"""
df_trend = pd.read_sql_query(query_trend, conn)
# 高频关键词（跨视频）
query_global_keywords = f"""
SELECT
k.keyword,
SUM(k.frequency) as total_frequency,
COUNT(DISTINCT k.video_id) as video_count
FROM keywords k
JOIN videos_comments v ON k.video_id = v.video_id
WHERE v.collect_time >= datetime('now', '-{days} days')
GROUP BY k.keyword
ORDER BY total_frequency DESC
LIMIT 15
"""
df_global_keywords = pd.read_sql_query(query_global_keywords, conn)
conn.close()
return {
'top_videos': df_videos.to_dict('records'),
'sentiment_trend': df_trend.to_dict('records'),
'global_keywords': df_global_keywords.to_dict('records')
}
# 使用示例
if __name__ == "__main__":
analyzer = CommentAnalyzer()
# 分析一个视频的评论
video_url = "https://www.douyin.com/video/视频ID"
result = analyzer.analyze_video_comments(video_url)
if result:
print(f"视频分析完成:")
print(f"描述: {result['video_desc']}")
print(f"评论数: {result['total_comments']}")
print(f"平均情感分数: {result['avg_sentiment']:.3f}")
print(f"热门关键词: {list(result['keyword_freq'].keys())[:5]}")
# 生成报告
report = analyzer.generate_sentiment_report(days=7)
if report:
print(f"\n=== 最近7天评论分析报告 ===")
print(f"分析视频数量: {len(report['top_videos'])}")
if report['top_videos']:
print(f"评论最多的视频: {report['top_videos'][0]['video_desc'][:50]}...")
print(f"评论数: {report['top_videos'][0]['total_comments']}")
print(f"\n总体情感趋势:")
for day in report['sentiment_trend'][-3:]: # 显示最近3天
print(f" {day['date']}: 情感{day['daily_avg_sentiment']:.3f}, 评论{day['daily_total_comments']}条")
print(f"\n高频关键词:")
for i, kw in enumerate(report['global_keywords'][:5], 1):
print(f" {i}. {kw['keyword']} (出现{kw['total_frequency']}次)")
这个评论分析系统可以帮助你理解用户对内容的真实反馈，识别常见的问题或赞美点，从而优化未来的内容创作。
4.2 用户画像构建
基于评论数据，我们还可以尝试构建用户画像，了解你的观众群体：
class UserProfileBuilder:
def __init__(self):
self.db_path = "comments_analysis.db"
def analyze_comment_authors(self, video_id=None, min_comments=2):
"""分析评论作者的特征"""
conn = sqlite3.connect(self.db_path)
if video_id:
# 单个视频的评论作者
query = """
SELECT
c.user_id,
c.nickname,
COUNT(*) as comment_count,
AVG(c.sentiment_score) as avg_sentiment,
AVG(c.digg_count) as avg_comment_likes,
GROUP_CONCAT(DISTINCT k.keyword) as common_keywords
FROM comments c
LEFT JOIN keywords k ON c.video_id = k.video_id
WHERE c.video_id = ?
GROUP BY c.user_id, c.nickname
HAVING COUNT(*) >= ?
ORDER BY comment_count DESC
"""
df_users = pd.read_sql_query(query, conn, params=(video_id, min_comments))
else:
# 所有视频的活跃评论者
query = f"""
SELECT
c.user_id,
c.nickname,
COUNT(DISTINCT c.video_id) as videos_commented,
COUNT(*) as total_comments,
AVG(c.sentiment_score) as avg_sentiment,
AVG(c.digg_count) as avg_comment_likes,
MIN(c.create_time) as first_comment_time,
MAX(c.create_time) as last_comment_time
FROM comments c
GROUP BY c.user_id, c.nickname
HAVING COUNT(*) >= ?
ORDER BY total_comments DESC
LIMIT 50
"""
df_users = pd.read_sql_query(query, conn, params=(min_comments,))
conn.close()
if df_users.empty:
return None
# 分析用户特征
user_profiles = []
for _, row in df_users.iterrows():
profile = {
'user_id': row['user_id'],
'nickname': row['nickname'],
'comment_count': row['comment_count'] if 'comment_count' in row else row['total_comments'],
'avg_sentiment': row['avg_sentiment'],
'engagement_level': self.calculate_engagement_level(row)
}
# 添加用户类型标签
profile['user_type'] = self.classify_user_type(row)
user_profiles.append(profile)
# 整体统计
overall_stats = {
'total_users': len(user_profiles),
'avg_comments_per_user': df_users['comment_count' if 'comment_count' in df_users.columns else 'total_comments'].mean(),
'avg_sentiment': df_users['avg_sentiment'].mean(),
'user_type_distribution': Counter([p['user_type'] for p in user_profiles])
}
return {
'user_profiles': user_profiles,
'overall_stats': overall_stats,
'top_commenters': user_profiles[:10] # 前10个最活跃的用户
}
def calculate_engagement_level(self, user_row):
"""计算用户参与度等级"""
comment_count = user_row.get('comment_count', user_row.get('total_comments', 0))
avg_sentiment = user_row.get('avg_sentiment', 0.5)
avg_likes = user_row.get('avg_comment_likes', 0)
# 简单评分算法
score = 0
score += min(comment_count * 0.5, 10) # 评论数量，最高10分
score += avg_sentiment * 5 # 情感积极性，最高5分
score += min(avg_likes * 0.1, 5) # 评论获赞，最高5分
if score >= 15:
return "高参与度"
elif score >= 10:
return "中参与度"
else:
return "低参与度"
def classify_user_type(self, user_row):
"""分类用户类型"""
avg_sentiment = user_row.get('avg_sentiment', 0.5)
comment_count = user_row.get('comment_count', user_row.get('total_comments', 0))
if avg_sentiment > 0.7:
if comment_count > 5:
return "忠实粉丝"
else:
return "积极观众"
elif avg_sentiment < 0.3:
if comment_count > 3:
return "批评者"
else:
return "消极观众"
else:
if comment_count > 10:
return "高频互动者"
else:
return "普通观众"
def generate_user_segments(self, user_profiles_data):
"""生成用户分群报告"""
if not user_profiles_data:
return None
profiles = user_profiles_data['user_profiles']
# 按用户类型分组
segments = {}
for profile in profiles:
user_type = profile['user_type']
if user_type not in segments:
segments[user_type] = {
'count': 0,
'total_comments': 0,
'avg_sentiment': 0,
'example_users': []
}
segments[user_type]['count'] += 1
segments[user_type]['total_comments'] += profile['comment_count']
segments[user_type]['avg_sentiment'] += profile['avg_sentiment']
# 添加示例用户（最多3个）
if len(segments[user_type]['example_users']) < 3:
segments[user_type]['example_users'].append({
'nickname': profile['nickname'],
'comment_count': profile['comment_count'],
'engagement': profile['engagement_level']
})
# 计算平均值
for segment in segments.values():
if segment['count'] > 0:
segment['avg_sentiment'] /= segment['count']
segment['avg_comments_per_user'] = segment['total_comments'] / segment['count']
# 生成建议
recommendations = []
if '忠实粉丝' in segments:
rec = f"你有{segments['忠实粉丝']['count']}位忠实粉丝，他们非常喜欢你的内容"
if segments['忠实粉丝']['example_users']:
rec += f"，例如: {', '.join([u['nickname'] for u in segments['忠实粉丝']['example_users']])}"
recommendations.append(rec)
if '批评者' in segments and segments['批评者']['count'] > 0:
recommendations.append(f"有{segments['批评者']['count']}位用户经常提出批评，建议关注他们的反馈")
if '高频互动者' in segments:
recommendations.append(f"{segments['高频互动者']['count']}位高频互动者是你的核心观众群体")
return {
'segments': segments,
'recommendations': recommendations,
'total_users': len(profiles)
}
# 使用示例
profile_builder = UserProfileBuilder()
# 分析单个视频的评论者
video_users = profile_builder.analyze_comment_authors("视频ID", min_comments=1)
if video_users:
print(f"=== 视频评论用户分析 ===")
print(f"总评论用户数: {video_users['overall_stats']['total_users']}")
p
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
1
0- 0
已为社区贡献6条内容
所有评论(0)