venv/lib/python3.11/site-packages/setuptools/command/upload_docs.py

   1 """upload_docs
   2
   3 Implements a Distutils 'upload_docs' subcommand (upload documentation to
   4 sites other than PyPi such as devpi).
   5 """
   6
   7 from base64 import standard_b64encode
   8 from distutils import log
   9 from distutils.errors import DistutilsOptionError
  10 import os
  11 import socket
  12 import zipfile
  13 import tempfile
  14 import shutil
  15 import itertools
  16 import functools
  17 import http.client
  18 import urllib.parse
  19
  20 from .._importlib import metadata
  21 from ..warnings import SetuptoolsDeprecationWarning
  22
  23 from .upload import upload
  24
  25
  26 def _encode(s):
  27     return s.encode('utf-8', 'surrogateescape')
  28
  29
  30 class upload_docs(upload):
  31     # override the default repository as upload_docs isn't
  32     # supported by Warehouse (and won't be).
  33     DEFAULT_REPOSITORY = 'https://pypi.python.org/pypi/'
  34
  35     description = 'Upload documentation to sites other than PyPi such as devpi'
  36
  37     user_options = [
  38         (
  39             'repository=',
  40             'r',
  41             "url of repository [default: %s]" % upload.DEFAULT_REPOSITORY,
  42         ),
  43         ('show-response', None, 'display full response text from server'),
  44         ('upload-dir=', None, 'directory to upload'),
  45     ]
  46     boolean_options = upload.boolean_options
  47
  48     def has_sphinx(self):
  49         return bool(
  50             self.upload_dir is None
  51             and metadata.entry_points(group='distutils.commands', name='build_sphinx')
  52         )
  53
  54     sub_commands = [('build_sphinx', has_sphinx)]
  55
  56     def initialize_options(self):
  57         upload.initialize_options(self)
  58         self.upload_dir = None
  59         self.target_dir = None
  60
  61     def finalize_options(self):
  62         log.warn(
  63             "Upload_docs command is deprecated. Use Read the Docs "
  64             "(https://readthedocs.org) instead."
  65         )
  66         upload.finalize_options(self)
  67         if self.upload_dir is None:
  68             if self.has_sphinx():
  69                 build_sphinx = self.get_finalized_command('build_sphinx')
  70                 self.target_dir = dict(build_sphinx.builder_target_dirs)['html']
  71             else:
  72                 build = self.get_finalized_command('build')
  73                 self.target_dir = os.path.join(build.build_base, 'docs')
  74         else:
  75             self.ensure_dirname('upload_dir')
  76             self.target_dir = self.upload_dir
  77         self.announce('Using upload directory %s' % self.target_dir)
  78
  79     def create_zipfile(self, filename):
  80         zip_file = zipfile.ZipFile(filename, "w")
  81         try:
  82             self.mkpath(self.target_dir)  # just in case
  83             for root, dirs, files in os.walk(self.target_dir):
  84                 if root == self.target_dir and not files:
  85                     tmpl = "no files found in upload directory '%s'"
  86                     raise DistutilsOptionError(tmpl % self.target_dir)
  87                 for name in files:
  88                     full = os.path.join(root, name)
  89                     relative = root[len(self.target_dir) :].lstrip(os.path.sep)
  90                     dest = os.path.join(relative, name)
  91                     zip_file.write(full, dest)
  92         finally:
  93             zip_file.close()
  94
  95     def run(self):
  96         SetuptoolsDeprecationWarning.emit(
  97             "Deprecated command",
  98             """
  99             upload_docs is deprecated and will be removed in a future version.
 100             Instead, use tools like devpi and Read the Docs; or lower level tools like
 101             httpie and curl to interact directly with your hosting service API.
 102             """,
 103             due_date=(2023, 9, 26),  # warning introduced in 27 Jul 2022
 104         )
 105
 106         # Run sub commands
 107         for cmd_name in self.get_sub_commands():
 108             self.run_command(cmd_name)
 109
 110         tmp_dir = tempfile.mkdtemp()
 111         name = self.distribution.metadata.get_name()
 112         zip_file = os.path.join(tmp_dir, "%s.zip" % name)
 113         try:
 114             self.create_zipfile(zip_file)
 115             self.upload_file(zip_file)
 116         finally:
 117             shutil.rmtree(tmp_dir)
 118
 119     @staticmethod
 120     def _build_part(item, sep_boundary):
 121         key, values = item
 122         title = '\nContent-Disposition: form-data; name="%s"' % key
 123         # handle multiple entries for the same name
 124         if not isinstance(values, list):
 125             values = [values]
 126         for value in values:
 127             if isinstance(value, tuple):
 128                 title += '; filename="%s"' % value[0]
 129                 value = value[1]
 130             else:
 131                 value = _encode(value)
 132             yield sep_boundary
 133             yield _encode(title)
 134             yield b"\n\n"
 135             yield value
 136             if value and value[-1:] == b'\r':
 137                 yield b'\n'  # write an extra newline (lurve Macs)
 138
 139     @classmethod
 140     def _build_multipart(cls, data):
 141         """
 142         Build up the MIME payload for the POST data
 143         """
 144         boundary = '--------------GHSKFJDLGDS7543FJKLFHRE75642756743254'
 145         sep_boundary = b'\n--' + boundary.encode('ascii')
 146         end_boundary = sep_boundary + b'--'
 147         end_items = (
 148             end_boundary,
 149             b"\n",
 150         )
 151         builder = functools.partial(
 152             cls._build_part,
 153             sep_boundary=sep_boundary,
 154         )
 155         part_groups = map(builder, data.items())
 156         parts = itertools.chain.from_iterable(part_groups)
 157         body_items = itertools.chain(parts, end_items)
 158         content_type = 'multipart/form-data; boundary=%s' % boundary
 159         return b''.join(body_items), content_type
 160
 161     def upload_file(self, filename):
 162         with open(filename, 'rb') as f:
 163             content = f.read()
 164         meta = self.distribution.metadata
 165         data = {
 166             ':action': 'doc_upload',
 167             'name': meta.get_name(),
 168             'content': (os.path.basename(filename), content),
 169         }
 170         # set up the authentication
 171         credentials = _encode(self.username + ':' + self.password)
 172         credentials = standard_b64encode(credentials).decode('ascii')
 173         auth = "Basic " + credentials
 174
 175         body, ct = self._build_multipart(data)
 176
 177         msg = "Submitting documentation to %s" % (self.repository)
 178         self.announce(msg, log.INFO)
 179
 180         # build the Request
 181         # We can't use urllib2 since we need to send the Basic
 182         # auth right with the first request
 183         schema, netloc, url, params, query, fragments = urllib.parse.urlparse(
 184             self.repository
 185         )
 186         assert not params and not query and not fragments
 187         if schema == 'http':
 188             conn = http.client.HTTPConnection(netloc)
 189         elif schema == 'https':
 190             conn = http.client.HTTPSConnection(netloc)
 191         else:
 192             raise AssertionError("unsupported schema " + schema)
 193
 194         data = ''
 195         try:
 196             conn.connect()
 197             conn.putrequest("POST", url)
 198             content_type = ct
 199             conn.putheader('Content-type', content_type)
 200             conn.putheader('Content-length', str(len(body)))
 201             conn.putheader('Authorization', auth)
 202             conn.endheaders()
 203             conn.send(body)
 204         except socket.error as e:
 205             self.announce(str(e), log.ERROR)
 206             return
 207
 208         r = conn.getresponse()
 209         if r.status == 200:
 210             msg = 'Server response (%s): %s' % (r.status, r.reason)
 211             self.announce(msg, log.INFO)
 212         elif r.status == 301:
 213             location = r.getheader('Location')
 214             if location is None:
 215                 location = 'https://pythonhosted.org/%s/' % meta.get_name()
 216             msg = 'Upload successful. Visit %s' % location
 217             self.announce(msg, log.INFO)
 218         else:
 219             msg = 'Upload failed (%s): %s' % (r.status, r.reason)
 220             self.announce(msg, log.ERROR)
 221         if self.show_response:
 222             print('-' * 75, r.read(), '-' * 75)