Scriptpara descargar leyes de mexico

leyes.py

import typer
import httpx
from bs4 import BeautifulSoup
from rich.console import Console
from rich.table import Table
from rich.progress import Progress, SpinnerColumn, TextColumn
from typing import List, Dict
from pydantic import BaseModel, HttpUrl
import asyncio
import os
from pathlib import Path
import shutil
from pypdf import PdfReader
import aiofiles
import time


class LeyVigente(BaseModel):
    id: str
    descripcion: str
    ultima_reforma: str
    texto_pdf: HttpUrl
    texto_doc: HttpUrl
    texto_pdf_mov: HttpUrl


app = typer.Typer()
console = Console()
BASE_URL = "https://www.diputados.gob.mx/LeyesBiblio"
LEYES_DIR = Path(".leyes")


def confirm_directory_cleanup() -> bool:
    """Pregunta al usuario si desea borrar el contenido del directorio .leyes."""
    if LEYES_DIR.exists():
        console.print("[yellow]El directorio .leyes ya existe.[/yellow]")
        console.print("[red]Se borrará todo el contenido del directorio.[/red]")
        respuesta = typer.confirm("¿Desea continuar?")
        if not respuesta:
            console.print("[yellow]Operación cancelada.[/yellow]")
            return False
        
        # Borrar todo el contenido del directorio
        for archivo in LEYES_DIR.glob("*"):
            if archivo.is_file():
                archivo.unlink()
            elif archivo.is_dir():
                shutil.rmtree(archivo)
        console.print("[green]Directorio .leyes limpiado.[/green]")
    
    return True


async def download_pdf(client: httpx.AsyncClient, ley: LeyVigente, progress: Progress) -> None:
    """Descarga el PDF móvil de una ley."""
    task_id = progress.add_task(f"Descargando {ley.descripcion}", total=None)
    
    try:
        response = await client.get(str(ley.texto_pdf_mov))
        response.raise_for_status()
        
        # Crear el directorio si no existe
        LEYES_DIR.mkdir(exist_ok=True)
        
        # Obtener el nombre del archivo de la URL
        url_filename = str(ley.texto_pdf_mov).split('/')[-1]
        # Generar nombre de archivo seguro
        filename = f"{ley.id}_{url_filename}"
        filepath = LEYES_DIR / filename
        
        # Guardar el archivo
        with open(filepath, "wb") as f:
            f.write(response.content)
        
        progress.update(task_id, completed=True, description=f"[green]✓[/green] [bold white]{ley.descripcion}[/bold white]")
    except Exception as e:
        progress.update(task_id, description=f"[red]✗[/red] [yellow]{ley.descripcion}: {str(e)}[/yellow]")


async def download_all_pdfs(laws: List[LeyVigente]) -> None:
    """Descarga todos los PDFs móviles de manera concurrente."""
    async with httpx.AsyncClient(verify=False) as client:
        with Progress(
            SpinnerColumn(),
            TextColumn("[progress.description]{task.description}"),
            console=console
        ) as progress:
            tasks = [download_pdf(client, ley, progress) for ley in laws]
            await asyncio.gather(*tasks)


@app.command()
def descargar():
    """Descarga los PDFs móviles de todas las leyes vigentes."""
    # Verificar y limpiar el directorio si es necesario
    if not confirm_directory_cleanup():
        return
    
    console.print("[yellow]Obteniendo lista de leyes...[/yellow]")
    laws = get_laws()
    
    console.print(f"[green]Se encontraron {len(laws)} leyes vigentes[/green]")
    console.print(f"[yellow]Iniciando descargas en el directorio {LEYES_DIR}[/yellow]")
    
    start_time = time.time()
    asyncio.run(download_all_pdfs(laws))
    end_time = time.time()
    
    console.print(f"[green]✓ Todas las descargas completadas en {end_time - start_time:.2f} segundos[/green]")


def get_laws() -> List[LeyVigente]:
    """Descarga la tabla de leyes del sitio web de la cámara de diputados"""
    
    
    with httpx.Client(verify=False) as client:
        response = client.get(f"{BASE_URL}/index.htm")
        response.raise_for_status()
        
        soup = BeautifulSoup(response.text, 'html.parser')
        for section in soup.find_all('section'):
            if "LEYES FEDERALES VIGENTES" in section.text:
                break
        
        table = soup.select_one('section table')
        
        laws = []
        for row in table.find_all('tr')[1:]:  # Saltar la fila de encabezado
            cols = row.find_all('td')
            if len(cols) >= 4:
                # Verificar si es una ley abrogada
                if cols[0].get_text(strip=True) == "A":
                    continue  # Por el momento no procesaremos leyes abrogadas
                
                # Extraer los enlaces de la cuarta columna
                enlaces = {}
                for link in cols[3].find_all('a'):
                    url = link['href']
                    match url.split('/')[0]:
                        case 'doc':
                            enlaces['texto_doc'] = f'{BASE_URL}/{url}'
                        case 'pdf':
                            enlaces['texto_pdf'] = f'{BASE_URL}/{url}'
                        case 'pdf_mov':
                            enlaces['texto_pdf_mov'] = f'{BASE_URL}/{url}'
                if not enlaces:
                    # Si no hay enlaces, continuar con la siguiente ley
                    continue
                law = LeyVigente(
                    id=cols[0].get_text(strip=True),
                    descripcion=cols[1].get_text(strip=True),
                    ultima_reforma=cols[2].get_text(strip=True),
                    **enlaces
                )
                laws.append(law)
        
        return laws

def display_laws(laws: List[LeyVigente]):
    """Muestra las leyes en una tabla formateada con rich."""
    table = Table(title="Leyes Federales")
    table.add_column("No.", style="cyan")
    table.add_column("Ley", style="magenta")
    table.add_column("Última Reforma", style="yellow")
    table.add_column("Enlaces", style="blue")
    
    for law in laws:
        # Formatear los enlaces para mostrarlos en la tabla
        enlaces_str = f"PDF: {law.texto_pdf}\nDOC: {law.texto_doc}\nPDF Móvil: {law.texto_pdf_mov}"
        table.add_row(
            law.id,
            law.descripcion,
            law.ultima_reforma,
            enlaces_str
        )
    
    console.print(table)

@app.command()
def listado():
    """Lista todas las leyes federales del sitio web del Congreso Mexicano."""

    laws = get_laws()
    display_laws(laws)

    # try:
    #     laws = get_laws()
    #     display_laws(laws)
    # except Exception as e:
    #     console.print(f"[red]Error: {str(e)}[/red]")
    #     raise typer.Exit(1)


async def extract_text_from_pdf(pdf_path: Path, progress: Progress, semaphore: asyncio.Semaphore) -> None:
    """Extrae el texto de un archivo PDF y lo guarda en un archivo .txt."""
    async with semaphore:  # Limitar el número de procesamientos simultáneos
        task_id = progress.add_task(f"Procesando {pdf_path.name}", total=None)
        
        try:
            # Leer el PDF
            reader = PdfReader(pdf_path)
            text = ""
            for page in reader.pages:
                text += page.extract_text() + "\n"
            
            # Crear el nombre del archivo de texto
            txt_path = pdf_path.with_suffix('.txt')
            
            # Guardar el texto de manera asíncrona
            async with aiofiles.open(txt_path, 'w', encoding='utf-8') as f:
                await f.write(text)
            
            progress.update(task_id, completed=True, description=f"[green]✓[/green] [bold white]{pdf_path.name}[/bold white]")
        except Exception as e:
            progress.update(task_id, description=f"[red]✗[/red] [yellow]{pdf_path.name}: {str(e)}[/yellow]")


async def extract_all_texts() -> None:
    """Extrae el texto de todos los PDFs en el directorio .leyes."""
    pdf_files = list(LEYES_DIR.glob("*.pdf"))
    
    if not pdf_files:
        console.print("[red]No se encontraron archivos PDF en el directorio .leyes[/red]")
        return
    
    console.print(f"[yellow]Se encontraron {len(pdf_files)} archivos PDF para procesar[/yellow]")
    console.print("[yellow]Procesando máximo 5 archivos a la vez...[/yellow]")
    
    # Crear un semáforo que limite a 5 procesamientos simultáneos
    semaphore = asyncio.Semaphore(5)
    
    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        console=console
    ) as progress:
        tasks = [extract_text_from_pdf(pdf, progress, semaphore) for pdf in pdf_files]
        await asyncio.gather(*tasks)


@app.command()
def extraer_texto():
    """Extrae el texto de todos los PDFs en el directorio .leyes."""
    if not LEYES_DIR.exists():
        console.print("[red]El directorio .leyes no existe[/red]")
        return
    
    console.print("[yellow]Iniciando extracción de texto...[/yellow]")
    
    start_time = time.time()
    asyncio.run(extract_all_texts())
    end_time = time.time()
    
    console.print(f"[green]✓ Extracción de texto completada en {end_time - start_time:.2f} segundos[/green]")


if __name__ == "__main__":
    app()