mirror of
https://github.com/python/cpython.git
synced 2025-11-04 11:49:12 +00:00
ce5fe83878
We now have one function to prepare multipart POST requests, and we use CRLF, as recommended by the HTTP spec (#10150). Initial patch by John Edmonds.
130 lines
4.5 KiB
Python
130 lines
4.5 KiB
Python
"""Upload HTML documentation to a project index."""
|
|
|
|
import os
|
|
import base64
|
|
import socket
|
|
import zipfile
|
|
import logging
|
|
import http.client
|
|
import urllib.parse
|
|
from io import BytesIO
|
|
|
|
from packaging import logger
|
|
from packaging.util import (read_pypirc, DEFAULT_REPOSITORY, DEFAULT_REALM,
|
|
encode_multipart)
|
|
from packaging.errors import PackagingFileError
|
|
from packaging.command.cmd import Command
|
|
|
|
|
|
def zip_dir(directory):
|
|
"""Compresses recursively contents of directory into a BytesIO object"""
|
|
destination = BytesIO()
|
|
with zipfile.ZipFile(destination, "w") as zip_file:
|
|
for root, dirs, files in os.walk(directory):
|
|
for name in files:
|
|
full = os.path.join(root, name)
|
|
relative = root[len(directory):].lstrip(os.path.sep)
|
|
dest = os.path.join(relative, name)
|
|
zip_file.write(full, dest)
|
|
return destination
|
|
|
|
|
|
class upload_docs(Command):
|
|
|
|
description = "upload HTML documentation to PyPI"
|
|
|
|
user_options = [
|
|
('repository=', 'r',
|
|
"repository URL [default: %s]" % DEFAULT_REPOSITORY),
|
|
('show-response', None,
|
|
"display full response text from server"),
|
|
('upload-dir=', None,
|
|
"directory to upload"),
|
|
]
|
|
|
|
def initialize_options(self):
|
|
self.repository = None
|
|
self.realm = None
|
|
self.show_response = False
|
|
self.upload_dir = None
|
|
self.username = ''
|
|
self.password = ''
|
|
|
|
def finalize_options(self):
|
|
if self.repository is None:
|
|
self.repository = DEFAULT_REPOSITORY
|
|
if self.realm is None:
|
|
self.realm = DEFAULT_REALM
|
|
if self.upload_dir is None:
|
|
build = self.get_finalized_command('build')
|
|
self.upload_dir = os.path.join(build.build_base, "docs")
|
|
if not os.path.isdir(self.upload_dir):
|
|
self.upload_dir = os.path.join(build.build_base, "doc")
|
|
logger.info('Using upload directory %s', self.upload_dir)
|
|
self.verify_upload_dir(self.upload_dir)
|
|
config = read_pypirc(self.repository, self.realm)
|
|
if config != {}:
|
|
self.username = config['username']
|
|
self.password = config['password']
|
|
self.repository = config['repository']
|
|
self.realm = config['realm']
|
|
|
|
def verify_upload_dir(self, upload_dir):
|
|
self.ensure_dirname('upload_dir')
|
|
index_location = os.path.join(upload_dir, "index.html")
|
|
if not os.path.exists(index_location):
|
|
mesg = "No 'index.html found in docs directory (%s)"
|
|
raise PackagingFileError(mesg % upload_dir)
|
|
|
|
def run(self):
|
|
name = self.distribution.metadata['Name']
|
|
version = self.distribution.metadata['Version']
|
|
zip_file = zip_dir(self.upload_dir)
|
|
|
|
fields = [(':action', 'doc_upload'),
|
|
('name', name), ('version', version)]
|
|
files = [('content', name, zip_file.getvalue())]
|
|
content_type, body = encode_multipart(fields, files)
|
|
|
|
credentials = self.username + ':' + self.password
|
|
auth = b"Basic " + base64.encodebytes(credentials.encode()).strip()
|
|
|
|
logger.info("Submitting documentation to %s", self.repository)
|
|
|
|
scheme, netloc, url, params, query, fragments = urllib.parse.urlparse(
|
|
self.repository)
|
|
if scheme == "http":
|
|
conn = http.client.HTTPConnection(netloc)
|
|
elif scheme == "https":
|
|
conn = http.client.HTTPSConnection(netloc)
|
|
else:
|
|
raise AssertionError("unsupported scheme %r" % scheme)
|
|
|
|
try:
|
|
conn.connect()
|
|
conn.putrequest("POST", url)
|
|
conn.putheader('Content-type', content_type)
|
|
conn.putheader('Content-length', str(len(body)))
|
|
conn.putheader('Authorization', auth)
|
|
conn.endheaders()
|
|
conn.send(body)
|
|
|
|
except socket.error as e:
|
|
logger.error(e)
|
|
return
|
|
|
|
r = conn.getresponse()
|
|
|
|
if r.status == 200:
|
|
logger.info('Server response (%s): %s', r.status, r.reason)
|
|
elif r.status == 301:
|
|
location = r.getheader('Location')
|
|
if location is None:
|
|
location = 'http://packages.python.org/%s/' % name
|
|
logger.info('Upload successful. Visit %s', location)
|
|
else:
|
|
logger.error('Upload failed (%s): %s', r.status, r.reason)
|
|
|
|
if self.show_response and logger.isEnabledFor(logging.INFO):
|
|
sep = '-' * 75
|
|
logger.info('%s\n%s\n%s', sep, r.read().decode('utf-8'), sep)
|