如何写短视频链接解析脚本

本文目录导读：

1. 基本思路
2. 完整示例代码
3. 增强版脚本（包含更多功能）
4. 命令行工具版本
5. 重要说明

我来为你详细讲解如何编写一个短视频链接解析脚本,并提供完整的示例代码。

基本思路

短视频链接解析主要包含以下步骤：

1. 获取分享链接：从用户输入或剪贴板获取
2. 解析链接类型：识别来自哪个平台（抖音、快手、B站等）
3. 提取视频ID：从URL中提取视频唯一标识
4. 获取视频信息：调用对应平台的API或网页解析
5. 返回解析结果：返回视频信息（标题、封面、下载地址等）

完整示例代码

以下是一个支持多平台的通用解析脚本：

import re
import requests
import json
from urllib.parse import urlparse, parse_qs
import html
class ShortVideoParser:
    # 平台识别规则
    PLATFORM_PATTERNS = {
        'douyin': [
            r'douyin\.com',
            r'iesdouyin\.com',
            r'www\.douyin\.com/video/',
            r'v\.douyin\.com/'
        ],
        'kuaishou': [
            r'kuaishou\.com',
            r'www\.kuaishou\.com/video/',
            r'v\.kuaishou\.com/'
        ],
        'bilibili': [
            r'bilibili\.com/video/',
            r'b23\.tv/'
        ],
        'weibo': [
            r'weibo\.com/\d+/',
            r'weibo\.tv/v/'
        ],
        'xiaohongshu': [
            r'xiaohongshu\.com/discovery/item/',
            r'xhslink\.com/'
        ]
    }
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        })
    def identify_platform(self, url):
        """识别视频来源平台"""
        for platform, patterns in self.PLATFORM_PATTERNS.items():
            for pattern in patterns:
                if re.search(pattern, url):
                    return platform
        return 'unknown'
    def extract_video_id(self, url, platform):
        """提取视频ID"""
        if platform == 'douyin':
            # 处理短链接
            if 'v.douyin.com' in url:
                response = requests.head(url, allow_redirects=True)
                url = response.url
            # 提取视频ID
            match = re.search(r'/video/(\d+)', url)
            return match.group(1) if match else None
        elif platform == 'kuaishou':
            # 处理短链接
            if 'v.kuaishou.com' in url:
                response = requests.head(url, allow_redirects=True)
                url = response.url
            # 提取视频ID
            match = re.search(r'/video/(\d+)', url)
            return match.group(1) if match else None
        elif platform == 'bilibili':
            # 提取BV号
            match = re.search(r'/video/(BV\w+)', url)
            return match.group(1) if match else None
        elif platform == 'xiaohongshu':
            # 处理短链接
            if 'xhslink.com' in url:
                response = requests.head(url, allow_redirects=True)
                url = response.url
            # 提取笔记ID
            match = re.search(r'/discovery/item/([a-f0-9]+)', url)
            return match.group(1) if match else None
        return url  # 如果无法提取，返回原始URL
    def parse_douyin(self, video_id):
        """解析抖音视频"""
        try:
            # 方法1：使用官方接口（需要cookie）
            url = f"https://www.iesdouyin.com/share/video/{video_id}/"
            response = self.session.get(url)
            # 从HTML中提取视频信息
            # 这里使用正则提取JSON数据
            pattern = r'<script id="RENDER_DATA" type="application/json">(.*?)</script>'
            match = re.search(pattern, response.text)
            if match:
                json_str = html.unescape(match.group(1))
                data = json.loads(json_str)
                # 解析视频信息（具体路径根据实际返回数据调整）
                video_info = {
                    'title': data.get('title', ''),
                    'cover': data.get('cover', ''),
                    'video_url': data.get('video', {}).get('url', ''),
                    'author': data.get('author', ''),
                    'platform': 'douyin'
                }
                return video_info
            # 方法2：使用第三方接口
            # api_url = f"https://api.douyin.wtf/api/video/{video_id}"
            # response = self.session.get(api_url)
            # return response.json()
        except Exception as e:
            print(f"解析抖音视频失败: {e}")
            return None
    def parse_kuaishou(self, video_id):
        """解析快手视频"""
        try:
            url = f"https://www.kuaishou.com/video/{video_id}"
            response = self.session.get(url)
            # 从页面中提取视频信息
            pattern = r'window.__INITIAL_STATE__\s*=\s*({.*?});'
            match = re.search(pattern, response.text, re.DOTALL)
            if match:
                data = json.loads(match.group(1))
                video_info = {
                    'title': data.get('video', {}).get('name', ''),
                    'cover': data.get('video', {}).get('poster', ''),
                    'video_url': data.get('video', {}).get('src', ''),
                    'author': data.get('user', {}).get('name', ''),
                    'platform': 'kuaishou'
                }
                return video_info
        except Exception as e:
            print(f"解析快手视频失败: {e}")
            return None
    def parse_bilibili(self, video_id):
        """解析B站视频"""
        try:
            # B站官方API
            api_url = f"https://api.bilibili.com/x/web-interface/view?bvid={video_id}"
            response = self.session.get(api_url)
            data = response.json()
            if data['code'] == 0:
                video_data = data['data']
                video_info = {
                    'title': video_data.get('title', ''),
                    'cover': video_data.get('pic', ''),
                    'video_url': f"https://www.bilibili.com/video/{video_id}",
                    'author': video_data.get('owner', {}).get('name', ''),
                    'platform': 'bilibili',
                    'description': video_data.get('desc', ''),
                    'duration': video_data.get('duration', 0)
                }
                return video_info
        except Exception as e:
            print(f"解析B站视频失败: {e}")
            return None
    def parse_url(self, url):
        """解析视频链接主函数"""
        # 1. 识别平台
        platform = self.identify_platform(url)
        if platform == 'unknown':
            return {'error': '无法识别的平台', 'platform': 'unknown'}
        # 2. 提取视频ID
        video_id = self.extract_video_id(url, platform)
        if not video_id:
            return {'error': '无法提取视频ID', 'platform': platform}
        # 3. 根据平台调用对应的解析方法
        parse_methods = {
            'douyin': self.parse_douyin,
            'kuaishou': self.parse_kuaishou,
            'bilibili': self.parse_bilibili
        }
        parser = parse_methods.get(platform)
        if not parser:
            return {'error': f'暂不支持{platform}平台解析', 'platform': platform}
        # 4. 执行解析
        result = parser(video_id)
        if result:
            result['url'] = url
            result['video_id'] = video_id
            return result
        else:
            return {'error': '解析失败', 'platform': platform, 'video_id': video_id}
# 使用示例
if __name__ == "__main__":
    parser = ShortVideoParser()
    # 测试各种链接
    test_urls = [
        "https://v.douyin.com/xxxxx/",  # 抖音短链接
        "https://www.douyin.com/video/123456789",  # 抖音长链接
        "https://v.kuaishou.com/xxxxx",  # 快手短链接
        "https://www.bilibili.com/video/BV1GJ411x7y7",  # B站链接
    ]
    for url in test_urls:
        print(f"正在解析: {url}")
        result = parser.parse_url(url)
        print(f"结果: {json.dumps(result, ensure_ascii=False, indent=2)}")
        print("-" * 50)

增强版脚本（包含更多功能）

import os
import tempfile
import uuid
from typing import Optional, Dict
class AdvancedVideoParser(ShortVideoParser):
    """增强版解析器"""
    def __init__(self):
        super().__init__()
        self.temp_dir = tempfile.mkdtemp()
    def download_video(self, video_url: str, filename: Optional[str] = None) -> str:
        """下载视频到本地"""
        if not filename:
            filename = f"{uuid.uuid4()}.mp4"
        filepath = os.path.join(self.temp_dir, filename)
        with self.session.get(video_url, stream=True) as r:
            r.raise_for_status()
            with open(filepath, 'wb') as f:
                for chunk in r.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)
        return filepath
    def get_video_metadata(self, url: str) -> Dict:
        """获取视频完整元数据"""
        result = self.parse_url(url)
        if 'error' in result:
            return result
        # 获取更多信息
        if result.get('video_url'):
            # 获取文件大小
            head = self.session.head(result['video_url'])
            result['file_size'] = head.headers.get('content-length', 0)
            result['content_type'] = head.headers.get('content-type', '')
        return result
    def batch_parse(self, urls: list) -> list:
        """批量解析多个链接"""
        results = []
        for url in urls:
            result = self.parse_url(url)
            results.append({
                'url': url,
                'result': result
            })
        return results
# 多线程版本
from concurrent.futures import ThreadPoolExecutor, as_completed
class ThreadedVideoParser(AdvancedVideoParser):
    """多线程解析器"""
    def batch_parse_parallel(self, urls: list, max_workers: int = 5) -> list:
        """并行批量解析"""
        results = []
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_url = {
                executor.submit(self.parse_url, url): url 
                for url in urls
            }
            for future in as_completed(future_to_url):
                url = future_to_url[future]
                try:
                    result = future.result()
                    results.append({
                        'url': url,
                        'result': result,
                        'status': 'success'
                    })
                except Exception as e:
                    results.append({
                        'url': url,
                        'result': str(e),
                        'status': 'failed'
                    })
        return results
# 简化版的API服务
from flask import Flask, request, jsonify
app = Flask(__name__)
parser = ThreadedVideoParser()
@app.route('/parse', methods=['POST'])
def parse_video():
    """视频解析API接口"""
    try:
        data = request.json
        url = data.get('url')
        if not url:
            return jsonify({'error': '缺少URL参数'}), 400
        result = parser.parse_url(url)
        if 'error' in result:
            return jsonify(result), 400
        return jsonify(result)
    except Exception as e:
        return jsonify({'error': str(e)}), 500
@app.route('/batch_parse', methods=['POST'])
def batch_parse():
    """批量解析API接口"""
    try:
        data = request.json
        urls = data.get('urls', [])
        if not urls:
            return jsonify({'error': '缺少URL列表'}), 400
        results = parser.batch_parse_parallel(urls)
        return jsonify({'results': results, 'total': len(results)})
    except Exception as e:
        return jsonify({'error': str(e)}), 500
# 启动API服务
# if __name__ == '__main__':
#     app.run(host='0.0.0.0', port=5000, debug=True)

命令行工具版本

#!/usr/bin/env python3
import sys
import argparse
import json
def main():
    parser = argparse.ArgumentParser(description='短视频链接解析工具')
    parser.add_argument('urls', nargs='+', help='要解析的视频链接')
    parser.add_argument('--output', '-o', help='输出文件路径')
    parser.add_argument('--format', '-f', choices=['text', 'json'], default='json', help='输出格式')
    parser.add_argument('--download', '-d', action='store_true', help='下载视频')
    args = parser.parse_args()
    video_parser = AdvancedVideoParser()
    results = []
    for url in args.urls:
        result = video_parser.parse_url(url)
        results.append({
            'url': url,
            'result': result
        })
        if args.download and 'url' in result and 'video_url' in result:
            try:
                filepath = video_parser.download_video(result['video_url'])
                results[-1]['downloaded'] = True
                results[-1]['filepath'] = filepath
                print(f"视频已下载到: {filepath}")
            except Exception as e:
                print(f"下载失败: {e}")
    if args.format == 'json':
        output = json.dumps(results, ensure_ascii=False, indent=2)
    else:
        output = ""
        for item in results:
            output += f"URL: {item['url']}\n"
            if 'error' in item['result']:
                output += f"状态: 失败 - {item['result']['error']}\n"
            else:
                output += f"平台: {item['result'].get('platform', '未知')}\n"
                output += f"标题: {item['result'].get('title', '未知')}\n"
                output += f"作者: {item['result'].get('author', '未知')}\n"
                output += f"封面: {item['result'].get('cover', '无')}\n"
                output += f"视频源: {item['result'].get('video_url', '无')}\n"
            output += "-" * 40 + "\n"
    if args.output:
        with open(args.output, 'w', encoding='utf-8') as f:
            f.write(output)
        print(f"结果已保存到: {args.output}")
    else:
        print(output)
if __name__ == "__main__":
    main()
# 使用方式：
# python video_parser.py "https://v.douyin.com/xxxxx/"
# python video_parser.py url1 url2 url3 --output result.json
# python video_parser.py "https://www.bilibili.com/video/BV1GJ411x7y7" --download

重要说明

1. 法律合规：

   • 仅用于个人学习和研究目的
   • 遵守各平台的用户协议和服务条款
   • 不要用于商业用途或大规模爬取

2. 技术限制：

   • 某些平台（如抖音）需要登录cookie才能获取完整信息
   • 视频下载可能受版权保护
   • 接口可能随时变更,需要定期更新

3. 性能优化：

   • 使用连接池减少网络开销
   • 实现缓存机制避免重复请求
   • 使用异步IO处理大量请求

4. 错误处理：

   • 添加重试机制
   • 处理网络超时
   • 验证返回数据完整性

这个脚本提供了一个完整的短视频链接解析框架,你可以根据实际需要扩展更多平台和功能，记得遵守相关法律法规和平台规则。