343 lines
13 KiB
Python
343 lines
13 KiB
Python
# SPDX-License-Identifier: BSD-2-Clause
|
|
# SPDX-FileCopyrightText: 2021 Volker Krause <vkrause@kde.org>
|
|
# SPDX-FileCopyrightText: 2023 KDE e.V.
|
|
# SPDX-FileContributor: Ingo Klöcker <dev@ingo-kloecker.de>
|
|
|
|
import dataclasses
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import zipfile
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from typing import List, Optional
|
|
|
|
from sftpnotary import apkutils, config, projects, util
|
|
from sftpnotary.aabsigner import SignAabBaseProcessor
|
|
from sftpnotary.core import Job, Request, Response, Worker
|
|
from sftpnotary.exceptions import Error, InvalidFileName
|
|
from sftpnotary.sftp import SFTPClient, SFTPError
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class PublishOnGooglePlayRequest(Request):
|
|
fileNames: Optional[List[str]] = None # can be *.aab or *.apk files
|
|
fastlaneFileName: Optional[str] = None
|
|
|
|
def validate(self):
|
|
if not self.fileNames:
|
|
raise InvalidFileName()
|
|
for fileName in self.fileNames:
|
|
util.validateFileName(fileName, strict=True)
|
|
if not fileName.endswith(".aab") and not fileName.endswith(".apk"):
|
|
raise InvalidFileName(fileName)
|
|
haveAAB = any(f for f in self.fileNames if f.endswith(".aab"))
|
|
if haveAAB and len(self.fileNames) > 1:
|
|
raise Error("Invalid request: Too many .aab files or .aab and .apk files given")
|
|
util.validateFileName(self.fastlaneFileName, strict=True)
|
|
|
|
|
|
PublishOnGooglePlayResponse = Response
|
|
|
|
|
|
class PublishOnGooglePlayJob(Job):
|
|
def __init__(
|
|
self,
|
|
sftp: SFTPClient,
|
|
filePaths: List[Path],
|
|
fastlaneFilePath: Path,
|
|
token: Optional[str] = None,
|
|
projectPath: Optional[str] = None,
|
|
branch: Optional[str] = None,
|
|
):
|
|
super().__init__(sftp)
|
|
|
|
self.filePaths = filePaths
|
|
self.fastlaneFilePath = fastlaneFilePath
|
|
self.request = PublishOnGooglePlayRequest(
|
|
fileNames=[],
|
|
fastlaneFileName=None,
|
|
token=token,
|
|
projectPath=projectPath,
|
|
branch=branch,
|
|
)
|
|
self.resultPath = self.filePaths[0].parent
|
|
|
|
def start(self):
|
|
for filePath in self.filePaths:
|
|
fileName = filePath.name
|
|
self.sftp.upload(filePath, self.task.path() / fileName)
|
|
self.request.fileNames.append(fileName)
|
|
fastlaneFileName = self.fastlaneFilePath.name
|
|
self.sftp.upload(self.fastlaneFilePath, self.task.path() / fastlaneFileName)
|
|
self.request.fastlaneFileName = fastlaneFileName
|
|
self.task.putRequest(self.request)
|
|
self.taskQueue.commitTask(self.task)
|
|
|
|
def waitForCompletion(self):
|
|
self.taskQueue.waitForCompletion(self.task)
|
|
log.debug(f"Fetching response of task '{self.task.id}' ...")
|
|
response = None
|
|
try:
|
|
response = self.task.fetchResponse(PublishOnGooglePlayResponse)
|
|
except SFTPError as e:
|
|
log.error("Error: %s", e)
|
|
if response is None:
|
|
log.error("Error: Got no response. Task failed.")
|
|
else:
|
|
self.success = True
|
|
self.downloadLogFiles(self.resultPath)
|
|
self.taskQueue.removeTask(self.task)
|
|
|
|
|
|
GEM_FILE_TEMPLATE = """
|
|
source "https://rubygems.org"
|
|
gem "fastlane"
|
|
"""
|
|
|
|
APP_FILE_TEMPLATE = """
|
|
json_key_file("{credentialsFilePath}")
|
|
package_name("{applicationId}")
|
|
"""
|
|
|
|
FAST_FILE_TEMPLATE = """
|
|
lane :upload_metadata do
|
|
google_version_code = google_play_track_version_codes(track: 'beta')[0]
|
|
supply(
|
|
package_name: "{applicationId}",
|
|
track: "beta",
|
|
skip_upload_apk: true,
|
|
skip_upload_aab: true,
|
|
skip_upload_changelogs: true,
|
|
skip_upload_metadata: false,
|
|
skip_upload_images: {skipImages},
|
|
skip_upload_screenshots: {skipImages},
|
|
version_code: google_version_code
|
|
)
|
|
end
|
|
"""
|
|
|
|
|
|
class PublishOnGooglePlayProcessor(SignAabBaseProcessor):
|
|
requestClass = PublishOnGooglePlayRequest
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
self.fastlaneUpdatePath = config.settings.getPath("GooglePlayPublishing", "FastlaneDirectory")
|
|
self.fastlaneWorkPath = config.settings.getPath("GooglePlayPublishing", "FastlaneWorkingDirectory")
|
|
self.credentialsFilePath = config.settings.getPath("GooglePlayPublishing", "Credentials")
|
|
self.bundleExecutable = config.settings.getPath("GooglePlayPublishing", "BundleExecutable")
|
|
self.bundlePath = config.settings.getPath("GooglePlayPublishing", "BundlePath")
|
|
self.bundleAppConfig = config.settings.getPath("GooglePlayPublishing", "BundleAppConfig")
|
|
|
|
self.applicationId = ""
|
|
|
|
def doProcess(self):
|
|
self.updateFastlane()
|
|
|
|
with TemporaryDirectory() as tempPathName:
|
|
tempPath = Path(tempPathName)
|
|
|
|
# download the AAB or APKs
|
|
localFilePaths = []
|
|
for fileName in self.request.fileNames:
|
|
localFilePath = tempPath / fileName
|
|
self.sftp.download(self.task.path() / fileName, localFilePath)
|
|
localFilePaths.append(localFilePath)
|
|
localApkFilePaths = [p for p in localFilePaths if p.suffix == ".apk"]
|
|
localAabFilePath = None if localApkFilePaths else localFilePaths[0]
|
|
# download the fastlane metadata file
|
|
localFastlaneFilePath = tempPath / self.request.fastlaneFileName
|
|
self.sftp.download(self.task.path() / self.request.fastlaneFileName, localFastlaneFilePath)
|
|
|
|
allowedApplicationId = projects.settings.get(self.request.projectPath, self.request.branch, "applicationid")
|
|
if localAabFilePath:
|
|
self.verifyApplicationId([localAabFilePath], allowedApplicationId)
|
|
self.applicationId = allowedApplicationId
|
|
signedAabFilePath = self.signAab(localAabFilePath)
|
|
self.publishAab(signedAabFilePath, localFastlaneFilePath)
|
|
else:
|
|
self.verifyApplicationId(localApkFilePaths, allowedApplicationId)
|
|
self.applicationId = allowedApplicationId
|
|
self.publishApks(localApkFilePaths, localFastlaneFilePath)
|
|
|
|
def updateFastlane(self):
|
|
command = [self.bundleExecutable, "update", "fastlane"]
|
|
workPath = self.fastlaneUpdatePath
|
|
environment = {
|
|
**os.environ,
|
|
"BUNDLE_EXECUTABLE": self.bundleExecutable,
|
|
"BUNDLE_PATH": self.bundlePath,
|
|
"BUNDLE_APP_CONFIG": self.bundleAppConfig,
|
|
}
|
|
try:
|
|
util.runCommand(command, cwd=workPath, env=environment)
|
|
except Error:
|
|
# continue the publishing process even if the update of Fastlane failed
|
|
pass
|
|
|
|
def configureFastlane(self, fastlaneFilePath):
|
|
workPath = self.fastlaneWorkPath / self.applicationId
|
|
os.makedirs(workPath / "fastlane", exist_ok=True)
|
|
|
|
# write Gemfile
|
|
with open(workPath / "Gemfile", "w") as f:
|
|
f.write(GEM_FILE_TEMPLATE)
|
|
# remove Gemfile.lock, otherwise the shared updates won't be visible in this remains
|
|
# pinned to the first fastlane version it ever saw
|
|
lockfile = workPath / "Gemfile.lock"
|
|
lockfile.unlink(missing_ok=True)
|
|
|
|
# write Appfile
|
|
with open(workPath / "fastlane" / "Appfile", "w") as f:
|
|
f.write(
|
|
APP_FILE_TEMPLATE.format(credentialsFilePath=self.credentialsFilePath, applicationId=self.applicationId)
|
|
)
|
|
|
|
self.updateMetaData(fastlaneFilePath)
|
|
|
|
def publishAab(self, aabFilePath, fastlaneFilePath):
|
|
self.configureFastlane(fastlaneFilePath)
|
|
|
|
log.info("Uploading AAB")
|
|
self.runFastlaneCommand(
|
|
[
|
|
"supply",
|
|
"--aab",
|
|
aabFilePath,
|
|
"--track",
|
|
"beta",
|
|
"--release_status",
|
|
"draft",
|
|
"--skip_upload_images",
|
|
"true",
|
|
"--skip_upload_screenshots",
|
|
"true",
|
|
]
|
|
)
|
|
|
|
def publishApks(self, apkFilePaths, fastlaneFilePath):
|
|
self.configureFastlane(fastlaneFilePath)
|
|
|
|
log.info("Uploading APKs")
|
|
self.runFastlaneCommand(
|
|
[
|
|
"supply",
|
|
"--apk_paths",
|
|
",".join(str(p) for p in apkFilePaths),
|
|
"--track",
|
|
"beta",
|
|
"--release_status",
|
|
"draft",
|
|
"--skip_upload_images",
|
|
"true",
|
|
"--skip_upload_screenshots",
|
|
"true",
|
|
]
|
|
)
|
|
|
|
def updateMetaData(self, fastlaneFilePath):
|
|
workPath = self.fastlaneWorkPath / self.applicationId
|
|
|
|
# Remove meta data from previous run; otherwise, nothing is downloaded
|
|
# Moreover, we get rid of translations with invalid locale IDs that were
|
|
# not yet filtered out in the previous run
|
|
metaDataPath = workPath / "fastlane" / "metadata" / "android"
|
|
if metaDataPath.exists():
|
|
shutil.rmtree(metaDataPath)
|
|
|
|
#
|
|
# Download existing meta data for comparison
|
|
#
|
|
log.info("Downloading existing metadata")
|
|
self.runFastlaneCommand(["supply", "init"])
|
|
# Running "fastlane supply init" fails if there is no release in the production track
|
|
# (https://github.com/fastlane/fastlane/issues/21505).
|
|
# As workaround one can create a production release but save it as draft.
|
|
|
|
#
|
|
# check whether there are metadata updates
|
|
#
|
|
textChanged = False
|
|
imageChanged = False
|
|
with zipfile.ZipFile(fastlaneFilePath, "r") as zipFile:
|
|
fileNames = apkutils.filterFastlaneFileNames(zipFile.namelist(), self.applicationId)
|
|
for fileName in fileNames:
|
|
# directories or top-level elements
|
|
if "/" not in fileName or fileName.endswith("/"):
|
|
continue
|
|
|
|
# determine the file name in the fastlane structure (without trailing <applicationId> folder)
|
|
_, relativeFileName = fileName.split("/", maxsplit=1)
|
|
outFileName = workPath / "fastlane" / "metadata" / "android" / relativeFileName
|
|
# adjust file names for screenshots
|
|
outFileName = Path(
|
|
re.sub(
|
|
r"/android/([^/]*)/images/([^/]*)/(\d)-.*\.png",
|
|
r"/android/\1/images/\2/\3_\1.png",
|
|
str(outFileName),
|
|
)
|
|
)
|
|
|
|
with zipFile.open(fileName) as inFile:
|
|
inData = inFile.read()
|
|
|
|
outData = b""
|
|
if outFileName.exists():
|
|
with open(outFileName, "rb") as outFile:
|
|
outData = outFile.read()
|
|
|
|
if inData == outData:
|
|
log.debug(f"comparing '{fileName}' and '{outFileName}'... no change")
|
|
continue
|
|
|
|
log.debug(f"comparing '{fileName}' and '{outFileName}'... needs updating")
|
|
outFileName.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(outFileName, "wb") as outFile:
|
|
outFile.write(inData)
|
|
|
|
# remember what we need to update
|
|
if outFileName.suffix == ".png":
|
|
imageChanged = True
|
|
else:
|
|
textChanged = True
|
|
|
|
#
|
|
# Upload metadata changes if necessary
|
|
#
|
|
if not textChanged and not imageChanged:
|
|
log.info("Skipping upload of metadata")
|
|
log.debug("No metadata changes found.")
|
|
else:
|
|
log.info("Uploading metadata")
|
|
log.debug(f"Needs image asset upload: {imageChanged}")
|
|
# write Fastfile
|
|
with open(workPath / "fastlane" / "Fastfile", "w") as f:
|
|
f.write(
|
|
FAST_FILE_TEMPLATE.format(
|
|
applicationId=self.applicationId, skipImages=("false" if imageChanged else "true")
|
|
)
|
|
)
|
|
# submit updates
|
|
self.runFastlaneCommand(["upload_metadata"])
|
|
# Running "fastlane upload_metadata" fails if there is no release in the beta track.
|
|
# As workaround one can create a beta release but save it as draft.
|
|
|
|
def runFastlaneCommand(self, command):
|
|
command = [self.bundleExecutable, "exec", "fastlane"] + command
|
|
workPath = self.fastlaneWorkPath / self.applicationId
|
|
environment = {
|
|
**os.environ,
|
|
"BUNDLE_EXECUTABLE": self.bundleExecutable,
|
|
"BUNDLE_PATH": self.bundlePath,
|
|
"BUNDLE_APP_CONFIG": self.bundleAppConfig,
|
|
}
|
|
util.runCommand(command, cwd=workPath, env=environment)
|
|
|
|
|
|
class PublishOnGooglePlayWorker(Worker):
|
|
processorClass = PublishOnGooglePlayProcessor
|