AutoBlockLists/main.py at main · hfsotomayor/AutoBlockLists · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
import os
import yaml
import requests
import re
import time
import warnings
import urllib3
import logging
from datetime import datetime, timedelta
from collections import defaultdict

# Configuración de logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('autoblocklist.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# Suprimir advertencias de SSL
warnings.filterwarnings('ignore', category=urllib3.exceptions.NotOpenSSLWarning)

class BlockListManager:
    def __init__(self):
        self.config = self.load_config()
        self.input_dir = self.config.get('input_dir', 'input')
        self.output_dir = self.config.get('output_dir', 'output')
        os.makedirs(self.input_dir, exist_ok=True)
        os.makedirs(self.output_dir, exist_ok=True)
        logger.info("BlockListManager iniciado")
        logger.info(f"Directorio de entrada: {self.input_dir}")
        logger.info(f"Directorio de salida: {self.output_dir}")

    def load_config(self):
        with open("config.yaml", 'r') as f:
            return yaml.safe_load(f)

    def process_lists(self):
        domains_by_category = defaultdict(set)
        all_domains = set()

        logger.info("Iniciando procesamiento de listas")
        for source in self.config['blocklist_urls']:
            try:
                url = source['url']
                filename = os.path.join(self.input_dir, url.split('/')[-1])

                logger.info(f"Descargando lista desde: {url}")
                response = requests.get(url)
                response.raise_for_status()

                # Guardar archivo descargado
                with open(filename, 'w') as f:
                    f.write(response.text)
                logger.info(f"Lista guardada en: {filename}")

                domains = self.extract_domains(response.text)

                logger.info(f"Dominios encontrados en {source['url']}: {len(domains)}")
                for category in source['categories']:
                    prev_count = len(domains_by_category[category])
                    domains_by_category[category].update(domains)
                    new_count = len(domains_by_category[category])
                    logger.info(f"Categoría {category}: +{new_count - prev_count} dominios")

                prev_total = len(all_domains)
                all_domains.update(domains)
                logger.info(f"Nuevos dominios únicos: {len(all_domains) - prev_total}")

            except requests.RequestException as e:
                logger.error(f"Error descargando {source['url']}: {e}")

        logger.info(f"Total dominios únicos: {len(all_domains)}")
        return all_domains, domains_by_category

    def extract_domains(self, content):
        domains = set()
        for line in content.splitlines():
            line = line.strip()
            if line and not line.startswith('#'):
                domain = re.sub(r'^(0\.0\.0\.0\s+|127\.0\.0\.1\s+|::1\s+|fe80::1%lo0\s+)', '', line)
                domain = re.sub(r'^(https?://)?(www\.)?|(\s+|\t).*', '', domain).split('/')[0]
                if domain and domain != "0.0.0.0" and '.' in domain:
                    domains.add(domain)
        return domains

    def save_lists(self, all_domains, domains_by_category):
        timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
        formats = self.config.get('output_formats', ['plain'])

        for format_type in formats:
            format_dir = os.path.join(self.output_dir, format_type)
            os.makedirs(format_dir, exist_ok=True)

            # Guardar lista total
            main_file = os.path.join(format_dir, f"AutoBlockLists-{format_type}-Total-{timestamp}.txt")
            self.save_file(main_file, all_domains, format_type)
            logger.info(f"Lista total guardada ({format_type}): {main_file}")

            # Guardar listas por categoría
            for category, domains in domains_by_category.items():
                cat_file = os.path.join(format_dir, f"AutoBlockLists-{format_type}-{category}-{timestamp}.txt")
                self.save_file(cat_file, domains, format_type)
                logger.info(f"Lista categoría {category} guardada ({format_type}): {cat_file}")

    def save_file(self, filepath, domains, format_type):
        with open(filepath, 'w') as f:
            f.write(f"# AutoBlockLists - Generated: {datetime.now()}\n")
            f.write(f"# Total domains: {len(domains)}\n")
            f.write(f"# Format: {format_type}\n\n")

            for domain in sorted(domains):
                if format_type == 'plain':
                    f.write(f"{domain}\n")
                elif format_type == 'hosts':
                    f.write(f"0.0.0.0 {domain}\n")
                elif format_type == 'adguard':
                    f.write(f"||{domain}^$important\n")

def main():
    manager = BlockListManager()
    while True:
        logger.info("=== Iniciando nueva actualización ===")
        start_time = time.time()

        all_domains, domains_by_category = manager.process_lists()
        manager.save_lists(all_domains, domains_by_category)

        execution_time = time.time() - start_time
        logger.info(f"Actualización completada en {execution_time:.2f} segundos")
        logger.info(f"Esperando {manager.config['update_frequency']} segundos para la próxima actualización")
        time.sleep(manager.config['update_frequency'])

if __name__ == "__main__":
    main()