Page MenuHomePhabricator
Paste P11658

(An Untitled Masterwork)
ActivePublic

Authored by Legoktm on Jun 25 2020, 7:23 AM.
Tags
None
Referenced Files
F31904812: raw.txt
Jun 25 2020, 7:23 AM
Subscribers
None
#!/usr/bin/env python3
"""
Copyright (C) 2020 Kunal Mehta <legoktm@member.fsf.org>
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License along
with this program; if not, write to the Free Software Foundation, Inc.,
51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""
import json
import requests
import os
import subprocess
from pathlib import Path
import shutil
session = requests.Session()
def get_provider(path: Path) -> str:
if (path / 'gitinfo.json').exists():
# TODO: gitinfo should point back to the extdist host
return 'extdist'
elif (path / '.git').exists():
return 'git'
else:
# TODO: add more types?
raise RuntimeError(f"Unsure how {path} was downloaded")
def get_current(path: Path) -> str:
provider = get_provider(path)
if provider == 'extdist':
with open(path / 'gitinfo.json') as f:
data = json.load(f)
return data['headSHA1'].strip()[:7]
else:
return subprocess.check_output(['git', 'rev-list', 'HEAD', '-n1'], cwd=str(path)).decode().strip()
def get_latest(path: Path, mwb: str) -> str:
provider = get_provider(path)
name = str(path.name)
print(f'Getting the latest version of {name}')
if provider == 'extdist':
req = session.get('https://www.mediawiki.org/w/api.php', params={
'action': 'query',
'list': 'extdistbranches',
'edbexts': name,
'format': 'json',
'formatversion': 2,
})
req.raise_for_status()
data = req.json()
# TODO: error handling
# TODO: API should provide raw SHA1
return data['query']['extdistbranches']['extensions'][name][mwb].split('-')[-1].split('.')[0]
else:
# TODO: non-origin remote?
subprocess.check_call(['git', 'fetch', 'origin'], cwd=str(path))
return subprocess.check_output(['git', 'show-ref', f'origin/{mwb}'], cwd=str(path)).decode().split(' ')[0]
def download_extdist(path: Path, mwb):
# TODO make this atomic and safe
name = str(path.name)
apireq = session.get('https://www.mediawiki.org/w/api.php', params={
'action': 'query',
'list': 'extdistbranches',
'edbexts': name,
'format': 'json',
'formatversion': 2,
})
apireq.raise_for_status()
data = apireq.json()
url = data['query']['extdistbranches']['extensions'][name][mwb]
# TODO: automatically restore bak if things go bad
path.rename(str(path) + '.bak')
tarball = path.parent / 'tarball.tar.gz'
print(f'[extdist] Downloading {url}')
with session.get(url, stream=True) as req:
req.raise_for_status()
with open(tarball, 'wb') as f:
for chunk in req.iter_content(chunk_size=8192):
f.write(chunk)
subprocess.check_call(['tar', '-xzf', tarball, '-C', str(tarball.parent)])
shutil.rmtree(str(path) + '.bak')
tarball.unlink()
# TODO: some integrity verification on success
def handle_composer(path: Path) -> bool:
cmp = path / 'composer.json'
if not cmp.exists():
return False
with open(cmp) as f:
info = json.load(f)
if 'require' not in info:
return False
ignore = ['php', 'composer/installers']
if not any(pkg for pkg in info['require'] if (pkg not in ignore and not pkg.startswith('ext-'))):
# no real composer deps
# TODO: consider removing if there are no deps??
return False
# TODO: in theory we could delete the extdist-provided vendor/
local = path.parent.parent / 'composer.local.json'
if local.exists():
with open(local) as f:
content = json.load(f)
else:
content = {'extra': {'merge-plugin': {'include': []}}}
save = False
if cmp not in content['extra']['merge-plugin']['include']:
# TODO: realpath() equivalency
save = True
content['extra']['merge-plugin']['include'].append(str(cmp))
if save:
with open(local, 'w') as f:
json.dump(content, f)
return True
def update(path: Path, mwb, check_only=False):
current = get_current(path)
latest = get_latest(path, mwb)
print(repr(latest), repr(current))
if current == latest:
print(f'{path} is already up to date!')
return False
if check_only:
print(f'{path} needs an update')
return True
provider = get_provider(path)
print(f'[{provider}]: Updating {path}...')
if provider == 'extdist':
download_extdist(path, mwb)
else:
# TODO: non-origin remote?
subprocess.check_call(['git', 'fetch', 'origin'], cwd=str(path))
subprocess.check_call(['git', 'checkout', f'origin/{mwb}'], cwd=str(path))
print(f'[{provider}] Successfully updated {path}')
return False
def run_composer(path: Path):
subprocess.check_call(['composer', 'update', '--no-dev'], cwd=path)
def main():
# Get this list from listing extensions/ or from user input
composer = False
for ext in ['TemplateStyles', 'AbuseFilter']:
path = Path(f'/home/user/projects/mwext-prototype/test/extensions/{ext}')
# TODO: Get MediaWiki branch from DefaultSettings.php (or elsewhere)
mwb = 'REL1_34'
if not update(path, mwb, check_only=True):
continue
update(path, mwb)
composer = handle_composer(path) or composer
if composer:
print('Triggering composer update...')
run_composer(path.parent.parent)
if __name__ == '__main__':
main()