[ Web Proxy ]
URL:
Viewing: https://github-redirect.dependabot.com/flypythoncom/python/pull/77/files [Back]  [Original]

Update check_links.py by Avijit-roy Pull Request #77 flypythoncom/python GitHub
Skip to content

Navigation Menu

Sign in
Search/
Sign up
Open
Changes from all commits
Commits
File filter

Filter by extension

Conversations
Failed to load comments. Retry
Loading
Jump to
Jump to file
Failed to load files. Retry
Loading
Diff view
[Apply and reload]
Diff view
[Apply and reload]
267 changes: 154 additions & 113 deletions tools/check_links.py
Show comments View file Edit file Delete file Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -1,189 +1,230 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
FlyPython
README
FlyPython Link Checker Tool
Periodically checks the validity of all external links in README files
"""

import re
import requests
import time
import json
import os
from pathlib import Path
from urllib.parse import urlparse
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


class LinkChecker:
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 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
})
self.timeout = 10
def __init__(self, timeout: int = 10, max_workers: int = 10):
self.session = self._create_session()
self.timeout = timeout
self.max_workers = max_workers
self.results = {
'working': [],
'broken': [],
'redirect': [],
'timeout': [],
'unknown': []
}

def extract_links_from_file(self, filename):
"""markdown"""
self.processed_urls = set()

def _create_session(self) -> requests.Session:
"""Create a requests session with retry strategy and headers"""
session = requests.Session()

# Configure retry strategy
retry_strategy = Retry(
total=2,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('http://', adapter)
session.mount('https://', adapter)

# Set user agent
session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})

return session

def extract_links_from_file(self, filename: str) -> List[Dict]:
"""Extract all external links from a markdown file"""
filepath = Path(filename)

if not filepath.exists():
print(f"File not found: {filename}")
return []

try:
with open(filename, 'r', encoding='utf-8') as f:
content = f.read()
content = filepath.read_text(encoding='utf-8')
except Exception as e:
print(f" {filename}: {e}")
print(f"Failed to read {filename}: {e}")
return []

# markdown [text](url)
markdown_links = re.findall(r'\[([^\]]*)\]\(([^)]+)\)', content)

#
url_pattern = r'https?://[^\s\])\}]+'
plain_links = re.findall(url_pattern, content)

links = []

# markdown
# Extract markdown links [text](url)
markdown_links = re.findall(r'\[([^\]]*)\]\(([^)]+)\)', content)
for text, url in markdown_links:
if url.startswith('http'):
links.append({
'text': text,
'url': url,
'file': filename,
'file': str(filepath),
'type': 'markdown'
})

#
for url in plain_links:
#
if not any(link['url'] == url for link in links):
# Extract plain URLs
plain_urls = re.findall(r'https?://[^\s\])\}]+', content)
seen = {link['url'] for link in links}

for url in plain_urls:
if url not in seen:
links.append({
'text': url,
'url': url,
'file': filename,
'file': str(filepath),
'type': 'plain'
})
seen.add(url)

return links
def check_link(self, link):
""""""

def check_link(self, link: Dict) -> Dict:
"""Check the status of a single link"""
url = link['url']

if url in self.processed_urls:
return link

self.processed_urls.add(url)

try:
# Try HEAD request first (faster)
response = self.session.head(url, timeout=self.timeout, allow_redirects=True)
status_code = response.status_code

if status_code == 200:
link['status'] = 'working'
link['status_code'] = status_code
self.results['working'].append(link)
elif 300 <= status_code < 400:
link['status'] = 'redirect'
link['status_code'] = status_code
link['final_url'] = response.url
self.results['redirect'].append(link)
else:
# GETHEAD
try:
response = self.session.get(url, timeout=self.timeout)
if response.status_code == 200:
link['status'] = 'working'
link['status_code'] = response.status_code
self.results['working'].append(link)
else:
link['status'] = 'broken'
link['status_code'] = response.status_code
self.results['broken'].append(link)
except:
link['status'] = 'broken'
link['status_code'] = status_code
self.results['broken'].append(link)
return self._process_response(link, response)

except requests.exceptions.Timeout:
link['status'] = 'timeout'
link['error'] = 'Request timeout'
self.results['timeout'].append(link)
return link

except requests.exceptions.RequestException as e:
link['status'] = 'unknown'
link['error'] = str(e)
self.results['unknown'].append(link)
# Fall back to GET request for servers that don't support HEAD
try:
response = self.session.get(url, timeout=self.timeout)
return self._process_response(link, response)
except requests.exceptions.RequestException:
link['status'] = 'unknown'
link['error'] = str(e)
self.results['unknown'].append(link)
return link

def _process_response(self, link: Dict, response: requests.Response) -> Dict:
"""Process HTTP response and categorize link"""
status_code = response.status_code

if status_code == 200:
link['status'] = 'working'
self.results['working'].append(link)
elif 300 <= status_code < 400:
link['status'] = 'redirect'
link['final_url'] = response.url
self.results['redirect'].append(link)
else:
link['status'] = 'broken'
self.results['broken'].append(link)

link['status_code'] = status_code
return link
def check_all_links(self, links, max_workers=10):
""""""
print(f" {len(links)} ...")

def check_all_links(self, links: List[Dict]) -> None:
"""Concurrently check all links"""
print(f"Checking {len(links)} links...\n")

with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_link = {executor.submit(self.check_link, link): link for link in links}
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = {executor.submit(self.check_link, link): link for link in links}

for i, future in enumerate(as_completed(future_to_link), 1):
link = future_to_link[future]
for i, future in enumerate(as_completed(futures), 1):
link = futures[future]
try:
result = future.result()
status = result.get('status', 'unknown')
print(f"[{i}/{len(links)}] {status.upper()}: {result['url']}")
time.sleep(0.1)
status = result.get('status', 'unknown').upper()
print(f"[{i}/{len(links)}] {status}: {result['url']}")
except Exception as e:
print(f" {link['url']}: {e}")
def generate_report(self):
""""""
print(f"Error checking {link['url']}: {e}")

def generate_report(self, output_dir: str = 'reports') -> None:
"""Generate and save detailed report"""
total = sum(len(links) for links in self.results.values())

print("\n" + "="*60)
print("")
print("="*60)
print(f": {total}")
print(f": {len(self.results['working'])}")
print(f": {len(self.results['redirect'])}")
print(f": {len(self.results['broken'])}")
print(f": {len(self.results['timeout'])}")
print(f": {len(self.results['unknown'])}")

#
os.makedirs('../reports', exist_ok=True)
with open('../reports/link_check_results.json', 'w', encoding='utf-8') as f:
report = f"""
{'='*60}
Link Check Report
{'='*60}
Total Links: {total}
Working: {len(self.results['working'])}
Redirects: {len(self.results['redirect'])}
Broken: {len(self.results['broken'])}
Timeouts: {len(self.results['timeout'])}
? Unknown: {len(self.results['unknown'])}
{'='*60}
"""
print(report)

# Save detailed results
output_path = Path(output_dir)
output_path.mkdir(exist_ok=True)

results_file = output_path / 'link_check_results.json'
with open(results_file, 'w', encoding='utf-8') as f:
json.dump(self.results, f, ensure_ascii=False, indent=2)

print(f"\n: reports/link_check_results.json")
print(f"Detailed results saved to: {results_file}")

def deduplicate_links(self, links: List[Dict]) -> List[Dict]:
"""Remove duplicate links by URL"""
seen = set()
unique = []
for link in links:
if link['url'] not in seen:
unique.append(link)
seen.add(link['url'])
return unique


def main():
checker = LinkChecker()
files = ['../README.md', '../README_cn.md']

# README ()
files_to_check = ['../README.md', '../README_cn.md']
# Extract links
checker = LinkChecker(timeout=10, max_workers=10)
all_links = []

for filename in files_to_check:
print(f" {filename} ...")
for filename in files:
print(f"Extracting links from {filename}...")
links = checker.extract_links_from_file(filename)
all_links.extend(links)
print(f" {len(links)} ")
if links:
all_links.extend(links)
print(f"Found {len(links)} links\n")

if not all_links:
print("!")
print("No links found!")
return

#
unique_links = []
seen_urls = set()
for link in all_links:
if link['url'] not in seen_urls:
unique_links.append(link)
seen_urls.add(link['url'])

print(f" {len(unique_links)} ")
# Deduplicate and check
unique_links = checker.deduplicate_links(all_links)
print(f"Checking {len(unique_links)} unique links\n")

#
checker.check_all_links(unique_links)

#
checker.generate_report()


if __name__ == '__main__':
main()
main()
Toggle all file notes Toggle all file annotations

Footer

© 2026 GitHub, Inc.

Web Proxy Viewer  |  New URL  |  Original Page