refactor: improve error handling and response formatting in API endpoints

batch.py

@@ -3,18 +3,25 @@ import json
 from pathlib import Path
 import os
 
-COOKIES = "cookies.txt"  # Get them using the browser extension "Get cookies.txt" and export them while logged in to your account.
-YTDLP = "yt-dlp.exe"  # Change this if you aren't on Windows. Get the latest release from github
+# Get cookies using the browser extension "Get cookies.txt" and export while
+# logged in to your account.
+COOKIES = "cookies.txt"
+
+# Change this if you aren't on Windows. Get the latest release from GitHub.
+YTDLP = "yt-dlp.exe"
 OUTPUT_DIR = Path(__file__).parent / "YOUR_OUTPUT_FOLDER"
 
 # Input options:
-# - If `urls.json` exists in the project root it will be used. The JSON should be an array of strings (URLs)
-# - Otherwise set URLs below or pass a custom JSON path via the `URLS_JSON` environment variable.
+# - If `urls.json` exists in the project root it will be used.
+#   The JSON should be an array of strings (URLs).
+# - Otherwise set URLs below or pass a custom JSON path via the
+#   `URLS_JSON` environment variable.
 URLS = [
     # fallback list — leave empty if you'll use urls.json
 ]
 
-URLS_JSON = os.environ.get("URLS_JSON", str(Path(__file__).parent / "urls.json"))
+default_urls = str(Path(__file__).parent / "urls.json")
+URLS_JSON = os.environ.get("URLS_JSON", default_urls)
 
 RESET = "\033[0m"
 BOLD = "\033[1m"
@@ -44,8 +51,13 @@ def fetch_info(url):
         "generic:impersonate",
         url,
     ]
+    cwd_dir = Path(__file__).parent
     result = subprocess.run(
-        cmd, capture_output=True, text=True, timeout=30, cwd=Path(__file__).parent
+        cmd,
+        capture_output=True,
+        text=True,
+        timeout=30,
+        cwd=cwd_dir,
     )
     if result.returncode != 0:
         return None, result.stderr.strip()
@@ -92,30 +104,40 @@ def main():
                 data = json.load(fh)
                 if isinstance(data, list) and data:
                     URLS.clear()
-                    URLS.extend([u for u in data if isinstance(u, str) and u.strip()])
+                    for u in data:
+                        if isinstance(u, str) and u.strip():
+                            URLS.append(u)
                     total = len(URLS)
     except Exception as e:
         log(f"Failed to load URLs from {URLS_JSON}: {e}", YELLOW)
 
     ok, skipped, failed = [], [], []
 
-    log(f"\n{'━'*55}", CYAN)
-    log(f"  yt-dlp batch — {total} posts → {OUTPUT_DIR.name}/", BOLD)
-    log(f"{'━'*55}\n", CYAN)
+    log("\n" + "━" * 55, CYAN)
+    log("  yt-dlp batch — {} posts → {}/".format(total, OUTPUT_DIR.name), BOLD)
+    log("" + "━" * 55 + "\n", CYAN)
 
     for i, url in enumerate(URLS, 1):
         post_id = get_post_id(url)
-        log(f"[{i:02d}/{total}] {DIM}{url}{RESET}")
+        header = "[{:02d}/{:d}]".format(i, total)
+        log(header + " " + DIM + url + RESET)
 
         # Fetch metadata first to get the title
-        log(f"       fetching info…", DIM)
+        log("       fetching info…", DIM)
         info, err = fetch_info(url)
 
         if info is None:
             # Likely a text post or unavailable
-            log(
-                f"       {YELLOW}⚠ skipped — no media ({err[:80] if err else 'no video found'}){RESET}"
+            reason = err[:80] if err else "no video found"
+            msg = (
+                "       "
+                + YELLOW
+                + "⚠ skipped — no media ("
+                + reason
+                + ")"
+                + RESET
             )
+            log(msg)
             skipped.append((url, err))
             print()
             continue
@@ -125,24 +147,27 @@ def main():
         out_dir = OUTPUT_DIR / folder_name
         out_dir.mkdir(parents=True, exist_ok=True)
 
-        log(f"       title: {BOLD}{title[:55]}{RESET}")
-        log(f"       dir:   {out_dir}")
-        log(f"       downloading…", DIM)
+        log("       title: " + BOLD + title[:55] + RESET)
+        log("       dir:   " + str(out_dir))
+        log("       downloading…", DIM)
 
         success, output = download(url, out_dir)
 
         if success:
             # Find what was downloaded
             files = list(out_dir.iterdir())
-            sizes = [f"{f.stat().st_size / 1e6:.1f} MB" for f in files if f.is_file()]
-            log(
-                f"       {GREEN}✓ done — {', '.join(sizes) if sizes else 'file saved'}{RESET}"
-            )
+            sizes = []
+            for f in files:
+                if f.is_file():
+                    sizes.append("{:.1f} MB".format(f.stat().st_size / 1e6))
+            msg = ", ".join(sizes) if sizes else "file saved"
+            log("       " + GREEN + "✓ done — " + msg + RESET)
             ok.append(url)
         else:
-            # Check if it's just no video (text post that slipped through info check)
+            # Check if it's just no video
+            # (text post that slipped through info check)
             if "no video" in output.lower() or "no formats" in output.lower():
-                log(f"       {YELLOW}⚠ skipped — text post{RESET}")
+                log("       " + YELLOW + "⚠ skipped — text post" + RESET)
                 skipped.append((url, "no video content"))
                 # Remove empty dir
                 try:
@@ -150,31 +175,43 @@ def main():
                 except OSError:
                     pass
             else:
-                log(f"       {RED}✗ failed{RESET}")
+                log("       " + RED + "✗ failed" + RESET)
                 # Print last few lines of output for context
                 for line in output.splitlines()[-3:]:
-                    log(f"         {DIM}{line}{RESET}")
-                failed.append(
-                    (url, output.splitlines()[-1] if output else "unknown error")
-                )
+                    log("         " + DIM + line + RESET)
+                if output:
+                    last_err = output.splitlines()[-1]
+                else:
+                    last_err = "unknown error"
+                failed.append((url, last_err))
 
         print()
 
     # Summary
-    log(f"{'━'*55}", CYAN)
-    log(f"  Summary", BOLD)
-    log(f"{'━'*55}", CYAN)
-    log(f"  {GREEN}✓ downloaded: {len(ok)}{RESET}")
-    log(f"  {YELLOW}⚠ skipped (text/no media): {len(skipped)}{RESET}")
-    log(f"  {RED}✗ failed: {len(failed)}{RESET}")
+    log("" + "━" * 55, CYAN)
+    log("  Summary", BOLD)
+    log("" + "━" * 55, CYAN)
+    log("  " + GREEN + "✓ downloaded: " + str(len(ok)) + RESET)
+    skipped_msg = (
+        "  " + YELLOW + "⚠ skipped (text/no media): "
+        + str(len(skipped))
+        + RESET
+    )
+    log(skipped_msg)
+    failed_msg = (
+        "  " + RED + "✗ failed: "
+        + str(len(failed))
+        + RESET
+    )
+    log(failed_msg)
 
     if failed:
-        log(f"\n  Failed URLs:", RED)
+        log("\n  Failed URLs:", RED)
         for url, reason in failed:
-            log(f"    • {url}", RED)
-            log(f"      {DIM}{reason[:80]}{RESET}")
+            log("    • " + url, RED)
+            log("      " + DIM + reason[:80] + RESET)
 
-    log(f"{'━'*55}\n", CYAN)
+    log("" + "━" * 55 + "\n", CYAN)
 
 
 if __name__ == "__main__":

with_ui.py

@@ -71,15 +71,17 @@ def get_info():
             cmd, capture_output=True, text=True, timeout=30, cwd=DOWNLOAD_DIR
         )
         if result.returncode != 0:
-            return jsonify({"error": result.stderr or "Failed to fetch info"}), 400
+            msg = result.stderr or "Failed to fetch info"
+            return jsonify({"error": msg}), 400
         info = json.loads(result.stdout)
         formats = []
         for f in info.get("formats", []):
+            res = f.get("resolution") or f.get("format_note") or ""
             formats.append(
                 {
                     "id": f.get("format_id"),
                     "ext": f.get("ext"),
-                    "resolution": f.get("resolution") or f.get("format_note") or "",
+                    "resolution": res,
                     "vcodec": f.get("vcodec", "none"),
                     "acodec": f.get("acodec", "none"),
                     "filesize": f.get("filesize") or f.get("filesize_approx"),
@@ -125,7 +127,11 @@ def start_download():
     cmd.append(url)
 
     job_id = str(uuid.uuid4())
-    thread = threading.Thread(target=run_ytdlp, args=(job_id, cmd), daemon=True)
+    thread = threading.Thread(
+        target=run_ytdlp,
+        args=(job_id, cmd),
+        daemon=True,
+    )
     thread.start()
 
     return jsonify({"job_id": job_id})
@@ -139,14 +145,17 @@ def job_status(job_id):
             with jobs_lock:
                 job = jobs.get(job_id)
             if not job:
-                yield f"data: {json.dumps({'error': 'Job not found'})}\n\n"
+                payload = json.dumps({"error": "Job not found"})
+                yield "data: " + payload + "\n\n"
                 break
             lines = job["lines"]
             while sent < len(lines):
-                yield f"data: {json.dumps({'line': lines[sent]})}\n\n"
+                payload = json.dumps({"line": lines[sent]})
+                yield "data: " + payload + "\n\n"
                 sent += 1
             if job["done"]:
-                yield f"data: {json.dumps({'done': True, 'error': job['error']})}\n\n"
+                payload = json.dumps({"done": True, "error": job["error"]})
+                yield "data: " + payload + "\n\n"
                 break
             time.sleep(0.2)
 
@@ -158,6 +167,6 @@ def job_status(job_id):
 
 
 if __name__ == "__main__":
-    print(f"[yt-dlp UI] Serving on http://localhost:5000")
-    print(f"[yt-dlp UI] Download directory: {DOWNLOAD_DIR}")
+    print("[yt-dlp UI] Serving on http://localhost:5000")
+    print("[yt-dlp UI] Download directory: {}".format(DOWNLOAD_DIR))
     app.run(debug=False, host="0.0.0.0", port=5000, threaded=True)