diff --git a/k2s_download.py b/k2s_download.py new file mode 100755 index 0000000..7dbb414 --- /dev/null +++ b/k2s_download.py @@ -0,0 +1,344 @@ +#!/usr/bin/env -S uv run +# /// script +# requires-python = ">=3.12" +# dependencies = ["httpx"] +# /// +""" +Keep2Share (k2s.cc) downloader — translated from JDownloader's K2SApi + Keep2ShareCc plugins. + +Usage: + ./k2s_download.py [--user EMAIL] [--pass PASSWORD] [--out DIR] + +Premium accounts skip captchas and get full speed. +Free/anonymous downloads require solving an image captcha (URL printed to terminal). +""" + +import re +import sys +import os +import time +import argparse +import getpass +from pathlib import Path + +import httpx + +# ── constants ──────────────────────────────────────────────────────────────── + +API_BASE = "https://k2s.cc/api/v2" +# Alias domains all resolve to k2s.cc internally +SUPPORTED_DOMAINS = { + "k2s.cc", "keep2share.cc", "keep2.cc", + "fileboom.me", "fboom.me", + "tezfiles.com", "publish2.me", +} + +FILE_ID_RE = re.compile( + r"(?i)/(?:f|file|preview)/(?:info/)?([a-z0-9_\-]{13,})" +) +FOLDER_ID_RE = re.compile(r"(?i)/folder(?:/info)?/([a-z0-9]{13,})") + +HEADERS = { + "User-Agent": "JDownloader", + "Accept-Language": "en-gb, en;q=0.8", + "Content-Type": "application/json", +} + + +# ── helpers ────────────────────────────────────────────────────────────────── + +def extract_file_id(url: str) -> str | None: + m = FILE_ID_RE.search(url) + if not m: + return None + fid = m.group(1) + # fixContentID: lowercase exactly-13-char IDs that aren't "special" + if len(fid) == 13 and fid.isalpha(): + fid = fid.lower() + return fid + + +def is_folder(url: str) -> bool: + return bool(FOLDER_ID_RE.search(url)) + + +def _post(session: httpx.Client, endpoint: str, payload: dict, auth_token: str | None) -> dict: + # auth_token goes in the JSON body, not as a header (per K2SApi.handleDownload bytecode) + if auth_token: + payload = {**payload, "auth_token": auth_token} + r = session.post(f"{API_BASE}{endpoint}", json=payload, headers=HEADERS, timeout=30) + try: + return r.json() + except Exception: + r.raise_for_status() + raise + + +# ── API calls (mirroring K2SApi methods) ───────────────────────────────────── + +def api_login(session: httpx.Client, username: str, password: str) -> str: + """POST /login → returns auth_token.""" + data = _post(session, "/login", {"username": username, "password": password}, None) + token = data.get("auth_token") or data.get("accessToken") + if not token: + raise RuntimeError(f"Login failed: {data.get('message') or data}") + return token + + +def api_get_file_status(session: httpx.Client, file_id: str, auth_token: str | None) -> dict: + """POST /getfilestatus → name, size, is_available, access, md5 …""" + return _post(session, "/getfilestatus", {"id": file_id}, auth_token) + + +def api_get_files_info(session: httpx.Client, file_ids: list[str], auth_token: str | None) -> dict: + """POST /getfilesinfo → bulk file metadata.""" + return _post(session, "/getfilesinfo", {"ids": file_ids}, auth_token) + + +def api_request_captcha(session: httpx.Client, file_id: str) -> dict: + """POST /requestcaptcha → captcha_url + captcha_challenge.""" + return _post(session, "/requestcaptcha", {"file_id": file_id}, None) + + +def api_request_recaptcha(session: httpx.Client, file_id: str) -> dict: + """POST /requestrecaptcha → re_captcha_challenge (site key etc.).""" + return _post(session, "/requestrecaptcha", {"file_id": file_id}, None) + + +def api_get_url( + session: httpx.Client, + file_id: str, + auth_token: str | None, + captcha_challenge: str | None = None, + captcha_response: str | None = None, + free_download_key: str | None = None, +) -> dict: + """POST /geturl → url (direct download link).""" + payload: dict = {"file_id": file_id} + if captcha_challenge: + payload["captcha_challenge"] = captcha_challenge + payload["captcha_response"] = captcha_response + if free_download_key: + payload["free_download_key"] = free_download_key + return _post(session, "/geturl", payload, auth_token) + + +# ── 2captcha ───────────────────────────────────────────────────────────────── + +class TwoCaptcha: + """ + Thin wrapper around the 2captcha API v2. + Mirrors abstractPluginForCaptchaSolverTwoCaptchaAPIV2 from JDownloader: + POST /createTask → taskId + POST /getTaskResult (poll) → solution.text + """ + + API = "https://api.2captcha.com" + POLL_INTERVAL = 5 # seconds between getTaskResult polls + POLL_TIMEOUT = 120 # seconds before giving up + + def __init__(self, api_key: str, client: httpx.Client): + self.api_key = api_key + self.client = client + + def _call(self, endpoint: str, payload: dict) -> dict: + r = self.client.post( + f"{self.API}{endpoint}", + json={"clientKey": self.api_key, **payload}, + headers={"Content-Type": "application/json"}, + timeout=30, + ) + r.raise_for_status() + data = r.json() + err = data.get("errorId") or data.get("errorCode") + if err and err != 0: + raise RuntimeError(f"2captcha error: {data.get('errorCode') or data.get('errorDescription') or data}") + return data + + def solve_image(self, image_url: str) -> str: + """Download image from url, submit as ImageToTextTask, return solution text.""" + img_r = self.client.get(image_url, timeout=30) + img_r.raise_for_status() + b64 = __import__("base64").b64encode(img_r.content).decode() + + task_id = self._call("/createTask", {"task": {"type": "ImageToTextTask", "body": b64}})["taskId"] + print(f" 2captcha task {task_id} submitted, polling…", flush=True) + + deadline = time.monotonic() + self.POLL_TIMEOUT + while time.monotonic() < deadline: + time.sleep(self.POLL_INTERVAL) + result = self._call("/getTaskResult", {"taskId": task_id}) + if result.get("status") == "ready": + text = result["solution"]["text"] + print(f" 2captcha solved: {text}") + return text + print(" …waiting for solution") + + raise RuntimeError(f"2captcha timed out after {self.POLL_TIMEOUT}s") + + +# ── captcha handling ────────────────────────────────────────────────────────── + +def solve_captcha(session: httpx.Client, file_id: str, solver: TwoCaptcha | None) -> tuple[str, str]: + """ + Request a K2S image captcha and return (challenge, response). + Uses 2captcha automatically when a solver is provided, otherwise prompts the user. + """ + cap = api_request_captcha(session, file_id) + captcha_url = cap.get("captcha_url") or cap.get("url") + challenge = cap.get("captcha_challenge") or cap.get("challenge") + if not captcha_url or not challenge: + raise RuntimeError(f"Unexpected captcha response: {cap}") + + if solver: + response = solver.solve_image(captcha_url) + else: + print(f"\nCaptcha required. Open this URL in your browser and read the code:") + print(f" {captcha_url}\n") + response = input("Enter captcha text: ").strip() + + return challenge, response + + +# ── download ────────────────────────────────────────────────────────────────── + +def stream_download(session: httpx.Client, url: str, dest: Path, filename: str): + dest.mkdir(parents=True, exist_ok=True) + out_path = dest / filename + print(f"Downloading → {out_path}") + with session.stream("GET", url, headers=HEADERS, timeout=None) as r: + r.raise_for_status() + total = int(r.headers.get("content-length", 0)) + done = 0 + with open(out_path, "wb") as f: + for chunk in r.iter_bytes(chunk_size=65536): + f.write(chunk) + done += len(chunk) + if total: + pct = done * 100 // total + print(f"\r {pct}% {done // 1024} / {total // 1024} KB", end="", flush=True) + print(f"\nDone: {out_path}") + + +# ── main flow ───────────────────────────────────────────────────────────────── + +def download(url: str, username: str | None, password: str | None, out_dir: Path, twocaptcha_key: str | None = None): + if is_folder(url): + sys.exit("Folder URLs are not supported — add individual file links.") + + file_id = extract_file_id(url) + if not file_id: + sys.exit(f"Could not extract file ID from: {url}") + + session = httpx.Client(follow_redirects=True) + auth_token: str | None = None + solver = TwoCaptcha(twocaptcha_key, session) if twocaptcha_key else None + + # ── 1. authenticate ─────────────────────────────────────────────────────── + if username and password: + print(f"Logging in as {username} …") + auth_token = api_login(session, username, password) + print("Logged in OK.") + + # ── 2. check file ───────────────────────────────────────────────────────── + print(f"Checking file {file_id} …") + info = api_get_file_status(session, file_id, auth_token) + + if not info.get("is_available") and info.get("status") != "success": + err = info.get("message") or info.get("error") or info + sys.exit(f"File not available: {err}") + + name = info.get("name") or file_id + size = info.get("size", 0) + access = info.get("access", "") + print(f" Name : {name}") + print(f" Size : {size // 1024 // 1024} MB" if size else " Size : unknown") + print(f" Access : {access}") + + if access == "private" and not auth_token: + sys.exit("Private file — login required.") + + # ── 3. get download URL ─────────────────────────────────────────────────── + result: dict = {} + direct_url: str | None = None + + if auth_token: + # Premium / logged-in path: no captcha needed + result = api_get_url(session, file_id, auth_token) + direct_url = result.get("url") or result.get("premlink") or result.get("freelink1") + else: + # Free / anonymous path — mirrors K2SApi.handleDownload free flow: + # 1. solve captcha + # 2. POST /geturl → may return free_download_key + time_wait instead of url + # 3. sleep time_wait, re-POST with free_download_key (no captcha) until key is gone + MAX_CAPTCHA_TRIES = 3 + for attempt in range(MAX_CAPTCHA_TRIES): + challenge, cap_response = solve_captcha(session, file_id, solver) + result = api_get_url(session, file_id, None, + captcha_challenge=challenge, + captcha_response=cap_response) + err_code = result.get("errorCode") or result.get("code") + if err_code in (30, 33): + print("Wrong captcha, try again…") + continue + + # Wait loop: server issues a free_download_key and a time_wait before + # it hands over the real url (JDownloader allows up to 4 re-polls). + MAX_WAITS = 4 + for wait_count in range(MAX_WAITS + 1): + free_key = result.get("free_download_key") + if not free_key: + break # server ready — url should be present now + + wait_secs = int((result.get("time_wait") or 0)) + if wait_secs > 180: + sys.exit(f"Rate limited — server wants {wait_secs}s wait, too long.") + if wait_count >= MAX_WAITS: + sys.exit("Too many wait loops from server, giving up.") + + print(f" Server asked to wait {wait_secs}s before download is ready…") + time.sleep(wait_secs) + result = api_get_url(session, file_id, None, free_download_key=free_key) + + direct_url = result.get("url") or result.get("freelink1") + if direct_url: + break + sys.exit(f"Unexpected response after wait loop: {result}") + + if not direct_url: + err = result.get("message") or result.get("error") or result + sys.exit(f"Could not get download URL: {err}") + + # ── 4. download ─────────────────────────────────────────────────────────── + stream_download(session, direct_url, out_dir, name) + + +# ── CLI ─────────────────────────────────────────────────────────────────────── + +def main(): + parser = argparse.ArgumentParser(description="Keep2Share downloader") + parser.add_argument("url", help="k2s.cc / keep2share.cc file URL") + parser.add_argument("--user", "-u", default=os.environ.get("K2S_USER"), help="Account email") + parser.add_argument("--pass", "-p", dest="password", + default=os.environ.get("K2S_PASS"), help="Account password") + parser.add_argument("--out", "-o", default=".", help="Output directory (default: .)") + parser.add_argument("--2captcha", dest="twocaptcha_key", + default=os.environ.get("TWOCAPTCHA_API_KEY"), + help="2captcha API key (or set TWOCAPTCHA_API_KEY env var)") + args = parser.parse_args() + + password = args.password + if args.user and not password: + password = getpass.getpass(f"Password for {args.user}: ") + + download( + url=args.url, + username=args.user, + password=password, + out_dir=Path(args.out), + twocaptcha_key=args.twocaptcha_key, + ) + + +if __name__ == "__main__": + main()