วิธีที่จะพิมพ์"จุด"(หรืออื่นๆเป็นการแจ้งเตือนการเริ่มทำงาน)ระหว่างทำการเขียนแฟ้มไพธอน?

requests.get ค่าปริยายของการดาวน์โหลดที่ entirety ของที่ร้องของทรัพยากรก่อนที่มันจะกลับมาเพื่อคุณ อย่างไรก็ตามมันต้องมีตัวเลือกข้อโต้แย้ง streamซึ่งอนุญาตให้คุณเรียกร้อง .iter_content หรือ .iter_lines บ Response วัตถุ. นี่จะอนุญาตให้คุณใช้การกระทำทุกเอ็นไบต์(หรือเป็นแต่ละชิ้นส่วนของข้อมูลมาถึง)หรือทุกสายตามลำดับ บางอย่างเหมือนนี้:

chunks = []
chunk_size = 16384     # 16Kb chunks
# alternately
# chunk_size = None    # whenever a chunk arrives
res = requests.get(url_to_file, stream=True)
for chunk in res.iter_content(chunk_size):
    chunks.append(chunk)
    print(".", end="")
data = b''.join(chunks)

มันยังช่วงตึกว่าดังนั้นไม่มีอะไรที่จะเกิดขึ้น ถ้าคุณต้องการมากกว่าของจาวาสคริปต์สไตล์ต่อ Grismar เป็นความคิดเห็นเธอควรวิ่งอยู่ใต้ปลั๊กอินสำหรับไพธอนเป็น async เรื่อยๆ ในกรณีนั้นผมขอแนะนำให้ใช้ aiohttp มากกว่า requestsอย่างที่มันเป็นสร้างขึ้นกับ async รูปแบบในใจแล้ว

นี่คือเวอร์ชั่นที่จะดาวน์โหลดแฟ้มเข้าไป bytearray แยกเป็นเส้นด้าย

ที่พูดถึงในอีกคำตอบและหมายเหตุ,มันยังมีอีกหลาย alternativs นั่นคือพัฒนากับ async ปฏิบัติการอยู่ในใจงั้นไม่ต้องอ่านมากเกินไปในการตัดสินใจที่จะไปกับ threadingมันเป็นเพียงเพื่อแสดงให้เห็นที่สงครามแย่งชิงยุทธศาส(และเพราะมันผิดธรรมเนียมต่อศาตั้งแต่มันมาพร้อมกับเป็นภาษาไพธอน).

ในทางด้านล่างรหัสถ้าขนาดของแฟ้มที่เป็นที่รู้จักแต่ละ . จะธ์กัน 1%ซะอีก เป็นโบนัสที่ดาวน์โหลดข้อมูลและหมายเลขรวมของไบต์จะถูกพิมพ์ออกมาเป็นที่ที่เริ่มต้นของเส้นเหมือน (1234 B / 1234567 B). ถ้าขนาดของคืนที่รู้จักกันที่สำหรับการใช้งานต่อเนื่องแก้ปัญหาก็คือต้องแต่ละ . เป็นตัวแทนของชิ้นส่วน.

import requests
import threading


def download_file(url: str):
    headers = {"<some_key>": "<some_value>"}
    data = bytearray()
    with requests.get(url, headers=headers, stream=True) as request:
        if file_size := request.headers.get("Content-Length"):
            file_size = int(file_size)
        else:
            file_size = None
        received = 0
        for chunk in request.iter_content(chunk_size=2**15):
            received += len(chunk)
            data += chunk
            try:
                num_dots = int(received * 100 / file_size)
                print(
                    f"({received} B/{file_size} B) "
                    + "." * num_dots, end="\r"
                )
            except TypeError:
                print(".", end="")
        print("\nDone!")

url = "<some_url>"
thread = threading.Thread(target=download_file, args=(url,))
thread.start()
# Do something in the meantime
thread.join()

ทำให้อยู่ในจิตใจที่ผมเหลืออกจากล็อคเพื่อปกป้องต่อต้านวนท่อนที่เข้าถึง stdout เพื่อลดเสียงดัง ฉันยังเหลือการเขียน bytarray ต้องแฟ้มที่สุด(หรือการเขียนท่อนต้องแฟ้มที่พวกเขาได้รับถ้าหากแฟ้มใหญ่),แต่เก็บไว้ในใจนั่นคุณอาจจะต้องการใช้งานล็อคนั่นเหมือนกันถ้าคุณอ่านและ/หรือเขียนไปยังแฟ้มเดียวกันอยู่อื่นส่วนหนึ่งของสคริปต์