واجهة Bigjpg البرمجية: تكبير الصور دفعة واحدة بلغة بايثون

لواجهة الويب ما يكفي لعدد قليل من الصور. أما مجلد فيه خمسمائة صورة فيستحق الـ API: أربعة مسارات، بلا متصفح، وسكربت تستطيع تشغيله من جديد كما تشاء. النسخة أدناه هي التي نسلّمها لزميل مباشرة: رفع، إنشاء، متابعة، تنزيل، مع معالجة الإخفاقات التي تظهر فعلاً في العمل.

قبل أن تبدأ

• الوصول إلى الـ API ميزة للحسابات المدفوعة. تجد المفتاح في صفحة الـ API بعد تسجيل الدخول بخطة سارية.
• بايثون 3 مع مكتبة requests. لا شيء غير ذلك.
• التكبير يعمل بشكل غير متزامن على السيرفرات، لذا يُنشئ السكربت المهام ثم يتابعها. مهمة من 500 صورة تستغرق زمن أبطأ مهمة في الطابور الذي أنت فيه، لا مجموع زمن الصور.

السكربت

import json
import os
import time
import requests

API = "https://bigjpg.com"
KEY = "YOUR_API_KEY"
STYLE = "art"          # "art" للرسوم، "photo" للصور الفوتوغرافية
NOISE = "1"            # -1 بلا، 0 منخفض، 1 متوسط، 2 عالٍ، 3 الأعلى
X2 = "2"               # 1 = 2x, 2 = 4x, 3 = 8x, 4 = 16x
POLL_SECONDS = 8
MAX_RETRY = 1

HEADERS = {"X-API-KEY": KEY}


def upload(path):
    """Get a presigned URL, PUT the file, return its public URL."""
    name = path.split("/")[-1]
    r = requests.post(
        API + "/api/img_upload",
        headers=HEADERS,
        data={"filename": name, "filetype": "image/jpeg"},
        timeout=60,
    )
    info = r.json()
    if info.get("status") != "ok":
        raise RuntimeError("upload failed: " + str(info))

    with open(path, "rb") as image:
        r = requests.put(
            info["url"],
            data=image,
            headers={"Content-Type": "image/jpeg"},
            timeout=300,
        )
    if r.status_code != 200:
        raise RuntimeError("upload failed with HTTP %s" % r.status_code)
    return info["fileurl"]


def create(image_url):
    """Create an enlargement task, return its task id."""
    payload = {
        "style": STYLE,
        "noise": NOISE,
        "x2": X2,
        "input": image_url,
    }
    r = requests.post(
        API + "/api/task/",
        headers=HEADERS,
        data=json.dumps(payload),
        timeout=60,
    )
    data = r.json()
    if "tid" not in data:
        raise RuntimeError("task failed: " + str(data))
    return data["tid"]


def poll(tids):
    """Query up to 100 task ids at once. Returns task id -> result dict."""
    r = requests.get(API + "/api/task/" + ",".join(tids), timeout=60)
    return r.json()


def download(url, out):
    """Download a finished image, checking the transfer really succeeded."""
    r = requests.get(url, timeout=300)
    if r.status_code != 200:
        raise RuntimeError("download failed with HTTP %s" % r.status_code)
    with open(out, "wb") as f:
        f.write(r.content)
    return out


def main(paths):
    pending = {}
    attempts = {}
    for path in paths:
        try:
            tid = create(upload(path))
        except Exception as e:
            print("skipped", path, e)
            continue
        pending[tid] = path
        attempts[tid] = 0
        print("queued", path, tid)

    while pending:
        for tid, info in poll(list(pending)[:100]).items():
            if info.get("status") == "success":
                out = os.path.splitext(pending[tid])[0] + "_bigjpg.jpg"
                try:
                    download(info["url"], out)
                except Exception as e:
                    print("download failed", pending[tid], e)
                    pending.pop(tid)
                    continue
                print("done", pending.pop(tid), "->", out)

            elif info.get("status") == "error":
                # محاولة إعادة واحدة، ثم نتخلى عن هذا الملف.
                attempts[tid] += 1
                if attempts[tid] <= MAX_RETRY:
                    requests.post(API + "/api/task/" + tid, headers=HEADERS, timeout=60)
                    print("retrying", pending[tid])
                else:
                    print("giving up on", pending.pop(tid))

        if pending:
            time.sleep(POLL_SECONDS)


if __name__ == "__main__":
    import sys
    main(sys.argv[1:])

وظيفة كل استدعاء

التزامن واللياقة

• تابع بوتيرة أقل مما تظن. التكبير ليس لحظيًا؛ 5–10 ثوانٍ بين الاستعلامات تكفي، وسؤال 100 معرف في استدعاء واحد يبقي عدد الطلبات منخفضًا.
• حدّد عدد عمليات الرفع. اثنتان إلى أربع عمليات رفع متزامنة سقف معقول لجهاز واحد. إن أردت إنتاجية أعلى فشغّل نسخًا أكثر، لا خيوطًا أكثر داخل النسخة نفسها.
• أعد المحاولة بتراجع زمني. خطأ الشبكة ليس فشل المهمة. أعد استدعاء HTTP حتى ثلاث مرات بفاصل متزايد قبل أن تحكم على الملف بالعطب.
• راقب remaining_api_calls. كل مهمة تُنشأ تبلّغ عنه؛ أوقف الحلقة عند حد منخفض بدل أن تكتشف السقف من جدار من الأخطاء.
• سجّل الجدول المقابل. احتفظ بملف فيه tid واسم المدخل والإعدادات واسم المخرج. إذا عملت المهمة طوال الليل، فهذا السجل هو الطريقة الوحيدة لمعرفة ما انتهى.

أخطاء ستواجهها فعلاً

• valid_api_key_required — الترويسة ناقصة أو المفتاح غير معروف. تحقق من اسم الترويسة ومن أن المفتاح يخص خطة لم تنتهِ.
• requires_vip — حساب مجاني أو منتهي. الـ API للخطط المدفوعة السارية.
• month_limit — الحصة الشهرية استُهلكت. كل مهمة تبلّغ بالمتبقي، فيمكنك التوقف قبل وصول ذلك.
• param_error — style تكون art أو photo، وnoise من -1 إلى 3، وx2 من 1 إلى 4.
• input_url_error — يجب أن يكون الـ input رابط صورة http أو https متاحًا للعموم. الروابط الخاصة الموقّعة تنتهي صلاحيتها؛ ارفع الملف أولاً ثم مرّر الرابط العام الذي يعود إليك.

تكلفة الدفعة

الصورة الواحدة تُحتسب رصيدًا واحدًا عند 2x و4x و8x، وعند 16x تُحتسب رصيدين. أي أن دفعة من 500 صورة على 4x تساوي 500 رصيد، ولو نفّذتها على 16x لصارت 1000. ولأن 4x هو المضاعف الصحيح لمعظم الصور الفوتوغرافية، فالخيار الأرخص غالبًا هو الأجمل كذلك — والسبب في دليل المضاعفات.

المرجع الكامل للمسارات، مع الوسائط وكل حالات الخطأ، في صفحة الـ API. وإذا وجدت في السكربت ما لا يطابق ما يعود إليك، فتلك الصفحة هي النسخة التي نحدّثها.