最近在整理虚拟歌手相关数据时，发现很多朋友对如何系统性地追踪和分析虚拟歌手（如初音未来、洛天依等）的外语作品热度很感兴趣。网上资料往往零散，缺乏一个从数据获取、处理到可视化排行的完整技术方案。本文将分享一套基于 Python 的自动化解决方案，手把手教你构建一个类似“周刊虚拟歌手外语排行榜”的数据分析系统。无论你是想学习数据爬虫、数据分析，还是想为自己的兴趣项目添砖加瓦，这套代码和思路都能直接复用。

1. 项目背景与核心概念

虚拟歌手外语排行榜 是什么？它通常指定期（如每周）对以初音未来、镜音双子、洛天依等虚拟歌手演唱的非母语（例如日语虚拟歌手演唱中文、英文歌曲，或中文虚拟歌手演唱日文歌曲）作品，在特定平台（如B站、Niconico、YouTube）上的播放量、评论数、收藏数等数据进行采集、加权计算，并生成的热度排名榜单。

这类榜单对于粉丝发现优质作品、观察流行趋势非常有价值。然而，手动统计费时费力，且难以保证客观性和持续性。因此，一个自动化的数据抓取与处理系统就显得尤为重要。

本项目的核心目标 是构建一个自动化流程，模拟实现“周刊虚拟歌手外语排行榜”的生成。我们将重点关注以下几个技术环节：

1. 数据源确定与采集 ：如何从视频平台获取结构化数据。
2. 数据清洗与处理 ：如何过滤出“外语作品”，并进行指标计算。
3. 排名算法设计 ：如何综合多个指标得出一个公平的热度分。
4. 结果持久化与展示 ：如何存储历史数据并生成可读的报告。

技术栈 ：本项目主要使用 Python，涉及 requests / selenium 用于网络请求， pandas 进行数据处理， sqlite3 进行数据存储，以及 matplotlib / pyecharts 进行可视化。

2. 环境准备与版本说明

在开始编码前，请确保你的开发环境已就绪。以下版本为本文撰写时的常用版本，实际操作时可根据情况调整。

• 操作系统 ：Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04+)。
• Python 版本 ：3.8 或以上。推荐使用 3.9+ 以获得更好的兼容性。
• 开发工具 ：任意你熟悉的 IDE 或编辑器，如 PyCharm, VSCode, Jupyter Notebook。
• 项目依赖库 ：我们将通过 requirements.txt 文件管理。

首先，创建一个新的项目目录，例如 virtual_singer_rank ，并在其中创建 requirements.txt 文件。

# requirements.txt
requests>=2.28.0
pandas>=1.5.0
beautifulsoup4>=4.11.0
selenium>=4.10.0
webdriver-manager>=4.0.0
sqlalchemy>=2.0.0
pyecharts>=2.0.0
matplotlib>=3.5.0
schedule>=1.2.0
lxml>=4.9.0

然后，在终端中进入项目目录，安装依赖：

cd /path/to/your/virtual_singer_rank
pip install -r requirements.txt

注意 ： selenium 用于模拟浏览器操作，应对反爬较严格的网站。你需要根据自己使用的浏览器（Chrome/Firefox）安装对应的 WebDriver。 webdriver-manager 库可以帮你自动管理驱动，简化流程。

3. 核心模块设计与原理拆解

我们将系统拆分为四个核心模块，便于理解和维护。

3.1 数据采集模块 (Crawler)

此模块负责从目标网站抓取视频列表和每个视频的详细数据。以B站为例，我们可以通过搜索特定关键词（如“初音未来 英语”）来获取视频列表。

原理 ：

1. 分析目标网站（如B站搜索页、排行榜页）的URL结构和返回的数据格式（通常是JSON）。
2. 使用 requests 库发送HTTP请求，并解析返回的JSON数据，这是效率最高的方式。
3. 如果网站数据通过JavaScript动态加载， requests 无法直接获取，则使用 selenium 模拟浏览器行为，等待页面加载完成后，再提取数据。

关键点 ：

• 遵守 robots.txt ：在爬取前，务必检查目标网站的 robots.txt 文件，尊重网站的爬虫协议。
• 设置请求头 ：模拟真实浏览器访问，避免被简单的反爬机制拦截。
• 添加延迟 ：在请求间添加随机延时（如 time.sleep(random.uniform(1, 3)) ），减轻服务器压力，也是良好的网络公民行为。
• 错误处理 ：网络请求可能失败，必须使用 try...except 包裹，并设计重试机制。

3.2 数据清洗与过滤模块 (DataProcessor)

原始数据包含很多无关信息。此模块负责：

1. 提取关键字段 ：如视频标题（ title ）、AV/BV号（ id ）、播放量（ view ）、弹幕数（ danmaku ）、收藏数（ favorite ）、投币数（ coin ）、分享数（ share ）、发布时间（ pubdate ）。
2. 过滤外语作品 ：这是本项目难点。可以通过规则判断：
   • 关键词过滤 ：在标题、标签或简介中搜索“英语”、“英文”、“English”、“Chinese”、“中文”等字样。
   • 语言检测库 ：使用 langdetect 库对标题或简介进行语言识别（注意准确率问题）。
   • 手动标记+学习 ：初期可结合少量手动标记，后期尝试简单的机器学习模型（如基于标题文本的分类）。
3. 数据格式化 ：将字符串数字（如“2.3万”）转换为整数（23000），统一时间格式。

3.3 排名算法模块 (RankingAlgorithm)

如何将播放、收藏、投币等不同量纲、不同权重的指标合成一个“热度分”？

常用方法 ：

1. 标准化（归一化） ：将每个指标缩放到[0,1]区间。例如， 播放量得分 = (当前视频播放量 - 最小播放量) / (最大播放量 - 最小播放量) 。
2. 加权求和 ：为不同指标赋予权重。例如，认为“收藏”比“播放”更能体现作品质量，赋予更高权重。

   热度分 = w1 * 标准化播放量 + w2 * 标准化弹幕数 + w3 * 标准化收藏数 + w4 * 标准化投币数 + w5 * 标准化分享数
   

   其中， w1 + w2 + w3 + w4 + w5 = 1 。权重需要根据业务理解调整，初期可以设为均等。
3. 时间衰减 ：考虑视频发布时间，新视频可能获得一定加成，或者对长期榜单引入时间衰减因子，让榜单能反映近期热度。

3.4 数据存储与展示模块 (Storage & Visualization)

• 存储 ：使用轻量级数据库 SQLite 存储每期排行榜数据。表结构可设计为：
  • rank_weekly ： id （主键）， period （期数，如 2026W28 ）， rank （名次）， video_id ， title ， singer ， language ， score ， view ， favorite ， pubdate ， crawl_time 。
• 展示 ：
  • 控制台打印 ：简单的文本表格。
  • 生成图表 ：使用 pyecharts 生成交互式HTML图表，展示TOP10视频热度对比、各虚拟歌手上榜数量等。
  • 生成Markdown报告 ：自动生成包含榜单、图表链接的周报文档。

4. 完整实战案例：构建第97期排行榜

假设我们要生成“2026年7月第2期”（即2026年的第28周）的榜单。下面我们分步骤实现。

4.1 项目结构创建

创建如下目录和文件：

virtual_singer_rank/
├── config.py          # 配置文件
├── crawler.py         # 数据采集模块
├── processor.py       # 数据清洗过滤模块
├── ranker.py          # 排名算法模块
├── storage.py         # 数据存储模块
├── visualizer.py      # 可视化模块
├── main.py            # 主程序入口
├── requirements.txt   # 依赖列表
├── data/              # 数据目录
│   └── database.db    # SQLite数据库
└── output/            # 输出目录（图表、报告）

4.2 编写配置文件

config.py 集中管理常量，如搜索关键词、API地址、权重、数据库路径等。

# config.py
class Config:
    # 搜索关键词 (虚拟歌手 + 语言)
    SEARCH_KEYWORDS = [
        "初音未来 英语",
        "初音ミク English",
        "洛天依 日语",
        "镜音リン Chinese",
        "巡音ルカ 英语",
        "GUMI 中文"
    ]
    
    # 数据源平台 (示例用B站)
    PLATFORM = "bilibili"
    # B站搜索API (示例，实际需根据最新情况调整)
    BILI_SEARCH_API = "https://api.bilibili.com/x/web-interface/search/all/v2"
    
    # 请求头
    HEADERS = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
        'Referer': 'https://www.bilibili.com/'
    }
    
    # 排名算法权重 (播放, 弹幕, 收藏, 投币, 分享)
    WEIGHTS = [0.3, 0.15, 0.25, 0.2, 0.1]
    
    # 数据库路径
    DB_PATH = 'data/database.db'
    
    # 本期期数标识
    CURRENT_PERIOD = '2026W28'  # 2026年第28周

4.3 实现数据采集模块

crawler.py 使用 requests 调用B站搜索接口。 请注意 ：公开API可能有频率限制，且接口格式可能变更，以下代码主要为示例逻辑。

# crawler.py
import requests
import time
import random
from config import Config

class BilibiliCrawler:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update(Config.HEADERS)
        
    def search_videos(self, keyword, page=1, page_size=20):
        """根据关键词搜索视频"""
        params = {
            'keyword': keyword,
            'page': page,
            'page_size': page_size,
            # 其他必要参数，可能需要从网页分析
            'search_type': 'video',
            'order': 'click', # 按播放量排序
        }
        try:
            resp = self.session.get(Config.BILI_SEARCH_API, params=params, timeout=10)
            resp.raise_for_status() # 检查HTTP错误
            data = resp.json()
            # 解析返回的JSON，提取视频列表
            video_list = self._parse_search_result(data)
            # 礼貌性延迟
            time.sleep(random.uniform(1, 2))
            return video_list
        except requests.exceptions.RequestException as e:
            print(f"搜索关键词 '{keyword}' 时出错: {e}")
            return []
    
    def _parse_search_result(self, data):
        """解析B站搜索API返回的JSON数据"""
        videos = []
        # 注意：此解析逻辑强烈依赖于API返回的实际结构，需要根据实际情况调整
        if data.get('code') == 0:
            # 假设结果在 data['data']['result'] 中
            for item in data.get('data', {}).get('result', []):
                if item.get('result_type') == 'video':
                    video_info = item.get('data', {})
                    video = {
                        'id': video_info.get('bvid'), # BV号
                        'title': video_info.get('title', '').replace('<em class=\"keyword\">', '').replace('</em>', ''),
                        'author': video_info.get('author', ''),
                        'view': self._parse_count(video_info.get('play', '0')),
                        'danmaku': self._parse_count(video_info.get('danmaku', '0')),
                        'favorite': self._parse_count(video_info.get('favorite', '0')),
                        'coin': self._parse_count(video_info.get('coin', '0')),
                        'share': self._parse_count(video_info.get('share', '0')),
                        'pubdate': video_info.get('pubdate', 0), # 时间戳
                        'keyword': 'placeholder' # 后续会填充
                    }
                    videos.append(video)
        return videos
    
    def _parse_count(self, count_str):
        """将‘2.3万’这样的字符串转换为整数"""
        if not isinstance(count_str, str):
            return int(count_str) if count_str else 0
        if '万' in count_str:
            return int(float(count_str.replace('万', '')) * 10000)
        try:
            return int(count_str)
        except ValueError:
            return 0

# 示例：抓取一个关键词
if __name__ == '__main__':
    crawler = BilibiliCrawler()
    test_results = crawler.search_videos(Config.SEARCH_KEYWORDS[0], page=1)
    print(f"抓取到 {len(test_results)} 条结果")
    for v in test_results[:2]:
        print(v['title'], v['view'])

4.4 实现数据清洗与过滤模块

processor.py 负责清洗和过滤。

# processor.py
import re
from langdetect import detect, DetectorFactory, LangDetectException
# 为了确定性结果，设置种子
DetectorFactory.seed = 0

class DataProcessor:
    def __init__(self):
        # 定义语言关键词 (用于初步过滤)
        self.language_keywords = {
            'english': ['英语', '英文', 'eng', 'english', '英翻', '英配'],
            'japanese': ['日语', '日文', 'jp', 'japanese', '日翻'],
            'chinese': ['中文', '汉语', 'chinese', '中翻', '国语']
        }
        # 虚拟歌手列表 (用于识别)
        self.singer_list = ['初音未来', '初音ミク', '洛天依', '镜音リン', '镜音レン', '巡音ルカ', 'GUMI', 'IA', 'MEIKO', 'KAITO', '言和', '乐正绫']
    
    def filter_by_language(self, video_item, title_key='title'):
        """基于标题关键词和语言检测过滤外语作品"""
        title = video_item.get(title_key, '').lower()
        
        # 1. 关键词匹配
        detected_langs = []
        for lang, keywords in self.language_keywords.items():
            for kw in keywords:
                if kw in title:
                    detected_langs.append(lang)
                    break # 找到该语言的一个关键词就跳出
        
        # 2. 语言检测 (作为补充，可能不准)
        try:
            lang_detected = detect(title)
            # 将检测结果映射到我们的分类
            if lang_detected == 'en' and 'english' not in detected_langs:
                detected_langs.append('english')
            elif lang_detected == 'ja' and 'japanese' not in detected_langs:
                detected_langs.append('japanese')
            elif lang_detected == 'zh-cn' or lang_detected == 'zh-tw':
                if 'chinese' not in detected_langs:
                    detected_langs.append('chinese')
        except LangDetectException:
            pass
        
        # 3. 判断是否为“外语”作品
        # 逻辑：如果视频标题中包含某语言关键词，但该语言不是歌手的“主要”语言，则算外语。
        # 简化：我们假设主要语言是日语（对于日系V家）和中文（对于中文V家）。
        # 这是一个非常简化的模型，实际应用需要更复杂的规则或模型。
        singer = self._identify_singer(title)
        if singer:
            # 简单判断：如果歌手是日系，且标题包含中文或英文关键词，则可能是外语歌。
            if any(s in singer for s in ['初音', 'ミク', 'リン', 'レン', 'ルカ', 'MEIKO', 'KAITO', 'GUMI', 'IA']):
                if 'chinese' in detected_langs or 'english' in detected_langs:
                    video_item['language'] = 'chinese' if 'chinese' in detected_langs else 'english'
                    video_item['singer'] = singer
                    return True
            # 如果歌手是中文系，且标题包含日文或英文关键词
            elif any(s in singer for s in ['洛天依', '言和', '乐正绫']):
                if 'japanese' in detected_langs or 'english' in detected_langs:
                    video_item['language'] = 'japanese' if 'japanese' in detected_langs else 'english'
                    video_item['singer'] = singer
                    return True
        return False
    
    def _identify_singer(self, title):
        """从标题中识别虚拟歌手"""
        for singer in self.singer_list:
            if singer in title:
                return singer
        return None
    
    def clean_data(self, raw_video_list):
        """清洗数据：去重、转换类型、添加标识"""
        cleaned_list = []
        seen_ids = set()
        
        for video in raw_video_list:
            vid = video.get('id')
            if not vid or vid in seen_ids:
                continue
            seen_ids.add(vid)
            
            # 确保关键字段存在
            video['view'] = int(video.get('view', 0))
            video['danmaku'] = int(video.get('danmaku', 0))
            video['favorite'] = int(video.get('favorite', 0))
            video['coin'] = int(video.get('coin', 0))
            video['share'] = int(video.get('share', 0))
            # 时间戳转可读日期 (可选)
            # from datetime import datetime
            # video['pubdate_str'] = datetime.fromtimestamp(video['pubdate']).strftime('%Y-%m-%d') if video['pubdate'] else ''
            
            cleaned_list.append(video)
        return cleaned_list

# 示例用法
if __name__ == '__main__':
    processor = DataProcessor()
    test_video = {'id': 'BV1xx411c7XX', 'title': '【初音未来】Perfect Night (英文填词翻唱)【英语】', 'view': '12.5万'}
    test_video['view'] = processor._parse_count(test_video['view'])
    if processor.filter_by_language(test_video):
        print(f"识别为外语作品: {test_video['title']}, 语言: {test_video.get('language')}, 歌手: {test_video.get('singer')}")

4.5 实现排名算法模块

ranker.py 实现加权评分算法。

# ranker.py
import pandas as pd
import numpy as np
from config import Config

class RankCalculator:
    def __init__(self, weights=None):
        self.weights = weights or Config.WEIGHTS
        # 确保权重和为1
        if abs(sum(self.weights) - 1.0) > 0.001:
            self.weights = [w/sum(self.weights) for w in self.weights]
    
    def calculate_score(self, video_df):
        """
        计算视频热度综合得分
        video_df 应包含列：view, danmaku, favorite, coin, share
        """
        # 复制一份，避免修改原数据
        df = video_df.copy()
        metrics = ['view', 'danmaku', 'favorite', 'coin', 'share']
        
        # 检查所需列是否存在
        for m in metrics:
            if m not in df.columns:
                raise ValueError(f"DataFrame 中缺少必要列: {m}")
        
        # 1. 数据标准化 (Min-Max Scaling)
        for col in metrics:
            col_min = df[col].min()
            col_max = df[col].max()
            # 防止除零
            if col_max - col_min > 0:
                df[f'{col}_norm'] = (df[col] - col_min) / (col_max - col_min)
            else:
                df[f'{col}_norm'] = 0.0
        
        # 2. 加权求和
        weighted_cols = [f'{m}_norm' for m in metrics]
        df['score'] = np.dot(df[weighted_cols].values, self.weights)
        
        # 3. 按得分降序排序
        df = df.sort_values(by='score', ascending=False).reset_index(drop=True)
        df['rank'] = df.index + 1
        
        # 4. 选择需要返回的列
        result_cols = ['rank', 'id', 'title', 'singer', 'language', 'score', 'view', 'danmaku', 'favorite', 'coin', 'share', 'pubdate']
        # 确保列存在
        available_cols = [col for col in result_cols if col in df.columns]
        return df[available_cols]

# 示例：创建测试数据并计算排名
if __name__ == '__main__':
    # 模拟10个视频的数据
    np.random.seed(42)
    test_data = {
        'id': [f'BV{10000+i}' for i in range(10)],
        'title': [f'测试视频{i}' for i in range(10)],
        'view': np.random.randint(1000, 1000000, 10),
        'danmaku': np.random.randint(10, 5000, 10),
        'favorite': np.random.randint(100, 50000, 10),
        'coin': np.random.randint(50, 20000, 10),
        'share': np.random.randint(10, 5000, 10),
    }
    test_df = pd.DataFrame(test_data)
    test_df['singer'] = ['初音未来']*5 + ['洛天依']*5
    test_df['language'] = ['english', 'chinese']*5
    
    ranker = RankCalculator()
    ranked_df = ranker.calculate_score(test_df)
    print(ranked_df[['rank', 'id', 'title', 'score', 'view']].head())

4.6 实现数据存储模块

storage.py 使用 SQLAlchemy 与 SQLite 交互，便于操作。

# storage.py
import pandas as pd
from sqlalchemy import create_engine, Table, Column, Integer, String, Float, DateTime, MetaData
from sqlalchemy.exc import SQLAlchemyError
from datetime import datetime
from config import Config

class RankStorage:
    def __init__(self, db_path=None):
        self.db_path = db_path or Config.DB_PATH
        # 创建数据库引擎
        self.engine = create_engine(f'sqlite:///{self.db_path}', echo=False)
        self.metadata = MetaData()
        self._define_table()
        # 创建表（如果不存在）
        self.metadata.create_all(self.engine)
    
    def _define_table(self):
        """定义排行榜数据表结构"""
        self.weekly_rank_table = Table(
            'weekly_rank', self.metadata,
            Column('id', Integer, primary_key=True, autoincrement=True),
            Column('period', String(20), nullable=False), # 如 '2026W28'
            Column('rank', Integer, nullable=False),
            Column('video_id', String(20), nullable=False),
            Column('title', String(500)),
            Column('singer', String(50)),
            Column('language', String(20)),
            Column('score', Float),
            Column('view', Integer),
            Column('danmaku', Integer),
            Column('favorite', Integer),
            Column('coin', Integer),
            Column('share', Integer),
            Column('pubdate', Integer), # 存储时间戳
            Column('crawl_time', DateTime, default=datetime.now),
            # 可以添加联合索引以提高查询性能
        )
    
    def save_weekly_rank(self, period, ranked_df):
        """将一期排行榜数据保存到数据库"""
        if ranked_df.empty:
            print("排行榜数据为空，未保存。")
            return False
        # 添加期数字段
        ranked_df['period'] = period
        ranked_df['crawl_time'] = datetime.now()
        
        try:
            # 使用 pandas 的 to_sql 方法，如果表存在则追加
            ranked_df.to_sql('weekly_rank', self.engine, if_exists='append', index=False)
            print(f"成功保存第 {period} 期排行榜数据，共 {len(ranked_df)} 条记录。")
            return True
        except SQLAlchemyError as e:
            print(f"保存数据到数据库时出错: {e}")
            return False
    
    def load_weekly_rank(self, period=None):
        """从数据库加载排行榜数据，默认加载最新一期"""
        try:
            with self.engine.connect() as conn:
                if period:
                    query = f"SELECT * FROM weekly_rank WHERE period = '{period}' ORDER BY rank ASC"
                else:
                    # 加载最新一期
                    subquery = "SELECT MAX(period) as latest_period FROM weekly_rank"
                    query = f"""
                        SELECT wr.* FROM weekly_rank wr
                        JOIN ({subquery}) lp ON wr.period = lp.latest_period
                        ORDER BY wr.rank ASC
                    """
                df = pd.read_sql_query(query, conn)
                return df
        except SQLAlchemyError as e:
            print(f"从数据库加载数据时出错: {e}")
            return pd.DataFrame()

# 示例：保存和加载数据
if __name__ == '__main__':
    storage = RankStorage('data/test.db')
    # 假设 ranked_df 是上一节计算出的 DataFrame
    # storage.save_weekly_rank('2026W28', ranked_df)
    # loaded_df = storage.load_weekly_rank('2026W28')
    # print(loaded_df.head())

4.7 实现可视化模块

visualizer.py 使用 pyecharts 生成交互式图表。

# visualizer.py
from pyecharts import options as opts
from pyecharts.charts import Bar, Pie, Page
import pandas as pd
import os

class RankVisualizer:
    def __init__(self, output_dir='output'):
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)
    
    def create_top10_bar(self, ranked_df, period):
        """生成TOP10视频热度得分柱状图"""
        top10 = ranked_df.head(10)
        # 简化标题用于显示
        titles = [f"{row['rank']}. {row['title'][:20]}..." if len(row['title']) > 20 else f"{row['rank']}. {row['title']}" for _, row in top10.iterrows()]
        scores = top10['score'].round(4).tolist()
        
        bar = (
            Bar()
            .add_xaxis(titles)
            .add_yaxis("热度得分", scores,
                       label_opts=opts.LabelOpts(position="right", formatter="{c}"))
            .reversal_axis()
            .set_global_opts(
                title_opts=opts.TitleOpts(title=f"{period} 虚拟歌手外语榜 TOP10", subtitle="综合热度得分"),
                xaxis_opts=opts.AxisOpts(name="热度得分"),
                yaxis_opts=opts.AxisOpts(name="视频", axislabel_opts=opts.LabelOpts(rotate=0)),
                tooltip_opts=opts.TooltipOpts(trigger="axis", axis_pointer_type="shadow"),
            )
        )
        return bar
    
    def create_singer_pie(self, ranked_df, period):
        """生成上榜虚拟歌手分布饼图"""
        singer_counts = ranked_df['singer'].value_counts()
        singers = singer_counts.index.tolist()
        counts = singer_counts.values.tolist()
        
        pie = (
            Pie()
            .add("", [list(z) for z in zip(singers, counts)],
                 radius=["30%", "60%"])
            .set_global_opts(
                title_opts=opts.TitleOpts(title=f"{period} 上榜歌手分布"),
                legend_opts=opts.LegendOpts(orient="vertical", pos_top="15%", pos_left="2%"),
            )
            .set_series_opts(label_opts=opts.LabelOpts(formatter="{b}: {c} ({d}%)"))
        )
        return pie
    
    def generate_report(self, ranked_df, period):
        """生成包含图表的HTML报告"""
        page = Page(layout=Page.SimplePageLayout)
        page.add(
            self.create_top10_bar(ranked_df, period),
            self.create_singer_pie(ranked_df, period)
        )
        report_path = os.path.join(self.output_dir, f'rank_report_{period}.html')
        page.render(report_path)
        print(f"可视化报告已生成: {report_path}")
        return report_path

# 示例：生成报告
if __name__ == '__main__':
    # 假设 ranked_df 是已有的排行榜 DataFrame
    # visualizer = RankVisualizer()
    # visualizer.generate_report(ranked_df, '2026W28')
    pass

4.8 编写主程序入口

main.py 将以上所有模块串联起来，形成完整工作流。

# main.py
import pandas as pd
from config import Config
from crawler import BilibiliCrawler
from processor import DataProcessor
from ranker import RankCalculator
from storage import RankStorage
from visualizer import RankVisualizer
import time

def main():
    print(f"开始生成虚拟歌手外语排行榜：{Config.CURRENT_PERIOD}")
    print("="*50)
    
    # 1. 初始化各模块
    crawler = BilibiliCrawler()
    processor = DataProcessor()
    ranker = RankCalculator()
    storage = RankStorage()
    visualizer = RankVisualizer()
    
    all_videos = []
    
    # 2. 遍历关键词进行搜索和抓取
    print("步骤1：数据采集...")
    for keyword in Config.SEARCH_KEYWORDS:
        print(f"  正在抓取关键词: {keyword}")
        # 这里为了示例只抓取第一页，实际可以抓取多页
        raw_videos = crawler.search_videos(keyword, page=1, page_size=30)
        for v in raw_videos:
            v['keyword'] = keyword
        all_videos.extend(raw_videos)
        time.sleep(2) # 更友好的延迟
    
    print(f"  共抓取到 {len(all_videos)} 条原始数据。")
    
    # 3. 数据清洗
    print("步骤2：数据清洗与过滤...")
    cleaned_videos = processor.clean_data(all_videos)
    print(f"  去重后剩余 {len(cleaned_videos)} 条数据。")
    
    # 4. 过滤外语作品
    print("步骤3：过滤外语作品...")
    foreign_lang_videos = []
    for video in cleaned_videos:
        if processor.filter_by_language(video):
            foreign_lang_videos.append(video)
    
    print(f"  识别出 {len(foreign_lang_videos)} 首外语作品。")
    if not foreign_lang_videos:
        print("未找到足够的外语作品，程序终止。")
        return
    
    # 5. 转换为DataFrame并计算排名
    print("步骤4：计算综合热度排名...")
    df = pd.DataFrame(foreign_lang_videos)
    # 确保必要的数值列存在
    for col in ['view', 'danmaku', 'favorite', 'coin', 'share']:
        if col not in df.columns:
            df[col] = 0
    ranked_df = ranker.calculate_score(df)
    
    # 6. 保存到数据库
    print("步骤5：保存数据到数据库...")
    storage.save_weekly_rank(Config.CURRENT_PERIOD, ranked_df)
    
    # 7. 生成可视化报告
    print("步骤6：生成可视化报告...")
    report_path = visualizer.generate_report(ranked_df, Config.CURRENT_PERIOD)
    
    # 8. 打印本期榜单TOP10
    print("\n" + "="*50)
    print(f"{Config.CURRENT_PERIOD} 虚拟歌手外语排行榜 TOP10")
    print("="*50)
    top10_display = ranked_df.head(10)[['rank', 'title', 'singer', 'language', 'score', 'view']].copy()
    top10_display['view'] = top10_display['view'].apply(lambda x: f'{x:,}')
    print(top10_display.to_string(index=False))
    
    print(f"\n榜单生成完毕！详细报告请查看: {report_path}")

if __name__ == '__main__':
    main()

4.9 运行与验证

在项目根目录下运行主程序：

python main.py

如果一切顺利，你将在控制台看到抓取、过滤、计算、保存的日志，并最终输出一个TOP10的文本榜单。同时，在 output 目录下会生成一个名为 rank_report_2026W28.html 的文件，用浏览器打开即可看到交互式图表。

预期输出示例 ：

开始生成虚拟歌手外语排行榜：2026W28
==================================================
步骤1：数据采集...
  正在抓取关键词: 初音未来 英语
  正在抓取关键词: 初音ミク English
...
  共抓取到 187 条原始数据。
步骤2：数据清洗与过滤...
  去重后剩余 162 条数据。
步骤3：过滤外语作品...
  识别出 45 首外语作品。
步骤4：计算综合热度排名...
步骤5：保存数据到数据库...
成功保存第 2026W28 期排行榜数据，共 45 条记录。
步骤6：生成可视化报告...
可视化报告已生成: output/rank_report_2026W28.html

==================================================
2026W28 虚拟歌手外语排行榜 TOP10
==================================================
 rank                                  title      singer language     score     view
    1      【初音未来】Perfect Night (英文填词)   初音未来  english  0.8923  1,234,567
    2        【洛天依】Lemon (日语翻唱)【中日字幕】     洛天依  japanese  0.8567    987,654
    3  【镜音リン】Bad Apple!! (英语版)【PV付】   镜音リン  english  0.8121    876,543
...
榜单生成完毕！详细报告请查看: output/rank_report_2026W28.html

5. 常见问题与排查思路

在实际运行中，你可能会遇到以下问题：

问题现象	可能原因	解决思路
运行 main.py 报 ModuleNotFoundError	依赖库未安装	检查是否在项目目录下执行了 pip install -r requirements.txt 。
抓取数据返回空列表或状态码非200	1. 网站API变更 2. 请求头被识别为爬虫 3. 关键词搜索无结果	1. 使用浏览器开发者工具（F12）重新分析网络请求，更新 crawler.py 中的API URL和参数。 2. 更新 config.py 中的 HEADERS ，模拟更真实的浏览器。 3. 尝试更换搜索关键词或检查网络连接。
语言过滤结果不准确，漏掉或误判很多	1. 关键词列表不完善 2. 语言检测库 ( langdetect ) 对短文本不准 3. 规则过于简单	1. 扩充 config.py 和 processor.py 中的关键词和歌手列表。 2. 考虑使用更准确的语言识别API（如Google Cloud Translation），但需注意调用成本和频率限制。 3. 引入更复杂的规则，或尝试训练一个简单的文本分类模型。
排名得分看起来不合理（如播放量极高但得分低）	权重设置不合适或数据标准化出现问题	1. 检查 config.py 中的 WEIGHTS ，根据业务理解调整（例如，提高“收藏”的权重）。 2. 检查 ranker.py 中的标准化计算，确保没有除零错误。可以打印中间结果 df[weighted_cols] 进行调试。
数据库保存失败	1. 数据库文件路径不可写 2. 表结构不匹配	1. 检查 data/ 目录是否存在且有写入权限。 2. 如果修改了表结构，可能需要删除旧的 .db 文件重新运行，或使用数据库迁移工具。
生成的HTML图表无法显示或空白	1. PyEcharts版本问题 2. 数据为空或格式错误	1. 确保安装的 pyecharts 版本正确。生成的HTML文件需要用浏览器打开。 2. 检查传递给可视化模块的 ranked_df 是否为空或包含必要的列（如 score , singer ）。

6. 最佳实践与工程建议

将脚本升级为可维护、可扩展的工程化项目，需要考虑以下几点：

1. 配置化管理 ：将所有可调参数（如搜索关键词、权重、API地址、数据库连接字符串）放入配置文件（如 config.yaml ）或环境变量中，避免硬编码。
2. 日志记录 ：使用 Python 的 logging 模块替代 print ，可以输出不同级别（INFO, WARNING, ERROR）的日志到文件和控制台，便于后期排查问题。
3. 异常处理与重试 ：在网络请求、数据库操作等可能失败的环节，实现完整的异常捕获和重试机制（例如使用 tenacity 库）。
4. 反爬策略应对 ：
   • 使用代理IP池 ：频繁请求单一IP容易被封，可以集成代理服务。
   • 模拟用户行为 ：除了请求头，还可以使用 selenium 模拟点击、滚动等操作，但效率较低。
   • 遵守法律法规与协议 ：严格控制爬取频率，不对目标网站造成压力。仅用于个人学习与研究。
5. 数据更新自动化 ：使用 schedule 库或操作系统的定时任务（如 cron 或 Windows Task Scheduler ）定期（如每周一）运行主脚本，实现榜单自动更新。
6. 代码模块化与测试 ：保持当前模块化的设计。为关键函数编写单元测试（使用 pytest ），确保数据过滤、评分计算等核心逻辑的正确性。
7. 前端展示优化 ：除了生成静态HTML报告，可以考虑使用轻量级Web框架（如 Flask 或 Streamlit ）搭建一个简单的Web应用，动态展示历史榜单和趋势图。
8. 数据备份 ：定期备份 SQLite 数据库文件。如果数据量增大，可以考虑迁移到 PostgreSQL 或 MySQL 等更专业的数据库。
9. 算法持续优化 ：热度算法是核心。可以收集用户反馈（如点赞、点踩），引入更复杂的模型（如考虑视频发布时间衰减因子 score = base_score / (log(time_diff + 1) + 1) ），让榜单更科学。

通过以上步骤，你不仅完成了一个具体的“虚拟歌手外语排行榜”项目，更掌握了一套通用的数据采集、处理、分析和可视化的Python实战流程。这套框架经过简单修改，便可应用于其他领域的榜单生成或数据分析任务，例如游戏热度榜、开源项目趋势榜等。