todogrep

#!/usr/bin/python

import aiohttp
import argparse
import asyncio
import colorlog
import os
import pyalpm
import re
import tempfile

MIRROR = "https://mirror.pkgbuild.com/"

logger = colorlog.getLogger()
logger.setLevel(colorlog.INFO)

handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter())
logger.addHandler(handler)

async def get_db(db, session, callback):
    full_url = f"{MIRROR}{db}/os/x86_64/{db}.db"
    logger.debug(f"Fetching {full_url}")
    async with session.get(full_url) as resp:
        db_content = await resp.read()
        await callback(db, db_content)

async def get_todo(todo_url, session, callback):
    logger.debug(f"Fetching {todo_url}")
    full_url = f'https://www.archlinux.org{todo_url}json'
    async with session.get(full_url) as resp:
        todo = await resp.json()
        todo["url"] = todo_url
        await callback(todo)

async def analyze_overlaps(todo_callback, db_callback):
    tasks = []
    logger.debug("Fetching todo lists")
    async with aiohttp.ClientSession() as session:
        for db in ("staging", "community-staging", "multilib-staging"):
            task = asyncio.ensure_future(get_db(db, session, db_callback))
            tasks.append(task)

        async with session.get('https://www.archlinux.org/todo') as resp:
            todolist_page = await resp.text()
            todolist = re.findall('(/todo/.*?)"', todolist_page)
            completelist = re.findall('span class="(complete|incomplete)"', todolist_page)

            for todo_url, iscomplete in zip(todolist, completelist):
                if iscomplete == "incomplete":
                    task = asyncio.ensure_future(get_todo(todo_url, session, todo_callback))
                    tasks.append(task)

            responses = await asyncio.gather(*tasks)
            return responses

async def main():
    parser = argparse.ArgumentParser(description='Grep for packages in ongoing todos')
    parser.add_argument('-b', '--pkgbase', action='store_true', help='Look for pkgbases instead of pkgnames')
    parser.add_argument('-d', '--debug', action='store_true', help='Enable debug logging')
    parser.add_argument('package', nargs='+', help='Packages to look for')
    args = parser.parse_args()

    if args.debug:
        logger.setLevel(colorlog.colorlog.logging.DEBUG)

    async def check_todo(todo):
        for p in args.package:
            for package in todo["packages"]:
                if package is None:
                    # Package was removed
                    continue
                if args.pkgbase and package["pkgbase"] == p or package["pkgname"] == p:
                    if todo["kind"] == "Rebuild":
                        logger.error(f"{p} overlaps with {todo['url']}: {todo['name']}")
                    else:
                        logger.warning(f"{p} overlaps with {todo['url']}: {todo['name']}")

    async def check_db(db, db_content):
        with tempfile.TemporaryDirectory() as tmpdirname:
            os.mkdir(f'{tmpdirname}/sync')
            with open(f"{tmpdirname}/sync/{db}.db", "wb") as f:
                f.write(db_content)

            handle = pyalpm.Handle(".", tmpdirname)
            db_handle = handle.register_syncdb(db, 0)
            for package in db_handle.search(""):
                for p in args.package:
                    if args.pkgbase and package.base == p or package.name == p:
                        logger.error(f"{p} is already in [{db}]")

    await analyze_overlaps(check_todo, check_db)

if __name__ == "__main__":
    asyncio.run(main())