#!/usr/bin/env python3 # # Copyright (C) 2024 Simon Quigley # # 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 . 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/" parser = argparse.ArgumentParser(description="") parser.add_argument("--user", "-u", required=True) parser.add_argument("--ppa", "-p", required=True) parser.add_argument("--ppa2", "-p2") parser.add_argument("--override-output", "-o") args = parser.parse_args() if args.override_output: BASE_OUTPUT_DIR = args.override_output LOG_DIR = os.path.join(BASE_OUTPUT_DIR, "logs/lintian/") 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")