Muzifs/PortProton_2.0
1
0
Fork 0
forked from CastroFidel/PortProton_2.0
Code Pull Requests Activity

Merge branch 'main' of github.com:xpamych/PortProton_2.0 into xpamych-main

Browse Source
This commit is contained in:
Mikhail Tergoev
2025-03-07 02:00:32 +03:00
parent 9dc6144ffd a9852dbffa
commit be95574b16
8 changed files with 272 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
modules/downloader.py
View File

@ -6,7 +6,7 @@ from .files_worker import *
def try_download(url, save_path=None):
"""
Скачивает файл по указанному URL с отображением прогресса.
Скачивает файл по-указанному URL с отображением прогресса.
:param url: URL файла для скачивания.
:param save_path: Путь для сохранения файла. Если None или директория, то используется имя файла из URL.

51
modules/files_worker.py
View File

@ -31,10 +31,10 @@ def try_force_link_file(source, link):
os.symlink(source, link)
except Exception as e:
print(f"failed to create link for file: {e}")
log.error(f"failed to create link for file: {e}")
def try_remove_file(path):
def try_remove_file(file_path):
if os.path.exists(file_path) and os.path.isfile(file_path):
try:
os.remove(file_path)
@ -60,21 +60,55 @@ def try_force_link_dir(path, link):
os.symlink(path, link)
except Exception as e:
print(f"failed to create link for file: {e}")
log.error(f"failed to create link for file: {e}")
def replace_file(file_path, file_name): # функция замены файла (сначала запись во временный файл, потом замена)
try:
if os.path.exists(file_name) and os.path.getsize(file_name) > 0:
os.replace(file_name, file_path) # Меняем местами файлы, если временный файл не пуст
log.info(f"Данные успешно обновлены в {file_path}.")
else:
log.warning(f"Временный файл {file_name} пуст, замена в {file_path} не выполнена.")
if os.path.exists(file_name):
os.remove(file_name) # Удаляем пустой временный файл
except Exception as e:
log.error(f"Ошибка при замене файла: {e}")
def try_write_temp_file(file_path, file_name): # функция записи в tmp
try:
with open(file_path, 'w') as file:
file.write("\n".join(file_name)) # Записываем все имена файлов во временный файл
except Exception as e:
log.error(f"Ошибка при записи во временный файл {file_path}: {e}")
def try_remove_dir(path):
if os.path.exist(path) and os.path.isdir(path):
if os.path.exists(path) and os.path.isdir(path):
try:
shutil.rmtree(path)
except Exception as e:
log.error(f"failed to remove directory: {e}")
def get_last_modified_time(file_path): # функция получения времени последнего изменения файла
try:
return os.path.getmtime(file_path)
except FileNotFoundError:
log.warning(f"Файл не найден: {file_path}")
return None
except Exception as e:
log.error(f"Ошибка при получении времени изменения файла {file_path}: {e}")
return None
def unpack(archive_path, extract_to=None):
# Проверяем, существует ли архив
if not os.path.isfile(archive_path):
log.error(f"Архив {archive_path} не найден.")
return False
try:
if extract_to is None:
# TODO: перенести распаковку по умолчанию в tmp
extract_to = os.path.dirname(archive_path)
elif not os.exists.isdir(extract_to):
elif not os.path.isdir(extract_to):
create_new_dir(extract_to)
with tarfile.open(archive_path, mode="r:*") as tar:
@ -83,8 +117,15 @@ def unpack(archive_path, extract_to=None):
log.info(f"Архив {archive_path} успешно распакован в {full_path}")
except tarfile.TarError as e:
log.error(f"Ошибка при распаковке архива {archive_path}: {e}")
return False
except PermissionError:
log.error(f"Ошибка доступа к файлу {archive_path}. Убедитесь, что у вас есть права на чтение.")
return False
except Exception as e:
log.error(f"Неизвестная ошибка: {e}")
return False
return True
def check_hash_sum(check_file, check_sum):
if check_sum and isinstance(check_sum, str):

37
modules/log.py
View File

@ -2,8 +2,7 @@ import logging
import sys
class ColoredFormatter(logging.Formatter):
# ANSI escape sequences for colors
COLORS = {
COLORS = { # ANSI escape sequences for colors
'DEBUG': '\033[35m', # Purple
'INFO': '\033[36m', # Green
'WARNING': '\033[33m', # Yellow
@ -21,19 +20,33 @@ class ColoredFormatter(logging.Formatter):
sys.exit(1)
return formatted_message
# Настраиваем логирование
log = logging.getLogger()
# TODO: добавить case с переменной для управление уровнем
log.setLevel(logging.DEBUG)
# Создаем консольный обработчик
handler = logging.StreamHandler()
log = logging.getLogger() # Настраиваем логирование
def set_logging_level(level_string):
levels = {
'DEBUG': logging.DEBUG,
'INFO': logging.INFO,
'WARNING': logging.WARNING,
'ERROR': logging.ERROR,
'CRITICAL': logging.CRITICAL
}
level = levels.get(level_string, logging.WARNING) # Уровень по умолчанию
if level == logging.WARNING:
print(f"Неизвестный уровень логирования: {level_string}. Устанавливается уровень WARNING.")
return level
log_level_input = 'DEBUG' # Задаем уровень логирования через функцию
log.setLevel(set_logging_level(log_level_input))
handler = logging.StreamHandler() # Создаем консольный обработчик
handler.setFormatter(ColoredFormatter('%(levelname)s: %(message)s'))
# Создаем файловый обработчик
# TODO: добавить условие для управления перемееной пути сохранения лога
file_handler = logging.FileHandler('app.log')
# TODO: добавить условие для управления переменой пути сохранения лога
log_file_path = 'app.log' # Это может быть переменная, установленная пользователем
file_handler = logging.FileHandler(log_file_path) # Создаем файловый обработчик
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s: %(message)s'))
log.addHandler(file_handler)
log.addHandler(file_handler)
log.addHandler(handler)

109
modules/source_fetcher.py Normal file
View File

@ -0,0 +1,109 @@
import re
import requests
import time
from modules.downloader import try_download
from modules.files_worker import *
count_wines = 25 # Количество элементов для записи в .tmp файл
repos = { # Список репозиториев для обработки и их короткие имена
"GloriousEggroll/proton-ge-custom": "proton-ge-custom",
"Kron4ek/Wine-Builds": "Kron4ek",
"GloriousEggroll/wine-ge-custom": "wine-ge-custom",
"CachyOS/proton-cachyos": "proton-cachyos",
"Castro-Fidel/wine_builds": "LG"
}
def source_list_checker(tmp_path): # Проверка наличия и обновления файлов со списками исходников
for repo, short_name in repos.items():
output_file = os.path.join(tmp_path, f"{short_name}.tmp")
if not os.path.exists(output_file):
log.info(f"Файл {output_file} не существует. Получаем данные из репозитория.")
source_list_downloader(repo, tmp_path, short_name, output_file)
continue # Переходим к следующему репозиторию
if os.path.getsize(output_file) == 0: # Проверяем, является ли файл пустым
log.info(f"Файл {output_file} пуст. Обновляем данные.")
source_list_downloader(repo, tmp_path, short_name, output_file)
continue # Переходим к следующему репозиторию
last_modified_time = get_last_modified_time(output_file) # Получаем время последнего изменения файла
if last_modified_time is None: # Если время не удалось получить, пробуем обновить файл
log.info(f"Не удалось получить время последнего изменения для {output_file}. Попытаемся обновить.")
source_list_downloader(repo, tmp_path, short_name, output_file)
continue # Переходим к следующему репозиторию
current_time = time.time()
if current_time - last_modified_time >= 10800: # 10800 секунд = 3 часа, проверяем, устарел ли файл
log.info(f"Файл {output_file} устарел. Обновляем данные.")
source_list_downloader(repo, tmp_path, short_name, output_file)
else:
log.info(f"Файл {output_file} существует и был обновлён менее 3 часов назад. Используем кэшированные данные.")
def source_list_downloader(repo, tmp_path, short_name, output_file):
url = f"https://api.github.com/repos/{repo}/releases?per_page={count_wines}"
temp_file = os.path.join(tmp_path, f"{short_name}.tmp.new") # Временный файл
try:
response = requests.get(url)
response.raise_for_status() # Возбудим исключение для ошибок HTTP
releases = response.json()
log.debug(f"Ответ API: {releases}")
tar_files = [] # Проверяем каждый релиз
for release in releases:
assets = release.get('assets', [])
for asset in assets:
asset_name = asset['name']
if (
(re.search(r'(wine|proton)', asset_name, re.IGNORECASE) or
re.search(r'^GE-Proton\d+-\d+\.tar\.gz$', asset_name) or
re.search(r'^GE-Proton\d+(-\d+)?\.tar\.xz$', asset_name)) and
(asset_name.endswith('.tar.gz') or asset_name.endswith('.tar.xz'))
):
tar_files.append(asset_name) # Собираем все подходящие файлы
log.debug(f"Найденный файл: {asset_name}")
if not tar_files:
log.warning(f"Нет подходящих файлов в репозитории {repo}.")
return # Выходим из функции, если нет файлов
with open(temp_file, 'w') as file: # Записываем найденные файлы во временный файл
file.write("\n".join(tar_files))
log.info(f"Данные успешно записаны в временный файл {temp_file}.")
if os.path.exists(temp_file): # Если запись прошла успешно, заменяем основной файл
os.replace(temp_file, output_file)
log.info(f"Файл {output_file} успешно обновлен.")
except requests.exceptions.RequestException as e:
log.error(f"Ошибка при получении данных из {repo}: {str(e)}")
def get_sources(args, tmp_path, dist_path):
os.makedirs(tmp_path, exist_ok=True)
source_list_checker(tmp_path)
if args:
for arg in args:
for repo, short_name in repos.items(): # определяем короткое имя репозитория
tmp_file_path = os.path.join(tmp_path, f"{short_name}.tmp")
if os.path.exists(tmp_file_path): # проверяем наличие файла
with open(tmp_file_path, 'r') as file:
all_tar_gz_files = file.read().splitlines()
for file_to_download in all_tar_gz_files: # Ищем совпадение в файле
if arg in file_to_download:
log.info(f"Найдено совпадение для '{arg}': {file_to_download} в файле {tmp_file_path}")
url = f"https://github.com/{repo}/releases/latest/download/{file_to_download}" # Получаем URL файла
tmp_file = os.path.join(tmp_path, file_to_download)
if not os.path.exists(tmp_file):
try:
try_download(url, tmp_path)
unpack(tmp_file, dist_path)
try_remove_file(tmp_file)
except Exception as e:
log.error(f"Ошибка при загрузке или распаковке: {str(e)}")