脚本怎样过滤失效图片链接

本文目录导读：

1. JavaScript 前端过滤
2. Python 后端过滤
3. Node.js 服务端过滤
4. 数据清洗脚本（Python）
5. 命令行工具（Linux/Mac）
6. 使用建议

JavaScript 前端过滤

使用 Image 对象检测

function filterBrokenImages(urls) {
    return new Promise((resolve) => {
        const validUrls = [];
        let completed = 0;
        urls.forEach(url => {
            const img = new Image();
            img.onload = () => {
                validUrls.push(url);
                completed++;
                if (completed === urls.length) resolve(validUrls);
            };
            img.onerror = () => {
                completed++;
                if (completed === urls.length) resolve(validUrls);
            };
            img.src = url;
        });
        if (urls.length === 0) resolve([]);
    });
}
// 使用示例
const imageUrls = ['https://example.com/img1.jpg', 'https://example.com/img2.jpg'];
filterBrokenImages(imageUrls).then(valid => {
    console.log('有效图片:', valid);
});

批量检测（含超时控制）

async function validateImages(urls, timeout = 5000) {
    const results = [];
    for (const url of urls) {
        try {
            const valid = await Promise.race([
                new Promise((resolve) => {
                    const img = new Image();
                    img.onload = () => resolve(true);
                    img.onerror = () => resolve(false);
                    img.src = url;
                }),
                new Promise(resolve => setTimeout(() => resolve(false), timeout))
            ]);
            if (valid) results.push(url);
        } catch {
            // 忽略错误
        }
    }
    return results;
}

Python 后端过滤

使用 requests 检测

import requests
from concurrent.futures import ThreadPoolExecutor
def check_image_url(url, timeout=5):
    """检查图片链接是否有效"""
    try:
        resp = requests.head(url, timeout=timeout, allow_redirects=True)
        content_type = resp.headers.get('Content-Type', '')
        return resp.status_code == 200 and content_type.startswith('image/')
    except:
        return False
def filter_broken_images(urls, max_workers=10):
    """批量过滤失效图片链接"""
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        results = list(executor.map(check_image_url, urls))
    return [url for url, valid in zip(urls, results) if valid]
# 使用示例
image_urls = ['https://example.com/img1.jpg', 'https://example.com/img2.jpg']
valid_urls = filter_broken_images(image_urls)
print(f'有效图片: {valid_urls}')

异步版本

import aiohttp
import asyncio
async def check_image(session, url):
    try:
        async with session.head(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
            if resp.status == 200:
                content_type = resp.headers.get('Content-Type', '')
                return url if content_type.startswith('image/') else None
    except:
        return None
async def filter_broken_images_async(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [check_image(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return [url for url in results if url]

Node.js 服务端过滤

const axios = require('axios');
const { performance } = require('perf_hooks');
async function validateImageUrl(url, timeout = 5000) {
    try {
        const response = await axios.head(url, {
            timeout,
            validateStatus: status => status === 200
        });
        const contentType = response.headers['content-type'];
        return contentType && contentType.startsWith('image/');
    } catch {
        return false;
    }
}
async function filterBrokenImages(urls, concurrency = 5) {
    const results = [];
    const queue = [...urls];
    async function worker() {
        while (queue.length > 0) {
            const url = queue.shift();
            const valid = await validateImageUrl(url);
            if (valid) results.push(url);
        }
    }
    const workers = Array(concurrency).fill().map(() => worker());
    await Promise.all(workers);
    return results;
}
// 使用示例
const urls = ['https://example.com/img1.jpg', 'https://example.com/img2.jpg'];
filterBrokenImages(urls).then(valid => {
    console.log('有效图片:', valid);
});

数据清洗脚本（Python）

import pandas as pd
import requests
from urllib.parse import urlparse
def clean_image_urls(df, column_name='image_url'):
    """清洗DataFrame中的图片链接"""
    def is_valid_image_url(url):
        if not url or not isinstance(url, str):
            return False
        # 1. 检查URL格式
        try:
            result = urlparse(url)
            if not all([result.scheme, result.netloc]):
                return False
        except:
            return False
        # 2. 检查文件扩展名
        valid_extensions = ['.jpg', '.jpeg', '.png', '.gif', '.webp', '.bmp', '.svg']
        if not any(url.lower().endswith(ext) for ext in valid_extensions):
            return False
        # 3. 检查URL是否可访问
        try:
            resp = requests.head(url, timeout=5, allow_redirects=True)
            return resp.status_code == 200
        except:
            return False
    # 应用过滤
    df['is_valid'] = df[column_name].apply(is_valid_image_url)
    return df[df['is_valid']].drop('is_valid', axis=1)
# 使用示例
df = pd.DataFrame({
    'id': [1, 2, 3],
    'image_url': [
        'https://example.com/valid.jpg',
        'https://example.com/broken.jpg',
        'invalid-url'
    ]
})
cleaned_df = clean_image_urls(df)

命令行工具（Linux/Mac）

#!/bin/bash
# validate_images.sh
while read url; do
    response=$(curl -s -o /dev/null -w "%{http_code}" --head "$url" --max-time 5)
    if [ "$response" = "200" ]; then
        content_type=$(curl -s -o /dev/null -w "%{content_type}" --head "$url" --max-time 5)
        if [[ "$content_type" == image/* ]]; then
            echo "$url" >> valid_images.txt
        fi
    fi
done < image_urls.txt

使用建议

1. 性能优化：

   • 批量检测时使用并发/异步
   • 设置合理的超时时间（3-5秒）
   • 缓存已检测的结果

2. 错误处理：

   • 处理网络异常
   • 处理SSL证书问题
   • 处理重定向

3. 注意事项：

   • 部分网站会屏蔽HEAD请求,可改用GET部分内容
   • 注意反爬策略,适当添加延时
   • CDN图片可能需要特殊处理

根据你的具体场景选择合适的方案,如果需要处理海量链接，建议使用Python或Node.js的异步方案。