ci-tools/lintian-ppa

#!/usr/bin/env python3
#
# Copyright (C) 2024 Simon Quigley <tsimonq2@ubuntu.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

import argparse
import logging
import os
import shutil
import subprocess
import tempfile
import uuid
from common import clean_old_logs
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
from datetime import datetime, timedelta, timezone
from debian.deb822 import Changes
from launchpadlib.launchpad import Launchpad
from pathlib import Path

BASE_OUTPUT_DIR = "/srv/lubuntu-ci/output/"
LOG_DIR = os.path.join(BASE_OUTPUT_DIR, "logs/lintian/")

parser = argparse.ArgumentParser(description="")
parser.add_argument("--user", "-u", required=True)
parser.add_argument("--ppa", "-p", required=True)
parser.add_argument("--ppa2", "-p2")
args = parser.parse_args()

os.makedirs(LOG_DIR, exist_ok=True)
current_time = datetime.utcnow().strftime("%H-%M-%S")
log_file = os.path.join(LOG_DIR, f"{current_time}.log")
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler(log_file)
    ]
)
logger = logging.getLogger("TimeBasedLogger")

launchpad = Launchpad.login_with("lintian-ppa", "production", version="devel")
ubuntu = launchpad.distributions["ubuntu"]
current_series = ubuntu.current_series

user = launchpad.people[args.user]
ppa = user.getPPAByName(distribution=ubuntu, name=args.ppa)
if args.ppa2:
    ppa2 = user.getPPAByName(distribution=ubuntu, name=args.ppa)

lintian = os.path.join(BASE_OUTPUT_DIR, "lintian")
lintian_tmp = os.path.join(BASE_OUTPUT_DIR, f".lintian.tmp.{str(uuid.uuid4())[:8]}")
if not os.path.exists(lintian):
    os.mkdir(lintian)
if os.path.exists(lintian_tmp):
    shutil.rmtree(lintian_tmp)
os.mkdir(lintian_tmp)

def rsync(source, destination):
    src = Path(source)
    dst = Path(destination)
    dst.mkdir(parents=True, exist_ok=True)

    for item in src.iterdir():
        src_path = item
        dst_path = dst / item.name

        if src_path.is_symlink():
            if dst_path.exists() or dst_path.is_symlink():
                dst_path.unlink()
            os.symlink(os.readlink(src_path), dst_path)
        elif src_path.is_dir():
            shutil.copytree(src_path, dst_path, symlinks=True, dirs_exist_ok=True)
        else:
            shutil.copy2(src_path, dst_path)

def process_sources(url):
    tmpdir = os.path.join(BASE_OUTPUT_DIR, f".lintian.tmp.{str(uuid.uuid4())[:8]}")
    os.mkdir(tmpdir)
    changes_file = url.split("/")[-1]
    logging.info(f"Downloading {changes_file} and friends via dget")
    dget_command = ["dget", "-u", url]
    result = subprocess.run(dget_command, cwd=tmpdir, capture_output=True)

    with open(os.path.join(tmpdir, changes_file), "r") as f:
        changes_obj = Changes(f)
    source = changes_obj["Source"]
    arch = changes_obj["Architecture"].replace("all", "").replace("_translations", "").split(" ")[0].strip()

    if arch == "":
        return

    logging.info(f"Running Lintian for {source} on {arch}")
    lintian_command = ["lintian", "-EvIL", "+pedantic", changes_file]
    result = subprocess.run(lintian_command, cwd=tmpdir, capture_output=True)
    stderr = result.stderr.decode("utf-8").strip()
    stdout = result.stdout.decode("utf-8").strip()

    if stderr == stdout:
        lintian_output = stderr
    elif stderr != "" and stdout == "":
        lintian_output = stderr
    elif stderr == "" and stdout != "":
        lintian_output = stdout
    else:
        lintian_output = f"{stderr}\n{stdout}"

    output_path = os.path.join(lintian_tmp, source)
    if not os.path.exists(output_path):
        os.mkdir(output_path)

    with open(os.path.join(output_path, f"{arch}.txt"), "w") as f:
        f.write(lintian_output)

    shutil.rmtree(tmpdir)

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = set()

    def main_source_iter():
        last_run_file = os.path.join(lintian, ".LAST_RUN")
        last_run_datetime = datetime.now(timezone.utc) - timedelta(days=365)
        if os.path.exists(last_run_file):
            with open(last_run_file, "r") as file:
                last_run_time = file.read().strip()

            last_run_datetime = datetime.fromisoformat(last_run_time)
            last_run_datetime = last_run_datetime.replace(tzinfo=timezone.utc)
            logging.info(f"Last run: {last_run_datetime}")

        with open(last_run_file, "w") as file:
            current_time = datetime.now(timezone.utc).isoformat()
            file.write(current_time)

        for source in ppa.getPublishedSources(status="Published", distro_series=current_series):
            for build in source.getBuilds():
                if build.buildstate == "Successfully built" and build.datebuilt >= last_run_datetime:
                    futures.add(executor.submit(process_sources, build.changesfile_url))

    futures.add(executor.submit(main_source_iter))

    while futures:
        done, not_done = wait(futures, return_when=FIRST_COMPLETED)

        for future in done:
            try:
                result = future.result()
            except Exception as e:
                logging.exception("Task generated an exception:")
            finally:
                futures.remove(future)

rsync(lintian_tmp, lintian)
shutil.rmtree(lintian_tmp)
clean_old_logs(LOG_DIR)
logging.info("Done")