web-development-kb-asia.site

Gửi tệp bằng cách sử dụng POST từ tập lệnh Python

Có cách nào để gửi tệp bằng cách sử dụng POST từ tập lệnh Python không?

124
Readonly

Từ: http://docs.python-requests.org/en/latest/user/quickstart/#post-a-multipart-encoding-file

Yêu cầu làm cho nó rất đơn giản để tải lên các tệp được mã hóa nhiều phần:

with open('report.xls', 'rb') as f:
    r = requests.post('http://httpbin.org/post', files={'report.xls': f})

Đó là nó. Tôi không nói đùa - đây là một dòng mã. Các tập tin đã được gửi. Hãy kiểm tra:

>>> r.text
{
  "Origin": "179.13.100.4",
  "files": {
    "report.xls": "<censored...binary...data>"
  },
  "form": {},
  "url": "http://httpbin.org/post",
  "args": {},
  "headers": {
    "Content-Length": "3196",
    "Accept-Encoding": "identity, deflate, compress, gzip",
    "Accept": "*/*",
    "User-Agent": "python-requests/0.8.0",
    "Host": "httpbin.org:80",
    "Content-Type": "multipart/form-data; boundary=127.0.0.1.502.21746.1321131593.786.1"
  },
  "data": ""
}
190
Piotr Dobrogost

Đúng. Bạn sẽ sử dụng mô đun urllib2 và mã hóa bằng loại nội dung multipart/form-data. Đây là một số mã mẫu để giúp bạn bắt đầu - nó hơi nhiều hơn là chỉ tải lên tệp, nhưng bạn sẽ có thể đọc qua nó và xem cách nó hoạt động:

user_agent = "image uploader"
default_message = "Image $current of $total"

import logging
import os
from os.path import abspath, isabs, isdir, isfile, join
import random
import string
import sys
import mimetypes
import urllib2
import httplib
import time
import re

def random_string (length):
    return ''.join (random.choice (string.letters) for ii in range (length + 1))

def encode_multipart_data (data, files):
    boundary = random_string (30)

    def get_content_type (filename):
        return mimetypes.guess_type (filename)[0] or 'application/octet-stream'

    def encode_field (field_name):
        return ('--' + boundary,
                'Content-Disposition: form-data; name="%s"' % field_name,
                '', str (data [field_name]))

    def encode_file (field_name):
        filename = files [field_name]
        return ('--' + boundary,
                'Content-Disposition: form-data; name="%s"; filename="%s"' % (field_name, filename),
                'Content-Type: %s' % get_content_type(filename),
                '', open (filename, 'rb').read ())

    lines = []
    for name in data:
        lines.extend (encode_field (name))
    for name in files:
        lines.extend (encode_file (name))
    lines.extend (('--%s--' % boundary, ''))
    body = '\r\n'.join (lines)

    headers = {'content-type': 'multipart/form-data; boundary=' + boundary,
               'content-length': str (len (body))}

    return body, headers

def send_post (url, data, files):
    req = urllib2.Request (url)
    connection = httplib.HTTPConnection (req.get_Host ())
    connection.request ('POST', req.get_selector (),
                        *encode_multipart_data (data, files))
    response = connection.getresponse ()
    logging.debug ('response = %s', response.read ())
    logging.debug ('Code: %s %s', response.status, response.reason)

def make_upload_file (server, thread, delay = 15, message = None,
                      username = None, email = None, password = None):

    delay = max (int (delay or '0'), 15)

    def upload_file (path, current, total):
        assert isabs (path)
        assert isfile (path)

        logging.debug ('Uploading %r to %r', path, server)
        message_template = string.Template (message or default_message)

        data = {'MAX_FILE_SIZE': '3145728',
                'sub': '',
                'mode': 'regist',
                'com': message_template.safe_substitute (current = current, total = total),
                'resto': thread,
                'name': username or '',
                'email': email or '',
                'pwd': password or random_string (20),}
        files = {'upfile': path}

        send_post (server, data, files)

        logging.info ('Uploaded %r', path)
        Rand_delay = random.randint (delay, delay + 5)
        logging.debug ('Sleeping for %.2f seconds------------------------------\n\n', Rand_delay)
        time.sleep (Rand_delay)

    return upload_file

def upload_directory (path, upload_file):
    assert isabs (path)
    assert isdir (path)

    matching_filenames = []
    file_matcher = re.compile (r'\.(?:jpe?g|gif|png)$', re.IGNORECASE)

    for dirpath, dirnames, filenames in os.walk (path):
        for name in filenames:
            file_path = join (dirpath, name)
            logging.debug ('Testing file_path %r', file_path)
            if file_matcher.search (file_path):
                matching_filenames.append (file_path)
            else:
                logging.info ('Ignoring non-image file %r', path)

    total_count = len (matching_filenames)
    for index, file_path in enumerate (matching_filenames):
        upload_file (file_path, index + 1, total_count)

def run_upload (options, paths):
    upload_file = make_upload_file (**options)

    for arg in paths:
        path = abspath (arg)
        if isdir (path):
            upload_directory (path, upload_file)
        Elif isfile (path):
            upload_file (path)
        else:
            logging.error ('No such path: %r' % path)

    logging.info ('Done!')
27
John Millikin

Điều duy nhất ngăn bạn sử dụng urlopen trực tiếp trên một đối tượng tệp là thực tế là đối tượng tệp dựng sẵn thiếu một định nghĩa len. Một cách đơn giản là tạo một lớp con, cung cấp urlopen với tệp chính xác. Tôi cũng đã sửa đổi tiêu đề Kiểu nội dung trong tệp bên dưới.

import os
import urllib2
class EnhancedFile(file):
    def __init__(self, *args, **keyws):
        file.__init__(self, *args, **keyws)

    def __len__(self):
        return int(os.fstat(self.fileno())[6])

theFile = EnhancedFile('a.xml', 'r')
theUrl = "http://example.com/abcde"
theHeaders= {'Content-Type': 'text/xml'}

theRequest = urllib2.Request(theUrl, theFile, theHeaders)

response = urllib2.urlopen(theRequest)

theFile.close()


for line in response:
    print line
4
ilmarinen

Có vẻ như python yêu cầu không xử lý các tệp đa phần cực lớn.

Tài liệu khuyên bạn nên xem xét requests-toolbelt.

Đây là trang thích hợp từ tài liệu của họ.

4
rye

Chris Atlee's poster thư viện hoạt động thực sự tốt cho việc này (đặc biệt là chức năng tiện lợi poster.encode.multipart_encode()). Như một phần thưởng, nó hỗ trợ truyền phát các tệp lớn mà không tải toàn bộ tệp vào bộ nhớ. Xem thêm Vấn đề Python 3244 .

2
gotgenes

Tôi đang cố gắng kiểm tra Django phần còn lại api và nó hoạt động với tôi:

def test_upload_file(self):
        filename = "/Users/Ranvijay/tests/test_price_matrix.csv"
        data = {'file': open(filename, 'rb')}
        client = APIClient()
        # client.credentials(HTTP_AUTHORIZATION='Token ' + token.key)
        response = client.post(reverse('price-matrix-csv'), data, format='multipart')

        print response
        self.assertEqual(response.status_code, status.HTTP_200_OK)
2
Ranvijay Sachan

Bạn cũng có thể muốn xem omeplib2 , với ví dụ . Tôi thấy việc sử dụng omeplib2 ngắn gọn hơn so với sử dụng các mô đun HTTP tích hợp.

0
pdc
def visit_v2(device_code, camera_code):
    image1 = MultipartParam.from_file("files", "/home/yuzx/1.txt")
    image2 = MultipartParam.from_file("files", "/home/yuzx/2.txt")
    datagen, headers = multipart_encode([('device_code', device_code), ('position', 3), ('person_data', person_data), image1, image2])
    print "".join(datagen)
    if server_port == 80:
        port_str = ""
    else:
        port_str = ":%s" % (server_port,)
    url_str = "http://" + server_ip + port_str + "/adopen/device/visit_v2"
    headers['nothing'] = 'nothing'
    request = urllib2.Request(url_str, datagen, headers)
    try:
        response = urllib2.urlopen(request)
        resp = response.read()
        print "http_status =", response.code
        result = json.loads(resp)
        print resp
        return result
    except urllib2.HTTPError, e:
        print "http_status =", e.code
        print e.read()
0
user6081103