mirror of
https://github.com/ail-project/ail-framework.git
synced 2024-11-14 02:28:23 +00:00
134 lines
4.5 KiB
Python
134 lines
4.5 KiB
Python
import json
|
|
import requests
|
|
import subprocess
|
|
import re
|
|
import os
|
|
from time import sleep
|
|
from typing import List, Optional
|
|
from pathlib import Path
|
|
|
|
BUILD_PATH = "/opt/ailbuilder/build"
|
|
|
|
class Repo:
|
|
"""Base class for repository tracking and update checking."""
|
|
|
|
def __init__(self, id: str, args: List[str], name: str, outputdir: str) -> None:
|
|
self.id = id
|
|
self.args = args
|
|
self.name = name
|
|
self.outputdir = outputdir
|
|
self.last_seen_update = None
|
|
|
|
def _check_for_new_update(self) -> bool:
|
|
latest_update = self._get_latest_update()
|
|
if latest_update and (latest_update != self.last_seen_update):
|
|
print(f"New update found for {self.id}")
|
|
self.last_seen_update = latest_update
|
|
return True
|
|
return False
|
|
|
|
def _get_latest_update(self):
|
|
raise NotImplementedError
|
|
|
|
def _save_state(self):
|
|
try:
|
|
with open(f'{BUILD_PATH}/systemd/state.json', 'r') as file:
|
|
states = json.load(file)
|
|
except FileNotFoundError:
|
|
states = {}
|
|
|
|
states[self.id] = self.last_seen_update
|
|
|
|
with open(f'{BUILD_PATH}/systemd/state.json', 'w') as file:
|
|
json.dump(states, file)
|
|
|
|
def load_state(self):
|
|
try:
|
|
with open(f'{BUILD_PATH}/systemd/state.json', 'r') as file:
|
|
states = json.load(file)
|
|
except FileNotFoundError:
|
|
states = {}
|
|
|
|
self.last_seen_update = states.get(self.id, None)
|
|
|
|
def build(self) -> None:
|
|
if self._check_for_new_update():
|
|
try:
|
|
cmd = [f'{BUILD_PATH}/build.sh'] + self.args + ["-o", self.outputdir]
|
|
print(f"Running {cmd}")
|
|
result = subprocess.run(cmd, check=False)
|
|
if result.returncode != 0:
|
|
print(f"Failed to run {cmd} for {self.id}")
|
|
return
|
|
most_recent_dir = max((d for d in Path(self.outputdir).iterdir() if d.is_dir()), key=os.path.getctime, default=None)
|
|
relative_path = most_recent_dir.relative_to(Path(self.outputdir))
|
|
if os.path.exists(f"{self.outputdir}/latest_{self.name}"):
|
|
os.remove(f"{self.outputdir}/latest_{self.name}")
|
|
os.symlink(relative_path, f"{self.outputdir}/latest_{self.name}")
|
|
print(f"Created symlink {self.outputdir}/latest_{self.name} to {relative_path}")
|
|
self._save_state()
|
|
except Exception as e:
|
|
print(f"Failed to run {cmd} for {self.id}: {e}")
|
|
|
|
class GitHub(Repo):
|
|
"""Class for tracking GitHub repositories."""
|
|
|
|
def __init__(self, id: str, mode: str, args: List[str], name: str, outputdir: str) -> None:
|
|
super().__init__(id, args, name, outputdir)
|
|
self.mode = mode
|
|
|
|
def _get_latest_update(self) -> Optional[str]:
|
|
print(f"Fetching {self.mode} for {self.id}")
|
|
url=f'https://api.github.com/repos/{self.id}/{self.mode}'
|
|
response = requests.get(url)
|
|
if response.status_code == 200:
|
|
return response.json()[0]['sha']
|
|
else:
|
|
print(f"Failed to fetch {self.mode} for {self.id}")
|
|
return None
|
|
|
|
class APT(Repo):
|
|
"""Class for tracking APT packages."""
|
|
|
|
def __init__(self, id: str, args: List[str], name: str, outputdir: str) -> None:
|
|
super().__init__(id, args, name, outputdir)
|
|
|
|
def _get_latest_update(self) -> Optional[str]:
|
|
try:
|
|
cmd = ["apt-cache", "policy", self.id]
|
|
print (f"Running {cmd}")
|
|
output = subprocess.check_output(cmd).decode('utf-8')
|
|
match = re.search(r'Candidate: (\S+)', output)
|
|
if match:
|
|
return match.group(1)
|
|
else:
|
|
return None
|
|
except:
|
|
return None
|
|
|
|
def main():
|
|
with open(f'{BUILD_PATH}/conf/tracker.json') as f:
|
|
config = json.load(f)
|
|
|
|
repos = []
|
|
for repo in config["github"]:
|
|
repos.append(GitHub(repo["id"], repo["mode"], repo["args"], repo["name"], config["outputdir"]))
|
|
|
|
aptpkg = []
|
|
for package in config["apt"]:
|
|
aptpkg.append(APT(package["id"], package["args"], package["name"], config["outputdir"]))
|
|
|
|
for repo in repos + aptpkg:
|
|
if config["sign"]:
|
|
repo.args.append("-s")
|
|
repo.load_state()
|
|
|
|
while True:
|
|
for repo in repos:
|
|
repo.build()
|
|
for package in aptpkg:
|
|
package.build()
|
|
sleep(config["check_interval"])
|
|
|
|
if __name__ == "__main__":
|
|
main()
|