script.py - master - kunda/broken_links

script.py
387 строк15 KB
Kekovich-kw
Initial
04 июн 2026, 11:00
04 июн 2026, 11:00c284884
import asyncio
import aiohttp
import openpyxl
from urllib.parse import urlparse, urljoin
from bs4 import BeautifulSoup
import re
from typing import Optional, Tuple
import sys
import os
import random
import time
from collections import OrderedDict

PATH = './sskindustry.xlsx'

if sys.platform == 'win32':
    sys.stdout.reconfigure(encoding='utf-8')
    sys.stderr.reconfigure(encoding='utf-8')

class RedirectFinder:
    def __init__(self, input_file: str, max_concurrent: int = 3):
        self.input_file = input_file
        self.max_concurrent = max_concurrent
        self.session = None
        self.semaphore = None
        
        self.processed = 0
        self.skipped = 0
        self.found = 0
        self.not_found = 0
        self.get_params_count = 0
        self.errors = 0
        
        self.url_cache = {}
        
        self.user_agents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Safari/605.1.15',
        ]
        
        self.request_delay = 1.0
        
    def get_headers(self):
        return {
            'User-Agent': random.choice(self.user_agents),
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7',
            'Accept-Encoding': 'gzip, deflate, br',
            'Cache-Control': 'no-cache',
            'Pragma': 'no-cache',
        }
        
    def fix_url(self, url: str) -> str:
        url = re.sub(r'(?<!:)/{2,}', '/', url)
        if url.endswith('/') and url.count('/') > 3:
            url = url.rstrip('/')
        return url
    
    def extract_redirect_segments(self, url: str) -> list:
        parsed = urlparse(url)
        base = f"{parsed.scheme}://{parsed.netloc}"
        path = parsed.path.rstrip('/')
        
        if not path:
            return [base]
        
        segments = [s for s in path.split('/') if s]
        
        test_urls = []
        for i in range(len(segments), -1, -1):
            test_path = '/'.join(segments[:i])
            if test_path:
                test_url = f"{base}/{test_path}"
            else:
                test_url = base
            test_urls.append(test_url)
            
        return test_urls
    
    def is_404_page(self, html_content: str) -> bool:
        if not html_content:
            return True
            
        try:
            soup = BeautifulSoup(html_content, 'html.parser')
            
            title = soup.find('title')
            if title:
                title_text = title.text.lower()
                if any(phrase in title_text for phrase in [
                    'страница не найдена', 'page not found', '404', 'ошибка 404'
                ]):
                    return True
            
            for h1 in soup.find_all('h1'):
                h1_text = h1.get_text().strip().lower()
                if any(phrase in h1_text for phrase in [
                    '404', 'материал не найден', 'страница не найдена',
                    'page not found', 'ничего не найдено', 'ошибка'
                ]):
                    return True
            
            page_text = soup.get_text().lower()
            error_phrases = [
                'материал не найден', 'страница не найдена',
                'ничего не найдено', 'page not found', 'error 404'
            ]
            
            found_phrases = sum(1 for phrase in error_phrases if phrase in page_text)
            if found_phrases >= 2:
                return True
            
            if len(page_text.strip()) < 200 and '404' in page_text:
                return True
                
        except:
            pass
        
        return False
    
    def has_get_params(self, url: str) -> bool:
        return '?' in url
    
    def normalize_catalog_url(self, url: str) -> str:
        if url.rstrip('/') == 'https://sskindustry.ru/catalog':
            return 'https://sskindustry.ru/'
        return url
    
    async def check_url(self, url: str) -> Tuple[int, str]:
        url = self.fix_url(url)
        
        if url in self.url_cache:
            return self.url_cache[url]
        
        for attempt in range(3):
            try:
                headers = self.get_headers()
                
                async with self.session.get(
                    url, 
                    headers=headers, 
                    timeout=aiohttp.ClientTimeout(total=15),
                    allow_redirects=True
                ) as response:
                    content = await response.text()
                    
                    if response.status == 200:
                        is_404 = self.is_404_page(content)
                        if is_404:
                            result = (404, content)
                        else:
                            result = (200, content)
                    else:
                        result = (response.status, content)
                    
                    self.url_cache[url] = result
                    return result
                    
            except asyncio.TimeoutError:
                if attempt < 2:
                    await asyncio.sleep(2)
                    continue
                result = (None, "")
            except Exception as e:
                if attempt < 2:
                    await asyncio.sleep(2)
                    continue
                result = (None, "")
        
        self.url_cache[url] = result
        return result
    
    async def find_working_path(self, broken_url: str) -> Optional[str]:
        test_urls = self.extract_redirect_segments(broken_url)
        
        print(f"\n[*] Проверяю: {broken_url}")
        print(f"    Вариантов для проверки: {len(test_urls)}")
        
        for i, test_url in enumerate(test_urls):
            if i > 0:
                await asyncio.sleep(self.request_delay)
            
            status, content = await self.check_url(test_url)
            
            if status == 200:
                normalized = self.normalize_catalog_url(test_url)
                print(f"  [+] Найден ({i+1}/{len(test_urls)}): {test_url}")
                print(f"       -> Редирект: {normalized}")
                return normalized
            elif status == 404:
                print(f"  [-] 404 ({i+1}/{len(test_urls)}): {test_url}")
            elif status == 503:
                print(f"  [!] 503 ({i+1}/{len(test_urls)}): {test_url}")
                continue
            else:
                print(f"  [?] Статус {status} ({i+1}/{len(test_urls)}): {test_url}")
        
        print(f"  [x] Рабочий путь не найден")
        return None
    
    async def process_row(self, row: int, broken_url: str, ws, ws_params):
        async with self.semaphore:
            try:
                if self.has_get_params(broken_url):
                    ws_params.append([broken_url])
                    self.get_params_count += 1
                    return
                
                working_path = await self.find_working_path(broken_url)
                
                if working_path:
                    ws.cell(row=row, column=2, value=working_path)
                    self.found += 1
                else:
                    ws.cell(row=row, column=2, value="Рабочий путь не найден")
                    self.not_found += 1
                
                self.processed += 1
                
                if self.processed % 10 == 0 or self.processed <= 5:
                    self.print_progress()
                    
            except Exception as e:
                self.errors += 1
                print(f"\n[!] Ошибка в строке {row}: {str(e)}")
    
    def print_progress(self):
        total = self.processed + self.skipped
        percent = (self.processed / max(self.total_urls - self.skipped, 1)) * 100
        print(f"\rПрогресс: {self.processed}/{self.total_urls - self.skipped} ({percent:.1f}%) | "
              f"Найдено: {self.found} | Не найдено: {self.not_found} | "
              f"GET: {self.get_params_count} | В кэше: {len(self.url_cache)}", end='')
    
    async def process_excel(self):
        print(f"\n[1] Загружаю файл: {self.input_file}")
        
        try:
            wb = openpyxl.load_workbook(self.input_file)
        except Exception as e:
            print(f"[!] Ошибка загрузки файла: {str(e)}")
            return
        
        ws = wb.active
        print(f"[2] Активный лист: {ws.title}, строк: {ws.max_row}")
        
        if "GET параметры" in wb.sheetnames:
            ws_params = wb["GET параметры"]
        else:
            ws_params = wb.create_sheet("GET параметры")
            ws_params.append(["Битая ссылка с GET-параметрами"])
        
        if not ws.cell(row=1, column=1).value:
            ws.cell(row=1, column=1, value="Битая ссылка")
        if not ws.cell(row=1, column=2).value:
            ws.cell(row=1, column=2, value="Редирект")
        
        links_to_process = []
        for row in range(2, ws.max_row + 1):
            broken_url = ws.cell(row=row, column=1).value
            existing_redirect = ws.cell(row=row, column=2).value
            
            if not broken_url or not str(broken_url).startswith('http'):
                continue
            
            if existing_redirect and str(existing_redirect).strip():
                redirect_value = str(existing_redirect).strip()
                if redirect_value != "Рабочий путь не найден":
                    self.skipped += 1
                    continue
            
            links_to_process.append((row, str(broken_url)))
        
        self.total_urls = len(links_to_process) + self.skipped
        print(f"[3] Найдено {len(links_to_process)} ссылок для обработки")
        print(f"[4] Пропущено {self.skipped} уже обработанных")
        print(f"{'='*60}")
        
        if len(links_to_process) == 0:
            print("[!] Нет ссылок для обработки!")
            return
        
        connector = aiohttp.TCPConnector(
            limit=self.max_concurrent,
            force_close=True,
            enable_cleanup_closed=True,
        )
        
        timeout = aiohttp.ClientTimeout(total=20, connect=10)
        
        self.session = aiohttp.ClientSession(
            connector=connector,
            timeout=timeout,
        )
        
        self.semaphore = asyncio.Semaphore(self.max_concurrent)
        
        try:
            print("[5] Проверка доступности сайта...")
            status, _ = await self.check_url("https://sskindustry.ru/")
            print(f"[5] Главная страница: {'Доступна' if status == 200 else f'Статус {status}'}")
            
            if status != 200:
                print("[!] ВНИМАНИЕ: Сайт может быть недоступен или блокирует запросы")
                response = input("Продолжить? (y/n): ").lower()
                if response != 'y':
                    return
            
            print(f"\n[6] Начинаю обработку ({len(links_to_process)} ссылок)...")
            print(f"    Задержка между запросами: {self.request_delay}с")
            print(f"    Одновременных запросов: {self.max_concurrent}")
            print()
            
            tasks = [
                self.process_row(row, url, ws, ws_params)
                for row, url in links_to_process
            ]
            
            await asyncio.gather(*tasks, return_exceptions=True)
            
        finally:
            await self.session.close()
        
        output_file = self.input_file.replace('.xlsx', '_обработанный.xlsx')
        print(f"\n\n[7] Сохраняю результат в: {output_file}")
        wb.save(output_file)
        
        print(f"\n{'='*60}")
        print(f"ОБРАБОТКА ЗАВЕРШЕНА")
        print(f"{'='*60}")
        print(f"Всего ссылок в файле: {self.total_urls}")
        print(f"Обработано: {self.processed}")
        print(f"Найдено редиректов: {self.found}")
        print(f"Не найдено: {self.not_found}")
        print(f"Пропущено: {self.skipped}")
        print(f"GET-параметры: {self.get_params_count}")
        print(f"Уникальных URL проверено: {len(self.url_cache)}")
        print(f"Результат сохранен в: {output_file}")

async def main():
    print("="*60)
    print("ПОИСК РЕДИРЕКТОВ ДЛЯ БИТЫХ ССЫЛОК")
    print("="*60)
    
    if len(sys.argv) > 1:
        input_file = sys.argv[1]
    else:
        input_file = PATH
    
    input_file = input_file.strip('"\'')
    
    if not input_file.endswith('.xlsx'):
        input_file += '.xlsx'
    
    if not os.path.exists(input_file):
        print(f"[!] Файл не найден: {input_file}")
        return
    
    max_concurrent = 2
    request_delay = 1.5
    
    print(f"\nНастройки:")
    print(f"- Одновременных запросов: {max_concurrent}")
    print(f"- Задержка между запросами: {request_delay}с")
    print(f"- Примерное время обработки: ~{len(openpyxl.load_workbook(input_file).active['A']) * request_delay / 60:.0f} минут")
    
    try:
        finder = RedirectFinder(input_file, max_concurrent=max_concurrent)
        finder.request_delay = request_delay
        await finder.process_excel()
        
    except KeyboardInterrupt:
        print("\n\n[!] Обработка прервана пользователем")
    except Exception as e:
        print(f"\n[!] Критическая ошибка: {str(e)}")
        import traceback
        traceback.print_exc()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nПрограмма остановлена")
    except Exception as e:
        print(f"\nОшибка: {str(e)}")
    
    input("\nНажмите Enter для выхода...")