OpenMiInk/third_party/ci-notary-service/sftpnotary/microsoftstoresubmission.py
Hellings b8c8ff98e2 Hi
2026-10-16 09:05:56 -04:00

874 lines
23 KiB
Python

# SPDX-License-Identifier: BSD-2-Clause
# SPDX-FileCopyrightText: 2023 KDE e.V.
# SPDX-FileContributor: Ingo Klöcker <dev@ingo-kloecker.de>
import json
import logging
import os
from pathlib import Path
from tempfile import TemporaryDirectory
from urllib.parse import urlparse
from zipfile import ZipFile
import requests
from azure.storage.blob import BlobClient
from markdownify import markdownify # type: ignore
from microstore.MicrosoftStoreClient import APIError, AuthorizationError, MicrosoftStoreClient # type: ignore
class Error(Exception):
pass
logger = logging.getLogger(__name__)
# Maps AppStream language ID to Microsoft language ID
appstream_to_microsoft_language = {
"en": "en-us", # the apps.kde.org appdata is actually a bit mangled the real key would be C not 'en'
"sr": "sr-cyrl",
"sr-la": "sr-Latn", # preserve capital letter (by default we downcase)
"ca-va": "ca-es-valencia",
"az": "az-arab",
}
# Languages that are supported by Microsoft Store
# (https://docs.microsoft.com/en-us/windows/uwp/publish/supported-languages):
microsoft_languages = [
# Arabic
"ar",
"ar-sa",
"ar-ae",
"ar-bh",
"ar-dz",
"ar-eg",
"ar-iq",
"ar-jo",
"ar-kw",
"ar-lb",
"ar-ly",
"ar-ma",
"ar-om",
"ar-qa",
"ar-sy",
"ar-tn",
"ar-ye",
# Afrikaans
"af",
"af-za",
# Albanian
"sq",
"sq-al",
# Amharic
"am",
"am-et",
# Armenian
"hy",
"hy-am",
# Assamese
"as",
"as-in",
# Azerbaijani
"az-arab",
"az-arab-az",
"az-cyrl",
"az-cyrl-az",
"az-latn",
"az-latn-az",
# Basque (Basque)
"eu",
"eu-es",
# Belarusian
"be",
"be-by",
# Bangla
"bn",
"bn-bd",
"bn-in",
# Bosnian
"bs",
"bs-cyrl",
"bs-cyrl-ba",
"bs-latn",
"bs-latn-ba",
# Bulgarian
"bg",
"bg-bg",
# Catalan
"ca",
"ca-es",
"ca-es-valencia",
# Cherokee
"chr-cher",
"chr-cher-us",
"chr-latn",
# Chinese (Simplified)
"zh-Hans",
"zh-cn",
"zh-hans-cn",
"zh-sg",
"zh-hans-sg",
# Chinese (Traditional)
"zh-Hant",
"zh-hk",
"zh-mo",
"zh-tw",
"zh-hant-hk",
"zh-hant-mo",
"zh-hant-tw",
# Croatian
"hr",
"hr-hr",
"hr-ba",
# Czech
"cs",
"cs-cz",
# Danish
"da",
"da-dk",
# Dari
"prs",
"prs-af",
"prs-arab",
# Dutch
"nl",
"nl-nl",
"nl-be",
# English
"en",
"en-au",
"en-ca",
"en-gb",
"en-ie",
"en-in",
"en-nz",
"en-sg",
"en-us",
"en-za",
"en-bz",
"en-hk",
"en-id",
"en-jm",
"en-kz",
"en-mt",
"en-my",
"en-ph",
"en-pk",
"en-tt",
"en-vn",
"en-zw",
"en-053",
"en-021",
"en-029",
"en-011",
"en-018",
"en-014",
# Estonian
"et",
"et-ee",
# Filipino
"fil",
"fil-latn",
"fil-ph",
# Finnish
"fi",
"fi-fi",
# French
"fr",
"fr-be",
"fr-ca",
"fr-ch",
"fr-fr",
"fr-lu",
"fr-015",
"fr-cd",
"fr-ci",
"fr-cm",
"fr-ht",
"fr-ma",
"fr-mc",
"fr-ml",
"fr-re",
"frc-latn",
"frp-latn",
"fr-155",
"fr-029",
"fr-021",
"fr-011",
# Galician
"gl",
"gl-es",
# Georgian
"ka",
"ka-ge",
# German
"de",
"de-at",
"de-ch",
"de-de",
"de-lu",
"de-li",
# Greek
"el",
"el-gr",
# Gujarati
"gu",
"gu-in",
# Hausa
"ha",
"ha-latn",
"ha-latn-ng",
# Hebrew
"he",
"he-il",
# Hindi
"hi",
"hi-in",
# Hungarian
"hu",
"hu-hu",
# Icelandic
"is",
"is-is",
# Igbo
"ig-latn",
"ig-ng",
# Indonesian
"id",
"id-id",
# Inuktitut (Latin)
"iu-cans",
"iu-latn",
"iu-latn-ca",
# Irish
"ga",
"ga-ie",
# isiXhosa
"xh",
"xh-za",
# isiZulu
"zu",
"zu-za",
# Italian
"it",
"it-it",
"it-ch",
# Japanese
"ja",
"ja-jp",
# Kannada
"kn",
"kn-in",
# Kazakh
"kk",
"kk-kz",
# Khmer
"km",
"km-kh",
# K'iche',
"quc-latn",
"qut-gt",
"qut-latn",
# Kinyarwanda
"rw",
"rw-rw",
# KiSwahili
"sw",
"sw-ke",
# Konkani
"kok",
"kok-in",
# Korean
"ko",
"ko-kr",
# Kurdish
"ku-arab",
"ku-arab-iq",
# Kyrgyz
"ky-kg",
"ky-cyrl",
# Lao
"lo",
"lo-la",
# Latvian
"lv",
"lv-lv",
# Lithuanian
"lt",
"lt-lt",
# Luxembourgish
"lb",
"lb-lu",
# Macedonian
"mk",
"mk-mk",
# Malay
"ms",
"ms-bn",
"ms-my",
# Malayalam
"ml",
"ml-in",
# Maltese
"mt",
"mt-mt",
# Maori
"mi",
"mi-latn",
"mi-nz",
# Marathi
"mr",
"mr-in",
# Mongolian (Cyrillic)
"mn-cyrl",
"mn-mong",
"mn-mn",
"mn-phag",
# Nepali
"ne",
"ne-np",
# Norwegian
"nb",
"nb-no",
"nn",
"nn-no",
"no",
"no-no",
# Odia
"or",
"or-in",
# Persian
"fa",
"fa-ir",
# Polish
"pl",
"pl-pl",
# Portuguese (Brazil)
"pt-br",
# Portuguese (Portugal)
"pt",
"pt-pt",
# Punjabi
"pa",
"pa-arab",
"pa-arab-pk",
"pa-deva",
"pa-in",
# Quechua
"quz",
"quz-bo",
"quz-ec",
"quz-pe",
# Romanian
"ro",
"ro-ro",
# Russian
"ru",
"ru-ru",
# Scottish Gaelic
"gd-gb",
"gd-latn",
# Serbian (Latin)
"sr-Latn",
"sr-latn-cs",
"sr",
"sr-latn-ba",
"sr-latn-me",
"sr-latn-rs",
# Serbian (Cyrillic)
"sr-cyrl",
"sr-cyrl-ba",
"sr-cyrl-cs",
"sr-cyrl-me",
"sr-cyrl-rs",
# Sesotho sa Leboa
"nso",
"nso-za",
# Setswana
"tn",
"tn-bw",
"tn-za",
# Sindhi
"sd-arab",
"sd-arab-pk",
"sd-deva",
# Sinhala
"si",
"si-lk",
# Slovak
"sk",
"sk-sk",
# Slovenian
"sl",
"sl-si",
# Spanish
"es",
"es-cl",
"es-co",
"es-es",
"es-mx",
"es-ar",
"es-bo",
"es-cr",
"es-do",
"es-ec",
"es-gt",
"es-hn",
"es-ni",
"es-pa",
"es-pe",
"es-pr",
"es-py",
"es-sv",
"es-us",
"es-uy",
"es-ve",
"es-019",
"es-419",
# Swedish
"sv",
"sv-se",
"sv-fi",
# Tajik (Cyrillic)
"tg-arab",
"tg-cyrl",
"tg-cyrl-tj",
"tg-latn",
# Tamil
"ta",
"ta-in",
# Tatar
"tt-arab",
"tt-cyrl",
"tt-latn",
"tt-ru",
# Telugu
"te",
"te-in",
# Thai
"th",
"th-th",
# Tigrinya
"ti",
"ti-et",
# Turkish
"tr",
"tr-tr",
# Turkmen
"tk-cyrl",
"tk-latn",
"tk-tm",
"tk-latn-tr",
"tk-cyrl-tr",
# Ukrainian
"uk",
"uk-ua",
# Urdu
"ur",
"ur-pk",
# Uyghur
"ug-arab",
"ug-cn",
"ug-cyrl",
"ug-latn",
# Uzbek (Latin)
"uz",
"uz-cyrl",
"uz-latn",
"uz-latn-uz",
# Vietnamese
"vi",
"vi-vn",
# Welsh
"cy",
"cy-gb",
# Wolof
"wo",
"wo-sn",
# Yoruba
"yo-latn",
"yo-ng",
]
# Languages supported by KDE that are not supported by Microsoft:
# * ast (Asturian), ia (Interlingua), ie (Interlingue) are not supported by Microsoft
# * some SR variants are not a thing outside the KDE bubble
# * x-test
def getAppStreamData(appstreamId):
r = requests.get(f"https://apps.kde.org/appdata/{appstreamId}.json", timeout=5)
r.raise_for_status()
return r.json()
def createListing(title):
return {
"baseListing": {
"copyrightAndTrademarkInfo": "",
"keywords": [],
"licenseTerms": "",
"privacyPolicy": "",
"supportContact": "",
"websiteUrl": "",
"description": "",
"features": [],
"releaseNotes": "",
"images": [],
"recommendedHardware": [],
"minimumHardware": [],
"title": title,
"shortDescription": "",
"shortTitle": "",
"sortTitle": "",
"voiceTitle": "",
"devStudio": "",
}
}
def downloadFile(url, downloadPath):
if url is None:
return None
localPath = downloadPath / Path(urlparse(url).path).name
if localPath.exists():
logger.debug(f"File {localPath} already exists. Skipping download.")
return localPath
logger.debug(f"Downloading {url}")
r = requests.get(url, timeout=5)
if not r.ok:
logger.warning(f"Failed to download {url}")
return None
logger.debug(f"Writing file to {localPath}")
os.makedirs(downloadPath, exist_ok=True)
with open(localPath, "wb") as f:
f.write(r.content)
return localPath
def reformatDescription(description):
description = description.strip()
if description.startswith("<"):
# convert any linebreaks in the HTML code to spaces (linebreaks are
# meaningless in HTML, but markdownify does not touch them)
description = description.replace("\n", " ")
# convert HTML description to Markdown (Microsoft Store only supports
# plain text, but some Markdown looks nicer)
description = markdownify(description, heading_style="UNDERLINED").strip()
# remove leading (and trailing) whitespace from all lines
description = "\n".join(line.strip() for line in description.split("\n"))
return description
def updateBaseListingForLanguage(lang, storeLang, baseListing, defaultBaseListing, appstreamData, downloadPath, keep):
# update description for listing
storeDescription = baseListing.get("description", "")
if "description" not in keep or not storeDescription:
description = appstreamData.get("Description", {}).get(lang, "")
if description:
baseListing["description"] = reformatDescription(description)
elif storeDescription:
logger.debug(f"No description for {lang}. Keeping existing description in listing for language.")
else:
logger.debug(f"No description for {lang}. Skipping update of listing for language.")
return None
# copyrightAndTrademarkInfo - AppStream does not provide this
# update keywords for listing
if "keywords" not in keep or not baseListing["keywords"]:
baseListing["keywords"] = appstreamData.get("Keywords", {}).get(lang, [])
# update licenseTerms for listing
if "licenseterms" not in keep or not baseListing["licenseTerms"]:
baseListing["licenseTerms"] = appstreamData.get("ProjectLicense", "")
# privacyPolicy - This value is obsolete. The privacy policy URL must be set in Partner Center.
# supportContact - This value is obsolete. The support contact URL or email address must be set in Partner Center.
# websiteUrl - This value is obsolete. The URL of the web page for your app must be set in Partner Center.
# features - AppStream does not provide a list of features (https://github.com/ximion/appstream/issues/388)
# update releaseNotes for listing
if "releasenotes" not in keep or not baseListing["releaseNotes"]:
releases = appstreamData.get("Releases")
if releases: # not None and not empty
latestRelease = releases[0]
releaseNotes = latestRelease.get("description", {}).get(lang, "") or defaultBaseListing["releaseNotes"]
if not releaseNotes:
latestVersion = latestRelease.get("version")
if latestVersion:
releaseNotes = f"Update to version {latestVersion}"
baseListing["releaseNotes"] = reformatDescription(releaseNotes)
else:
baseListing["releaseNotes"] = ""
# update images for listing
if "images" not in keep or not baseListing["images"]:
newScreenshots = []
# prefer Windows-specific screenshots with environment="windows"
for screenshot in appstreamData.get("Screenshots", []):
if screenshot.get("environment") != "windows":
continue
url = screenshot.get("source-image", {}).get("url")
imagePath = downloadFile(url, downloadPath / "images")
if imagePath is not None:
description = screenshot.get("caption", {}).get(lang, "") or screenshot.get("caption", {}).get("en", "")
newScreenshots.append(
{
"fileName": "images/" + imagePath.name,
"fileStatus": "PendingUpload",
"description": description,
"imageType": "Screenshot",
}
)
# use standard screenshots if there are no Windows-specific screenshots
if not newScreenshots:
for screenshot in appstreamData.get("Screenshots", []):
if not screenshot.get("default", False):
continue
url = screenshot.get("source-image", {}).get("url")
imagePath = downloadFile(url, downloadPath / "images")
if imagePath is not None:
description = screenshot.get("caption", {}).get(lang, "") or screenshot.get("caption", {}).get(
"en", ""
)
# assume screenshot for mobile if image filename contains 'mobile'
imageType = "MobileScreenshot" if "mobile" in imagePath.name else "Screenshot"
newScreenshots.append(
{
"fileName": "images/" + imagePath.name,
"fileStatus": "PendingUpload",
"description": description,
"imageType": imageType,
}
)
# Microsoft Store supports "up to 9 desktop screenshots [and up to 8 screenshots for the other device families]"
if len(newScreenshots) > 9:
logger.info(
f"Microsoft Store supports up to 9 desktop screenshots. Ignoring {len(newScreenshots) - 9} screenshots."
)
newScreenshots = newScreenshots[:9]
if newScreenshots:
# replace old screenshots with new ones
baseListing["images"] = [
image for image in baseListing["images"] if not image["imageType"].endswith("Screenshot")
]
baseListing["images"].extend(newScreenshots)
# update additional logos and images
imageTypes = (
"StoreLogo9x16",
"StoreLogoSquare",
"Icon",
"PromotionalArt16x9",
)
for imageType in imageTypes:
url = appstreamData.get("Custom", {}).get(f"KDE::windows_store::{imageType}")
if url is None:
continue
imagePath = downloadFile(url, downloadPath / "images")
if imagePath is not None:
# replace old image with new one
baseListing["images"] = [image for image in baseListing["images"] if image["imageType"] != imageType]
baseListing["images"].append(
{
"fileName": "images/" + imagePath.name,
"fileStatus": "PendingUpload",
"imageType": imageType,
}
)
# recommendedHardware, minimumHardware - AppStream does not provide these
# title - title must be a reserved name; keep what's there
# update shortDescription for listing (only used for games)
if "shortdescription" not in keep or not baseListing["shortDescription"]:
baseListing["shortDescription"] = appstreamData.get("Summary", {}).get(lang, "")
# shortTitle, sortTitle, voiceTitle - AppStream does not provide these
# update devStudio for listing
if "devstudio" not in keep or not baseListing["devStudio"]:
baseListing["devStudio"] = (
appstreamData.get("DeveloperName", {}).get(lang, "") or defaultBaseListing["devStudio"]
)
logger.debug(f"Listing for {storeLang}: %s", json.dumps(baseListing))
return baseListing
def updateSubmissionWithAppStreamData(submissionData, appstreamData, downloadPath, keep, title):
listings = submissionData["listings"]
defaultBaseListing = listings["en-us"]["baseListing"]
if keep:
logger.debug("Keeping the following values of the existing listings: %s", ", ".join(keep))
# first update the (base) listing for en/en-us (the default listing)
updateBaseListingForLanguage(
"en", "en-us", defaultBaseListing, defaultBaseListing, appstreamData, downloadPath, keep
)
for lang in appstreamData["Name"]:
if lang == "en":
continue
storeLang = appstream_to_microsoft_language.get(lang, lang)
if storeLang not in microsoft_languages:
logger.debug(f"Skipping language {lang} that is not supported by the Microsoft Store")
continue
listing = listings.get(storeLang)
if listing is None:
listing = createListing(title)
baseListing = listing["baseListing"]
updatedBaseListing = updateBaseListingForLanguage(
lang, storeLang, baseListing, defaultBaseListing, appstreamData, downloadPath, keep
)
# we keep existing listings even if updateBaseListingForLanguage doesn't like them
if updatedBaseListing is not None:
listings[storeLang] = listing
return submissionData
def uploadPackagesFiles(zipFilePath, fileUploadUrl):
logger.debug(f"Uploading {zipFilePath} to {fileUploadUrl}")
blobClient = BlobClient.from_blob_url(fileUploadUrl)
with open(zipFilePath, "rb") as data:
result = blobClient.upload_blob(data, blob_type="BlockBlob")
logger.debug("Upload result: %s", result)
def _submitApp(client, *, appstreamId, appxuploadFile, storeId, keep, dryRun, skipCommit, **kwargs):
if not os.path.isfile(appxuploadFile):
raise Error(f"File not found: {appxuploadFile}")
if dryRun:
logger.info("Running in dry-run mode")
# fetch AppStream metadata of the app
logger.info(f"Fetching AppStream metadata for {appstreamId}")
appstreamData = getAppStreamData(appstreamId)
if not storeId:
# get the Store ID of the app from the AppStream data
storeUrl = appstreamData.get("Custom", {}).get("KDE::windows_store")
if storeUrl is None:
raise Error("No KDE::windows_store entry found in AppStream data.")
storeId = os.path.basename(urlparse(storeUrl).path)
logger.debug(f"Using Store ID {storeId}")
# get app data
logger.info("Retrieving information about the app from the Microsoft Store")
appData = client.getAppData(storeId)
lastSubmissionId = appData.get("lastPublishedApplicationSubmission", {}).get("id")
if not lastSubmissionId:
raise Error("The first submission needs to be done manually in the Microsoft Partner Center.")
# delete existing pending submission
pendingSubmissionId = appData.get("pendingApplicationSubmission", {}).get("id")
if pendingSubmissionId: # not None and not empty
logger.info("Deleting existing pending submission")
if not dryRun:
client.deleteSubmission(storeId, pendingSubmissionId)
# create a new submission (as copy of the last submission)
logger.info("Creating new submission")
if not dryRun:
submissionData = client.createSubmission(storeId)
else:
submissionData = client.getSubmission(storeId, lastSubmissionId)
submissionData["id"] = int(submissionData["id"]) + 1
submissionData["fileUploadUrl"] = "https://localhost"
submissionId = submissionData["id"]
logger.debug(f"submission ID: {submissionId}")
# remove some entries from the submission data
fileUploadUrl = submissionData["fileUploadUrl"]
logger.debug(f"fileUploadUrl: {fileUploadUrl}")
del submissionData["fileUploadUrl"] # remove fileUploadUrl
# remove deprecated "pricing.sales" entry
if "pricing" in submissionData and "sales" in submissionData["pricing"]:
del submissionData["pricing"]["sales"]
with TemporaryDirectory() as tmpdirname:
tempPath = Path(tmpdirname)
# set data for new submission
submissionData = updateSubmissionWithAppStreamData(
submissionData, appstreamData, tempPath, keep, title=appData["primaryName"]
)
# add information about the app package (if it has not yet been uploaded)
appPackageName = os.path.basename(appxuploadFile)
if any(
p["fileName"] == appPackageName and p["fileStatus"] == "Uploaded"
for p in submissionData["applicationPackages"]
):
# the app package has already been uploaded
logger.info(f"{appPackageName} has already been uploaded")
appPackageName = None
else:
logger.info(f"Adding {appPackageName} to list of application packages")
submissionData["applicationPackages"].append(
{
"fileName": appPackageName, # relative to root of uploaded ZIP file
"fileStatus": "PendingUpload",
}
)
# request manual publication
submissionData["targetPublishMode"] = "Manual"
# update the submission
logger.info("Updating submission")
if not dryRun:
client.updateSubmission(storeId, submissionId, submissionData)
else:
print(f"Updated submission data:\n{json.dumps(submissionData, indent=4)}")
# upload images and packages in a zip file
logger.info("Packaging files to upload")
zipFilePath = tempPath / "blob.zip"
# we use the default ZIP_STORED compression because the files should already be fairly compressed
with ZipFile(zipFilePath, "w") as zipFile:
if "images" not in keep:
for imagePath in (tempPath / "images").iterdir():
logger.debug(f"Adding {imagePath} to {zipFilePath}")
zipFile.write(imagePath, arcname="images/" + imagePath.name)
if appPackageName:
logger.debug(f"Adding {appPackageName} to {zipFilePath}")
zipFile.write(appxuploadFile, arcname=appPackageName)
logger.info("Uploading packaged files")
if not dryRun:
uploadPackagesFiles(zipFilePath, fileUploadUrl)
# commit the submission
if not skipCommit:
logger.info("Committing submission")
if not dryRun:
client.commitSubmission(storeId, submissionId, waitUntilCommitIsCompleted=True)
else:
logger.info("Skipped committing the submission")
def submitApp(appxuploadFile, appstreamId, keep, tenantId, clientId, clientSecret):
client = MicrosoftStoreClient(tenantId, clientId, clientSecret)
try:
_submitApp(
client,
appxuploadFile=appxuploadFile,
appstreamId=appstreamId,
storeId=None,
keep=keep,
dryRun=False,
skipCommit=False,
)
return
except AuthorizationError as e:
message = f"Authentication failed: {e}. Check server logs for details."
except APIError as e:
message = f"API request failed: {e}. Check server logs for details."
raise Error(message)