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