如何编写失效链接清理脚本

wen 实用脚本 2026-06-27 239

本文目录导读：

如何编写失效链接清理脚本

基础版：检查文件中的URL链接
使用方法
高级功能
注意事项

我来为您编写一个完整的失效链接清理脚本，这个脚本可以扫描指定目录中的文件，检查文件中的链接是否有效,并提供清理功能。

基础版：检查文件中的URL链接

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
失效链接清理脚本
功能：扫描文件中的URL链接，检查是否有效
"""
import os
import re
import requests
import argparse
from urllib.parse import urlparse
from concurrent.futures import ThreadPoolExecutor, as_completed
class BrokenLinkChecker:
    def __init__(self, directory, extensions=None, timeout=10, max_workers=10):
        """
        初始化链接检查器
        Args:
            directory: 要扫描的目录
            extensions: 要检查的文件扩展名列表
            timeout: 请求超时时间
            max_workers: 并发线程数
        """
        self.directory = directory
        self.extensions = extensions or ['.md', '.txt', '.html', '.rst', '.py']
        self.timeout = timeout
        self.max_workers = max_workers
        self.broken_links = []
    def find_links_in_file(self, filepath):
        """查找文件中的链接"""
        links = []
        url_pattern = re.compile(r'https?://[^\s\'\"<>\)]+')
        try:
            with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
                content = f.read()
                links = url_pattern.findall(content)
        except Exception as e:
            print(f"读取文件 {filepath} 时出错: {e}")
        return links
    def check_link(self, url, filepath, line_number=0):
        """检测单个链接是否有效"""
        try:
            # 清理URL
            url = url.rstrip('.,;:!?)>\'\"')
            # 跳过某些常见但不需要检查的域名
            skip_domains = ['localhost', '127.0.0.1', 'example.com']
            parsed = urlparse(url)
            if parsed.netloc in skip_domains:
                return None
            # 发送请求
            response = requests.get(
                url, 
                timeout=self.timeout,
                allow_redirects=True,
                headers={'User-Agent': 'Mozilla/5.0 (compatible; LinkChecker/1.0)'}
            )
            if response.status_code >= 400:
                return {
                    'url': url,
                    'file': filepath,
                    'status': response.status_code,
                    'error': f'HTTP {response.status_code}'
                }
        except requests.exceptions.ConnectionError:
            return {
                'url': url,
                'file': filepath,
                'status': 'ERROR',
                'error': '连接失败'
            }
        except requests.exceptions.Timeout:
            return {
                'url': url,
                'file': filepath,
                'status': 'ERROR',
                'error': '超时'
            }
        except requests.exceptions.RequestException as e:
            return {
                'url': url,
                'file': filepath,
                'status': 'ERROR',
                'error': str(e)
            }
        return None
    def scan_files(self):
        """扫描目录中的文件"""
        print(f"正在扫描目录: {self.directory}")
        print(f"检查的文件类型: {', '.join(self.extensions)}")
        print("=" * 60)
        files_to_check = []
        for root, dirs, files in os.walk(self.directory):
            # 跳过隐藏目录和.git等
            dirs[:] = [d for d in dirs if not d.startswith('.') and d != 'node_modules']
            for file in files:
                if any(file.endswith(ext) for ext in self.extensions):
                    files_to_check.append(os.path.join(root, file))
        print(f"找到 {len(files_to_check)} 个文件需要检查")
        # 使用线程池并发检查
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = []
            for filepath in files_to_check:
                links = self.find_links_in_file(filepath)
                for link in links:
                    future = executor.submit(self.check_link, link, filepath)
                    futures.append(future)
            # 处理结果
            completed = 0
            for future in as_completed(futures):
                completed += 1
                if completed % 50 == 0:
                    print(f"已检查 {completed} 个链接...")
                result = future.result()
                if result:
                    self.broken_links.append(result)
        return self.broken_links
    def generate_report(self, output_file=None):
        """生成失效链接报告"""
        if not self.broken_links:
            print("✓ 没有发现失效链接！")
            return
        print(f"\n发现 {len(self.broken_links)} 个失效链接:")
        print("-" * 80)
        report_content = []
        for link in sorted(self.broken_links, key=lambda x: (x['file'], x['url'])):
            line = f"文件: {link['file']}\n链接: {link['url']}\n状态: {link['status']} - {link['error']}\n"
            print(line)
            report_content.append(line + "\n")
        if output_file:
            with open(output_file, 'w', encoding='utf-8') as f:
                f.writelines(report_content)
            print(f"报告已保存到: {output_file}")
class LinkCleaner(BrokenLinkChecker):
    """扩展版：支持清理失效链接"""
    def clean_links(self, dry_run=True):
        """清理失效链接"""
        if not self.broken_links:
            print("没有需要清理的链接")
            return
        if dry_run:
            print("\n预演模式 - 不会实际修改文件")
            print("=" * 40)
        # 按文件分组
        links_by_file = {}
        for link in self.broken_links:
            if link['file'] not in links_by_file:
                links_by_file[link['file']] = []
            links_by_file[link['file']].append(link)
        for filepath, links in links_by_file.items():
            print(f"\n处理文件: {filepath}")
            try:
                with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
                    content = f.read()
                original_content = content
                # 替换失效链接
                for link in links:
                    old_url = link['url']
                    if not dry_run:
                        content = content.replace(old_url, f"[失效链接: {old_url}]")
                        print(f"  - 已标记失效链接: {old_url[:50]}...")
                    else:
                        print(f"  [预演] 将标记: {old_url[:50]}...")
                # 写入文件
                if not dry_run and content != original_content:
                    with open(filepath, 'w', encoding='utf-8') as f:
                        f.write(content)
                    print(f"  ✓ 文件已更新")
            except Exception as e:
                print(f"  ✗ 处理失败: {e}")
def main():
    parser = argparse.ArgumentParser(description='失效链接检查与清理工具')
    parser.add_argument('directory', help='要扫描的目录')
    parser.add_argument('--extensions', '-e', nargs='+', 
                       help='要检查的文件扩展名 (默认: .md .txt .html .rst .py)')
    parser.add_argument('--timeout', '-t', type=int, default=10,
                       help='请求超时时间(秒)')
    parser.add_argument('--workers', '-w', type=int, default=10,
                       help='并发线程数')
    parser.add_argument('--report', '-r', help='输出报告文件')
    parser.add_argument('--clean', '-c', action='store_true',
                       help='清理失效链接')
    parser.add_argument('--dry-run', '-d', action='store_true',
                       help='预演模式，不实际修改文件')
    args = parser.parse_args()
    if not os.path.exists(args.directory):
        print(f"错误: 目录 '{args.directory}' 不存在")
        return
    if args.clean:
        cleaner = LinkCleaner(
            args.directory,
            args.extensions,
            args.timeout,
            args.workers
        )
        cleaner.scan_files()
        if cleaner.broken_links:
            cleaner.generate_report(args.report)
            cleaner.clean_links(dry_run=args.dry_run)
    else:
        checker = BrokenLinkChecker(
            args.directory,
            args.extensions,
            args.timeout,
            args.workers
        )
        checker.scan_files()
        checker.generate_report(args.report)
if __name__ == '__main__':
    main()

使用方法

安装依赖

pip install requests

基本使用

# 检查当前目录
python broken_link_checker.py /path/to/your/directory
# 检查特定文件类型
python broken_link_checker.py /path/to/your/directory -e .md .txt
# 生成报告
python broken_link_checker.py /path/to/your/directory -r report.txt
# 清理失效链接（预演模式）
python broken_link_checker.py /path/to/your/directory -c -d
# 实际清理失效链接
python broken_link_checker.py /path/to/your/directory -c
# 自定义并发数和超时
python broken_link_checker.py /path/to/your/directory -w 20 -t 5

高级功能

添加白名单域名

# 在 check_link 方法中添加
whitelist_domains = ['github.com', 'stackoverflow.com']
if parsed.netloc in whitelist_domains:
    return None  # 跳过检查

支持更多文件类型

# 可以检查 Markdown 中的图片链接
img_pattern = re.compile(r'!\[.*?\]\((https?://[^\s\)]+)\)')
# 检查 HTML 中的链接
html_pattern = re.compile(r'<a\s+(?:[^>]*?\s+)?href="(https?://[^"]+)"')

批量替换方案

# 添加自动替换功能
replacements = {
    'old-domain.com': 'new-domain.com',
    'http://': 'https://'
}

注意事项

网络请求限制：避免发送过多请求导致被屏蔽
文件备份：清理前建议先备份文件
URL编码：有些URL可能包含特殊字符
：JavaScript生成的链接可能无法检测

这个脚本可以根据您的具体需求进行调整和扩展。

本文地址： https://www.bmjup.com/post/10320.html