使用來自Python腳本的POST發送文件

Question

有沒有一種方法可以使用來自Python腳本的POST發送文件？

Answer 1

是的。您將使用urllib2模塊，並使用multipart/form-data內容類型進行編碼。下面是一些示例代碼，讓你開始 - 這是多一點不僅僅是文件上傳，但你應該能夠通過它來閱讀，看看它是如何工作的：

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!') 

Answer 2

你也可以看看在httplib2，與examples。我發現使用httplib2比使用內置的HTTP模塊更簡潔。

Answer 3

Chris Atlee的poster庫對此非常有效（特別是便利功能poster.encode.multipart_encode()）。作爲獎勵，它支持大文件的流式傳輸，無需將整個文件加載到內存中。另見Python issue 3244。

Answer 4

阻止您直接在文件對象上使用urlopen的唯一原因是內置文件對象缺少定義。一個簡單的方法是創建一個提供正確文件的urlopen的子類。 我也在下面的文件中修改了Content-Type標題。

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 

Answer 5

從http://docs.python-requests.org/en/latest/user/quickstart/#post-a-multipart-encoded-file

的要求使得它非常簡單的上傳多部分編碼的文件：

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

就是這樣。我不是在開玩笑 - 這是一行代碼。文件已發送。讓我們來看看：

>>> 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": "" 
} 

Answer 6

看起來像python請求不處理非常大的多部分文件。

該文檔建議您查看requests-toolbelt。

Here's the pertinent page從他們的文檔。

Answer 7

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() 

Answer 8

我想測試Django的REST API和它的工作對我來說：

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)