The GitHub Gooner: for s|<rIpt kIddi3s who want the bleeding edge

gooner.py

#! /usr/bin/env python
# cython: language_level=3
# distutils: language=c++

"""A fun new way to wreck your system

Usage:
  gitify.py [--github-token=<token>] [--max-tries=<n>]

Options:
  --github-token=<token>    Auth token
  --max-tries=<n>           Number of URLs to try when re-installing a package

See:
  - <https://gist.github.com/callahantiff/0ae1c00df9bec7228be3f6bda5466d73>
"""

import itertools
import os
import re
import subprocess
import time
from typing                                import *

import docopt
from docopt                                import Dict as Arguments
import dotenv
from github                                import Auth
from github                                import Github
from github.GithubException                import RateLimitExceededException
from github.GitRelease                     import GitRelease
from github.PaginatedList                  import PaginatedList
from github.Repository                     import Repository
from multidict                             import MultiDict
from pydantic                              import BaseModel, Field
from structlog                             import get_logger
from tqdm                                  import tqdm

logger                         = get_logger()

class Config(BaseModel):
    github_token:Optional[str] = Field(env='GITHUB_TOKEN', default=None)
    max_tries   :int           = Field(env='MAX_TRIES',    default=10)

def main()->None:
    logger.debug('main()')
    dotenv.load_dotenv()
    arguments:Arguments      = docopt.docopt(__doc__, help=True)
    config   :Config         = Config(**arguments)
    auth     :Optional[Auth]
    ghkwargs :Dict[str,Any]  = {}
    if config.github_token:
        auth                 = Auth.Token(config.github_token)
        ghkwargs['auth']     = auth
    with Github(**ghkwargs) as github:
        for name, version in tqdm(get_packages()):
            reinstall(github, name, version, config.max_tries)
    logger.info('Todo:')
    for name, version in get_packages():
        logger.info('  - %s (%s)', name, version)

dev_pattern = re.compile(r'\.dev|\.alpha|\.beta|\.rc|\.post')

def get_packages()->List[Tuple[str,str]]:
    logger.debug('get_packages()')
    result = subprocess.run(['pip', 'list'], stdout=subprocess.PIPE, text=True, check=True)
    packages = []
    for line in result.stdout.splitlines()[2:]:  # Skip the header lines
        parts = line.split()
        package_name = parts[0]
        package_version = parts[1]
        if dev_pattern.search(package_version):
            continue
        packages.append((package_name, package_version))
    logger.debug('Packages:')
    for package in packages: logger.debug('  - %s', package)
    return packages

def reinstall(github:Github, name:str, version:str, max_tries:int)->None:
    logger.debug('reinstall(name=%s)', name)
    urls = get_url(github, name)
    urls = itertools.islice(urls, max_tries)
    for url in urls:
        try:
            install_from(url)
            if verify_mismatch(name, version):
                break
        except Exception as error:
            logger.error(error)
    logger.warn('Skipping: %s (%s)', name, version)

def get_url(github:Github, name:str)->Generator[str,None]:
    logger.debug('get_url(name=%s)', name)
    keywords    :List[str]                 = [name]
    query       :str                       = '+'.join(keywords)
    logger.debug('Query: %s', query)
    #results     :PaginatedList[Repository] = github.search_repositories(query, 'stars', 'desc', )
    qualifiers  :Dict[str,str]             = {
        #'in': 'name',
        #'in': 'readme',
        #'in': 'description',
    }
    results     :PaginatedList[Repository] = github.search_repositories(query, **qualifiers)

    #for _result in tqdm(results, total=results.totalCount):
    for _result in results:
    
        #try:
        logger.debug('Result: %s', _result.clone_url)
        yield _result.clone_url
        time.sleep(2)
        #except RateLimitExceededException:
        #    time.sleep(60)
        #    # TODO this looks janky

def install_from(url:str)->None:
    logger.debug('install_from(url=%s)', url)
    subprocess.run(['pip', 'install', f'git+{url}'], check=True)

def verify_mismatch(name:str, old_version:str)->bool:
    new_version:str = get_installed_version(name)
    return old_version != new_version

def get_installed_version(package_name: str) -> Optional[str]:
    """Get the currently installed version of a package."""
    result = subprocess.run(['pip', 'show', package_name], stdout=subprocess.PIPE, text=True, check=True)
    for line in result.stdout.splitlines():
        if line.startswith('Version:'):
            return line.split()[1]  # Return the version number
    #return None  # Return None if the package is not found
    raise Exception(f'Not found {package_name}')

if __name__ == '__main__':
    main()
__author__:str = 'you.com'