111 lines
5.5 KiB
Python
111 lines
5.5 KiB
Python
"""Check/import a local collection. Rejected games never get queued.
|
|
|
|
python -m app.import_library /games --check
|
|
python -m app.import_library /games/Some-Game --check --single
|
|
python -m app.import_library /games --import --owner admin
|
|
"""
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from app.converter.compatibility import CompatibilityError, Issue
|
|
from app.converter.preflight import inspect_archive, inspect_directory
|
|
|
|
ARCHIVES = (".zip", ".7z", ".rar", ".tar", ".tar.gz", ".tgz", ".tar.bz2", ".tar.xz")
|
|
|
|
|
|
def entries(root: Path):
|
|
"""An unpacked copy takes precedence over an archive with the identical name."""
|
|
directories = {p.name.casefold() for p in root.iterdir() if p.is_dir() and not p.name.startswith(".")}
|
|
for source in sorted(root.iterdir(), key=lambda p: p.name.casefold()):
|
|
if source.name.startswith(".") or source.is_symlink():
|
|
continue
|
|
suffix = next((ext for ext in ARCHIVES if source.name.lower().endswith(ext)), None)
|
|
if source.is_file() and suffix and source.name[:-len(suffix)].casefold() in directories:
|
|
continue
|
|
yield source
|
|
|
|
|
|
def check(source: Path) -> dict:
|
|
try:
|
|
if source.is_dir():
|
|
return inspect_directory(source)
|
|
if not source.name.lower().endswith(ARCHIVES):
|
|
if source.suffix.lower() == ".exe" and source.stat().st_size >= 4:
|
|
with source.open("rb") as handle:
|
|
handle.seek(-4, 2)
|
|
if handle.read(4) == b"GDPC":
|
|
raise CompatibilityError("godot", [Issue("unsupported_engine",
|
|
"В EXE обнаружен встроенный пакет Godot; его исполнение пока не поддерживается")])
|
|
raise CompatibilityError("unknown", [Issue("not_distribution",
|
|
"Нужна папка или архив полного дистрибутива; отдельные EXE и RPA не импортируются")])
|
|
return inspect_archive(source, source.name)
|
|
except CompatibilityError as exc:
|
|
return exc.report()
|
|
except Exception as exc:
|
|
return {"compatible": False, "engine": "unknown", "error": str(exc),
|
|
"issues": [{"code": "input_error", "message": str(exc), "location": source.name, "count": 1}]}
|
|
|
|
|
|
async def import_one(source: Path, report: dict, username: str | None) -> dict:
|
|
from sqlalchemy import select
|
|
|
|
from app.db import SessionLocal
|
|
from app.models import Game, GameVersion, Role, User
|
|
from app.worker import convert
|
|
|
|
async with SessionLocal() as db:
|
|
stmt = select(User).join(Role).order_by(User.id)
|
|
stmt = stmt.where(User.username == username) if username else stmt.where(Role.name == "admin")
|
|
owner = await db.scalar(stmt.limit(1))
|
|
if owner is None:
|
|
raise ValueError("Пользователь для импорта не найден")
|
|
game = Game(title=source.name, engine=report["engine"], created_by_id=owner.id)
|
|
db.add(game)
|
|
await db.flush()
|
|
version = GameVersion(game_id=game.id, owner_id=owner.id, name="Оригинал",
|
|
upload_name=source.name, status="importing", report=report)
|
|
db.add(version)
|
|
await db.commit()
|
|
game_id, version_id = game.id, version.id
|
|
ok = await convert(version_id, source, source.name)
|
|
return {"gameId": game_id, "versionId": version_id, "status": "ready" if ok else "failed"}
|
|
|
|
|
|
async def main(argv=None):
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("root", type=Path)
|
|
parser.add_argument("--single", action="store_true",
|
|
help="Проверить/импортировать одну игру: папку или архив вместо коллекции")
|
|
mode = parser.add_mutually_exclusive_group(required=True)
|
|
mode.add_argument("--check", action="store_true", help="Только проверка; без изменения библиотеки")
|
|
mode.add_argument("--import", dest="import_games", action="store_true", help="Импортировать совместимые игры")
|
|
parser.add_argument("--owner")
|
|
parser.add_argument("--output", type=Path, help="Записать полный JSON-отчёт")
|
|
args = parser.parse_args(argv)
|
|
if args.single and not args.root.exists():
|
|
parser.error("Игра не найдена: " + str(args.root))
|
|
if not args.single and not args.root.is_dir():
|
|
parser.error("root должен быть каталогом коллекции")
|
|
results = []
|
|
for source in ([args.root] if args.single else entries(args.root)):
|
|
print(json.dumps({"game": source.name, "status": "checking"}, ensure_ascii=False), flush=True)
|
|
report = await asyncio.to_thread(check, source)
|
|
result = {**report, "game": source.name}
|
|
if args.import_games and report.get("compatible"):
|
|
result.update(await import_one(source, report, args.owner))
|
|
results.append(result)
|
|
print(json.dumps(result, ensure_ascii=False), flush=True)
|
|
if args.output:
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
if any(not result.get("compatible") or result.get("status") == "failed" for result in results):
|
|
raise SystemExit(2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
asyncio.run(main())
|