backend/src/utils/parse/nodejs_version.py - master - nollieundergrob/PythonPackageStorage

backend/src/utils/parse/nodejs_version.py
127 строк6 KB
nollieundergrob
GENERAL FIX BACKEND
15 авг 2025, 16:01
15 авг 2025, 16:01a8b7dd5
import httpx
import asyncio
import os
import shutil
import logging
import tqdm
import hashlib
import base64
import json
from functools import lru_cache
from concurrent.futures import ThreadPoolExecutor
import argparse
from tenacity import retry, stop_after_attempt, wait_exponential
from pathlib import Path


# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve().parent.parent.parent.parent
from utils.parse.general import split_and_convert

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

visited_packages = set()  # Глобальный set для избежания циклов в рекурсии deps

@lru_cache(maxsize=256)
async def get_versions(package_name: str, json_data: dict = None) -> list[tuple[str, str, str, dict]]:
    """Fetch versions, tarball links, integrity, and dependencies. Use local JSON if provided."""
    if json_data:
        logger.info(f"Using provided JSON for {package_name}")
        versions = [(key, value["dist"]["tarball"], value["dist"].get("integrity", ""), value.get("dependencies", {}))
                    for key, value in json_data.get('versions', {}).items()]
    else:
        url = f"https://registry.npmjs.org/{package_name}"
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.get(url)
            response.raise_for_status()
            data = response.json()
            versions = [(key, value["dist"]["tarball"], value["dist"].get("integrity", ""), value.get("dependencies", {}))
                        for key, value in data['versions'].items()]

    # Sort versions descending
    versions.sort(key=lambda x: split_and_convert(x[0]), reverse=True)
    return versions

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def download_tarball(client: httpx.AsyncClient, tarball_link: str) -> bytes:
    """Download tarball with retry."""
    response = await client.get(tarball_link)
    response.raise_for_status()
    return response.content

def verify_integrity(content: bytes, integrity: str) -> bool:
    """Verify sha512 integrity (format: sha512-base64)."""
    if not integrity.startswith("sha512-"):
        return True  # Skip if no integrity
    expected_hash = integrity.split("-", 1)[1]
    actual_hash = base64.b64encode(hashlib.sha512(content).digest()).decode()
    return actual_hash == expected_hash

def write_file(path: str, content: bytes):
    """Sync file write (to be run in executor)."""
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, 'wb') as file:
        file.write(content)

async def do_install(client: httpx.AsyncClient, library: str, version: str, tarball_link: str, integrity: str, deps: dict, executor: ThreadPoolExecutor, depth: int) -> bool:
    try:
        content = await download_tarball(client, tarball_link)
        if not verify_integrity(content, integrity):
            raise ValueError(f"Integrity check failed for {library}@{version}")

        path = f"{BASE_DIR}/storage/npm/{library}/{version}/{library}-{version}.tgz"
        await asyncio.get_event_loop().run_in_executor(executor, write_file, path, content)
        logger.info(f"Successfully downloaded and verified {library}@{version}")

        # Скачивание зависимостей, если depth > 0
        if depth > 0 and deps:
            for dep_name, dep_version in deps.items():
                if dep_name not in visited_packages:
                    visited_packages.add(dep_name)
                    logger.info(f"Downloading dependencies for {library}@{version}: {dep_name}@{dep_version}")
                    await create_library_versions(dep_name, max_concurrent=10, current_depth=depth-1)
                else:
                    logger.info(f"Skipping already visited dependency: {dep_name}")

        return True
    except Exception as e:
        logger.error(f"Failed to download {library}@{version}: {e}")
        return False

async def create_library_versions(library: str, json_data: dict = None, max_concurrent: int = 10, current_depth: int = 0) -> list[bool]:
    if library in visited_packages:
        logger.info(f"Skipping already visited package: {library}")
        return []

    visited_packages.add(library)

    versions = await get_versions(library, json_data)
    semaphore = asyncio.Semaphore(max_concurrent)
    async with httpx.AsyncClient(timeout=30.0) as client:
        with ThreadPoolExecutor(max_workers=max_concurrent) as executor:
            async def bounded_install(v, link, integ, deps):
                async with semaphore:
                    return await do_install(client, library, v, link, integ, deps, executor, current_depth)

            tasks = [bounded_install(v, link, integ, deps) for v, link, integ, deps in versions]
            results = []
            for future in tqdm.tqdm(asyncio.as_completed(tasks), total=len(tasks), desc=f"Downloading {library}"):
                results.append(await future)
    return results

async def delete_empty_folders(library: str, storage_path: str):
    base_path = f"{storage_path}/{library}"
    for root, dirs, files in os.walk(base_path, topdown=False):
        for dir in dirs:
            dir_path = os.path.join(root, dir)
            if not os.listdir(dir_path):
                shutil.rmtree(dir_path)
                logger.info(f"Deleted empty directory: {dir_path}")


async def create_and_delete(library: str):
    """Create library versions and delete empty folders."""
    await create_library_versions(library)
    await delete_empty_folders(library, BASE_DIR / 'storage'/ 'npm')