定时检查并清理无效链接脚本

本文目录导读：

1. 基础版：检查单个文件中的链接
2. 使用说明
3. 高级功能

我来为您创建一个定时检查并清理无效链接的脚本,这个脚本可以检查文件中的链接是否有效，并自动清理无效的链接。

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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
定时检查并清理无效链接脚本
"""
import os
import re
import json
import requests
import time
import logging
from datetime import datetime, timedelta
from urllib.parse import urlparse
from concurrent.futures import ThreadPoolExecutor, as_completed
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('link_checker.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)
class LinkChecker:
    def __init__(self, config_file='config.json'):
        """初始化链接检查器"""
        self.config = self.load_config(config_file)
        self.valid_links = []
        self.invalid_links = []
        self.timeout = self.config.get('timeout', 5)
        self.retry_times = self.config.get('retry_times', 2)
        self.max_workers = self.config.get('max_workers', 10)
    def load_config(self, config_file):
        """加载配置文件"""
        default_config = {
            'files_to_check': ['README.md', 'links.txt'],  # 要检查的文件
            'backup_dir': 'backup',  # 备份目录
            'cleanup_mode': 'comment',  # 'comment' 或 'delete'
            'timeout': 5,  # 请求超时时间
            'retry_times': 2,  # 重试次数
            'max_workers': 10,  # 并发线程数
            'check_interval': 3600,  # 检查间隔（秒）
            'report_file': 'link_report.json',  # 报告文件
            'link_patterns': [  # 支持多种链接格式
                r'https?://[^\s<>"\'()]+\.[^\s<>"\'()]+',
                r'https?://[^\s<>"\'()\[\]]+'
            ]
        }
        try:
            if os.path.exists(config_file):
                with open(config_file, 'r', encoding='utf-8') as f:
                    loaded_config = json.load(f)
                    default_config.update(loaded_config)
                    logger.info(f"配置文件 {config_file} 加载成功")
            else:
                self.save_config(config_file, default_config)
                logger.info(f"已创建默认配置文件 {config_file}")
        except Exception as e:
            logger.error(f"加载配置文件失败: {e}")
        return default_config
    def save_config(self, config_file, config):
        """保存配置文件"""
        with open(config_file, 'w', encoding='utf-8') as f:
            json.dump(config, f, indent=4, ensure_ascii=False)
    def extract_links(self, file_path):
        """从文件中提取所有链接"""
        links = []
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()
            for pattern in self.config['link_patterns']:
                found_links = re.findall(pattern, content)
                links.extend(found_links)
            # 去重
            links = list(set(links))
            logger.info(f"从 {file_path} 中提取到 {len(links)} 个链接")
            return links
        except Exception as e:
            logger.error(f"提取链接失败 {file_path}: {e}")
            return []
    def check_link(self, url):
        """检查单个链接是否有效"""
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
        for attempt in range(self.retry_times + 1):
            try:
                response = requests.get(
                    url, 
                    headers=headers,
                    timeout=self.timeout,
                    allow_redirects=True
                )
                if response.status_code == 200:
                    return url, True, "OK"
                elif response.status_code == 403:
                    return url, True, "Forbidden (可能有效)"
                elif response.status_code == 404:
                    return url, False, f"404 Not Found"
                else:
                    return url, False, f"HTTP {response.status_code}"
            except requests.ConnectionError:
                if attempt < self.retry_times:
                    time.sleep(1)
                    continue
                return url, False, "Connection Error"
            except requests.Timeout:
                if attempt < self.retry_times:
                    time.sleep(1)
                    continue
                return url, False, "Timeout"
            except Exception as e:
                if attempt < self.retry_times:
                    time.sleep(1)
                    continue
                return url, False, str(e)
        return url, False, "Max retries exceeded"
    def check_all_links(self, file_path):
        """检查文件中的所有链接"""
        links = self.extract_links(file_path)
        if not links:
            logger.warning(f"文件 {file_path} 中没有找到链接")
            return [], []
        self.valid_links = []
        self.invalid_links = []
        logger.info(f"开始检查 {len(links)} 个链接...")
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            future_to_url = {executor.submit(self.check_link, url): url for url in links}
            for future in as_completed(future_to_url):
                url, is_valid, message = future.result()
                if is_valid:
                    self.valid_links.append(url)
                    logger.info(f"✓ {url[:50]}... - {message}")
                else:
                    self.invalid_links.append({'url': url, 'reason': message})
                    logger.warning(f"✗ {url[:50]}... - {message}")
        return self.valid_links, self.invalid_links
    def backup_file(self, file_path):
        """备份文件"""
        if not os.path.exists(self.config['backup_dir']):
            os.makedirs(self.config['backup_dir'])
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        filename = os.path.basename(file_path)
        backup_path = os.path.join(
            self.config['backup_dir'], 
            f"{filename}.{timestamp}.bak"
        )
        try:
            import shutil
            shutil.copy2(file_path, backup_path)
            logger.info(f"已备份文件到 {backup_path}")
            return True
        except Exception as e:
            logger.error(f"备份失败: {e}")
            return False
    def cleanup_links(self, file_path, invalid_links):
        """清理无效链接"""
        if not invalid_links:
            logger.info("没有发现无效链接，无需清理")
            return
        # 创建备份
        self.backup_file(file_path)
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()
            cleanup_mode = self.config.get('cleanup_mode', 'comment')
            invalid_urls = [link['url'] for link in invalid_links]
            for url in invalid_urls:
                if cleanup_mode == 'delete':
                    # 删除包含无效链接的整行
                    lines = content.split('\n')
                    content = '\n'.join([
                        line for line in lines 
                        if url not in line
                    ])
                    logger.info(f"已删除包含 {url[:50]}... 的行")
                else:
                    # 注释掉包含无效链接的行
                    lines = content.split('\n')
                    new_lines = []
                    for line in lines:
                        if url in line and not line.strip().startswith('#'):
                            new_lines.append(f"# INVALID LINK: {line}")
                        else:
                            new_lines.append(line)
                    content = '\n'.join(new_lines)
                    logger.info(f"已注释包含 {url[:50]}... 的行")
            with open(file_path, 'w', encoding='utf-8') as f:
                f.write(content)
            logger.info(f"清理完成，已处理 {len(invalid_urls)} 个无效链接")
        except Exception as e:
            logger.error(f"清理链接失败: {e}")
    def generate_report(self, file_path):
        """生成检查报告"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'file_checked': file_path,
            'total_links': len(self.valid_links) + len(self.invalid_links),
            'valid_links': len(self.valid_links),
            'invalid_links': len(self.invalid_links),
            'valid_urls': self.valid_links,
            'invalid_urls': self.invalid_links,
            'cleanup_mode': self.config['cleanup_mode']
        }
        report_file = self.config['report_file']
        try:
            with open(report_file, 'w', encoding='utf-8') as f:
                json.dump(report, f, indent=2, ensure_ascii=False)
            logger.info(f"报告已保存到 {report_file}")
        except Exception as e:
            logger.error(f"保存报告失败: {e}")
        return report
    def run_check(self, file_path=None):
        """运行链接检查"""
        if file_path is None:
            file_paths = self.config['files_to_check']
        else:
            file_paths = [file_path]
        for fp in file_paths:
            if not os.path.exists(fp):
                logger.warning(f"文件 {fp} 不存在，跳过")
                continue
            logger.info(f"="*50)
            logger.info(f"开始检查文件: {fp}")
            valid_links, invalid_links = self.check_all_links(fp)
            if invalid_links:
                logger.info(f"\n发现 {len(invalid_links)} 个无效链接:")
                for link in invalid_links:
                    logger.info(f"  {link['url'][:60]}... - {link['reason']}")
                # 清理无效链接
                self.cleanup_links(fp, invalid_links)
            else:
                logger.info("所有链接看起来都有效！")
            # 生成报告
            self.generate_report(fp)
    def run_scheduled(self):
        """定时运行检查"""
        logger.info(f"启动定时链接检查，间隔: {self.config['check_interval']}秒")
        while True:
            try:
                self.run_check()
                next_check = datetime.now() + timedelta(seconds=self.config['check_interval'])
                logger.info(f"下次检查时间: {next_check.strftime('%Y-%m-%d %H:%M:%S')}")
                time.sleep(self.config['check_interval'])
            except KeyboardInterrupt:
                logger.info("用户中断，停止检查")
                break
            except Exception as e:
                logger.error(f"检查过程出错: {e}")
                time.sleep(60)  # 出错后等待一分钟再试
def main():
    """主函数"""
    import argparse
    parser = argparse.ArgumentParser(description='定时检查并清理无效链接')
    parser.add_argument('--config', '-c', default='config.json', help='配置文件路径')
    parser.add_argument('--file', '-f', help='要检查的特定文件')
    parser.add_argument('--once', '-o', action='store_true', help='只运行一次检查')
    parser.add_argument('--mode', '-m', choices=['comment', 'delete'], help='清理模式')
    args = parser.parse_args()
    checker = LinkChecker(args.config)
    if args.mode:
        checker.config['cleanup_mode'] = args.mode
    if args.once:
        checker.run_check(args.file)
    else:
        checker.run_scheduled()
if __name__ == '__main__':
    main()

使用说明

安装依赖

pip install requests

创建配置文件 config.json

{
    "files_to_check": ["README.md", "links.txt", "docs/*.md"],
    "backup_dir": "backup",
    "cleanup_mode": "comment",
    "timeout": 5,
    "retry_times": 2,
    "max_workers": 10,
    "check_interval": 3600,
    "report_file": "link_report.json",
    "link_patterns": [
        "https?://[^\\s<>\"'()]+\\.[^\\s<>\"'()]+",
        "https?://[^\\s<>\"'()\\[\\]]+"
    ]
}

运行脚本

# 单次运行
python link_checker.py --once
# 定时运行（默认每小时检查一次）
python link_checker.py
# 检查特定文件
python link_checker.py --file README.md --once
# 设置清理模式为删除
python link_checker.py --mode delete --once
# 使用自定义配置
python link_checker.py --config my_config.json

高级功能

添加链接白名单

def should_skip(self, url):
    """检查链接是否应该跳过"""
    skip_patterns = [
        'localhost',
        '127.0.0.1',
        'example.com',
    ]
    for pattern in skip_patterns:
        if pattern in url:
            return True
    return False

增强的链接检测

def extract_all_links(self, file_path):
    """从文件中提取所有类型的链接"""
    import markdown
    from bs4 import BeautifulSoup
    # 读取文件
    with open(file_path, 'r', encoding='utf-8') as f:
        content = f.read()
    # 提取 Markdown 链接
    md_links = re.findall(r'\[([^\]]+)\]\(([^\)]+)\)', content)
    # 提取 HTML 链接
    soup = BeautifulSoup(markdown.markdown(content), 'html.parser')
    html_links = [a.get('href') for a in soup.find_all('a') if a.get('href')]
    return list(set(md_links + html_links))

定时任务调度

# 使用 systemd 或 crontab 实现定时运行
# 添加到 crontab
# 0 */2 * * * /usr/bin/python3 /path/to/link_checker.py --once

这个脚本提供了完整的链接检查和清理功能,可以根据需要调整配置和功能。