怎样实现链接批量校验脚本

wen 实用脚本 2026-06-29 24

本文目录导读：

怎样实现链接批量校验脚本

方案一：Python脚本（最常用）
方案二：Python精简版
方案三：JavaScript/Node.js版本
安装依赖
使用建议

我来帮你编写一个链接批量校验脚本,这里提供几个不同方案的实现：

Python脚本（最常用）

import requests
import pandas as pd
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urlparse
import time
from datetime import datetime
class LinkValidator:
    def __init__(self, timeout=10, max_workers=10):
        self.timeout = timeout
        self.max_workers = max_workers
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        })
    def check_link(self, url):
        """检查单个链接"""
        result = {
            'url': url,
            'status': 'unknown',
            'status_code': None,
            'response_time': None,
            'error': None
        }
        start_time = time.time()
        try:
            response = self.session.get(url, timeout=self.timeout, allow_redirects=True)
            result['status_code'] = response.status_code
            result['response_time'] = round(time.time() - start_time, 2)
            if response.status_code == 200:
                result['status'] = '有效'
            elif response.status_code == 301 or response.status_code == 302:
                result['status'] = '重定向'
            elif response.status_code == 404:
                result['status'] = '失效(404)'
            elif response.status_code >= 500:
                result['status'] = '服务器错误'
            else:
                result['status'] = f'其他错误({response.status_code})'
        except requests.ConnectionError:
            result['status'] = '连接失败'
            result['error'] = 'ConnectionError'
        except requests.Timeout:
            result['status'] = '超时'
            result['error'] = 'Timeout'
        except requests.RequestException as e:
            result['status'] = '请求异常'
            result['error'] = str(e)
        except Exception as e:
            result['status'] = '未知错误'
            result['error'] = str(e)
        result['response_time'] = round(time.time() - start_time, 2)
        return result
    def validate_links(self, urls):
        """批量校验链接"""
        results = []
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            future_to_url = {executor.submit(self.check_link, 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(result)
                    print(f"✓ {result['status']}: {url}")
                except Exception as e:
                    print(f"✗ 错误: {url} - {e}")
        return results
    def save_results(self, results, filename='link_check_results.xlsx'):
        """保存结果到Excel"""
        df = pd.DataFrame(results)
        # 添加统计信息
        stats = {
            '总计': len(results),
            '有效': len([r for r in results if r['status'] == '有效']),
            '失效': len([r for r in results if '失效' in r['status']]),
            '错误': len([r for r in results if r['status'] not in ['有效', '失效(404)']])
        }
        # 写入Excel
        with pd.ExcelWriter(filename, engine='openpyxl') as writer:
            df.to_excel(writer, sheet_name='链接检查结果', index=False)
            # 添加统计信息到新sheet
            stats_df = pd.DataFrame([stats])
            stats_df.to_excel(writer, sheet_name='统计信息', index=False)
        print(f"\n✓ 结果已保存到: {filename}")
        print(f"总计: {stats['总计']}, 有效: {stats['有效']}, 失效: {stats['失效']}, 错误: {stats['错误']}")
# 使用示例
def main():
    # 方式1: 从文件读取链接
    # urls = pd.read_excel('urls.xlsx')['url'].tolist()
    # 方式2: 直接提供链接列表
    urls = [
        'https://www.example.com',
        'https://www.google.com',
        'https://www.notexist123456.com',
        'https://httpbin.org/status/404',
        'https://httpbin.org/status/500',
    ]
    # 方式3: 从文本文件读取
    # with open('urls.txt', 'r') as f:
    #     urls = [line.strip() for line in f if line.strip()]
    validator = LinkValidator(timeout=10, max_workers=10)
    print("开始批量校验链接...")
    print("-" * 50)
    results = validator.validate_links(urls)
    # 保存结果
    validator.save_results(results)
    # 输出简要报告
    print("\n" + "=" * 50)
    print("验证报告")
    print("=" * 50)
    for result in results:
        print(f"{result['status']:20s} | {result['url']:50s} | {result['response_time']}s")
if __name__ == "__main__":
    main()

Python精简版

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
def check_link(url, timeout=10):
    try:
        response = requests.get(url, timeout=timeout, allow_redirects=True)
        return url, response.status_code, response.elapsed.total_seconds()
    except Exception as e:
        return url, None, str(e)
def validate_links(urls, max_workers=20):
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(check_link, url): url for url in urls}
        for future in as_completed(futures):
            url, status, info = future.result()
            if status == 200:
                results.append((url, '✓ 有效', info))
                print(f"✓ 有效: {url}")
            elif status:
                results.append((url, f'✗ {status}', info))
                print(f"✗ {status}: {url}")
            else:
                results.append((url, '✗ 错误', info))
                print(f"✗ 错误: {url} - {info}")
    return results
# 使用
urls = ['https://example.com', 'https://google.com', 'https://notexists']
results = validate_links(urls)
# 保存结果
with open('results.txt', 'w', encoding='utf-8') as f:
    for url, status, info in results:
        f.write(f"{status}\t{url}\t{info}\n")

JavaScript/Node.js版本

const axios = require('axios');
const fs = require('fs');
const path = require('path');
async function checkLink(url, timeout = 10000) {
    const startTime = Date.now();
    try {
        const response = await axios.get(url, {
            timeout: timeout,
            maxRedirects: 5,
            headers: {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            }
        });
        return {
            url: url,
            status: response.status === 200 ? '有效' : 
                   response.status >= 300 && response.status < 400 ? '重定向' :
                   response.status === 404 ? '失效(404)' : `其他(${response.status})`,
            statusCode: response.status,
            responseTime: Date.now() - startTime
        };
    } catch (error) {
        return {
            url: url,
            status: '错误',
            error: error.message,
            responseTime: Date.now() - startTime
        };
    }
}
async function validateLinks(urls, concurrency = 10) {
    const results = [];
    // 分批处理以控制并发
    for (let i = 0; i < urls.length; i += concurrency) {
        const batch = urls.slice(i, i + concurrency);
        const batchResults = await Promise.all(
            batch.map(url => checkLink(url))
        );
        results.push(...batchResults);
        batchResults.forEach(r => {
            console.log(`${r.status === '有效' ? '✓' : '✗'} ${r.status}: ${r.url}`);
        });
    }
    return results;
}
// 保存结果
function saveResults(results, filename = 'link_results.json') {
    fs.writeFileSync(filename, JSON.stringify(results, null, 2));
    // 统计
    const valid = results.filter(r => r.status === '有效').length;
    const invalid = results.filter(r => r.status.includes('失效')).length;
    const errors = results.filter(r => r.status === '错误').length;
    console.log(`\n结果已保存到: ${filename}`);
    console.log(`总计: ${results.length}, 有效: ${valid}, 失效: ${invalid}, 错误: ${errors}`);
}
// 使用示例
const urls = [
    'https://www.example.com',
    'https://www.google.com',
    'https://www.notexist123456.com'
];
validateLinks(urls).then(results => {
    saveResults(results);
});