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