"""Turbo Listing: build AutoTrader.ca listing payloads from inventory + a pasted PowerShell session and submit them (optional SOCKS proxy). Deterministic parsing; AI is used only for the description (wired in app.py via the provider router).""" import json import re import urllib.parse import requests try: from curl_cffi import requests as crequests except ImportError: crequests = None AUTOTRADER_LISTINGS_URL = "https://www.autotrader.ca/manual-listing-creation/api/private/listings" AUTOTRADER_MODELS_URL = "https://www.autotrader.ca/manual-listing-creation/models/" AUTOTRADER_REFERER_PATH = "https://www.autotrader.ca/manual-listing-creation/private/vehicle-listing/" AUTOTRADER_HOSTS = ("www.autotrader.ca", "autotrader.ca") AUTOTRADER_OFFER_BASE_URL = "https://www.autotrader.ca/offers/cat_ma" AUTOTRADER_REMOVED_MARKER = "This listing no longer exists." def _ps_unescape(s): """Reverse PowerShell backtick escapes inside double-quoted strings.""" if not s: return s return (s.replace("`$", "$").replace('`"', '"').replace("`n", "\n") .replace("`t", "\t").replace("`r", "\r").replace("``", "`")) def _simple_quoted_arg(text, name): """Match -Name "value" for args whose value has no embedded quotes.""" m = re.search(r'-' + name + r'\s+"([^"]*)"', text) return _ps_unescape(m.group(1)) if m else None def parse_powershell_request(text): """Parse a browser-captured PowerShell Invoke-WebRequest into a structured request. Returns {"method","url","cookies","headers","user_agent","content_type","body"}. Raises ValueError on clear parse failures.""" if not text or "Invoke-WebRequest" not in text: raise ValueError("No Invoke-WebRequest found in the pasted request.") cookies = [] for m in re.finditer( r'New-Object\s+System\.Net\.Cookie\(\s*"((?:[^"`]|`.)*)"\s*,\s*"((?:[^"`]|`.)*)"\s*,\s*"((?:[^"`]|`.)*)"\s*,\s*"((?:[^"`]|`.)*)"\s*\)', text, ): name, value, path, domain = m.groups() cookies.append({ "name": _ps_unescape(name), "value": _ps_unescape(value), "path": _ps_unescape(path), "domain": _ps_unescape(domain), }) headers = {} hm = re.search(r'-Headers\s*@\{(.*?)\}\s*`', text, re.S) if hm: for km in re.finditer(r'"([^"]+)"\s*=\s*"((?:[^"`]|`.)*)"', hm.group(1)): headers[_ps_unescape(km.group(1))] = _ps_unescape(km.group(2)) method = _simple_quoted_arg(text, "Method") or "POST" url = _simple_quoted_arg(text, "Uri") user_agent = _simple_quoted_arg(text, "UserAgent") if not user_agent: # Chrome-style "Copy as PowerShell" sets the UA as a session property, not a -UserAgent param. m = re.search(r'\$session\.UserAgent\s*=\s*"((?:[^"`]|`.)*)"', text) if m: user_agent = _ps_unescape(m.group(1)) if not user_agent: user_agent = headers.get("User-Agent") or headers.get("user-agent") or "" content_type = _simple_quoted_arg(text, "ContentType") # Body: a double-quoted PowerShell string whose embedded quotes are backtick-escaped (`"). # It is typically the last argument; take from `-Body "` to the final quote. body = None bm = re.search(r'-Body\s+"', text) if bm: start = bm.end() chunk = text[start:].rstrip() if chunk.endswith('"'): chunk = chunk[:-1] body = _ps_unescape(chunk) if not url: raise ValueError("Could not find -Uri in the request.") if not cookies: raise ValueError("No session cookies found in the request.") return { "method": method, "url": url, "cookies": cookies, "headers": headers, "user_agent": user_agent or "", "content_type": content_type or "application/json", "body": body or "", } def _norm(s): return re.sub(r'[^a-z0-9]', '', (s or '').lower()) BODY_TYPE_SYNONYMS = { "suv": 14, "crossover": 14, "cuv": 14, "sedan": 6, "saloon": 6, "coupe": 3, "coup": 3, "convertible": 2, "cabriolet": 2, "roadster": 2, "hatchback": 1, "hatch": 1, "wagon": 5, "estate": 5, "stationwagon": 5, "minivan": 12, "van": 12, "pickup": 15, "pickuptruck": 15, "truck": 15, "pickup truck": 15, } def resolve_make_id(db, make_name): """Exact normalized match against turbo_makes. Returns int id or None.""" n = _norm(make_name) if not n: return None for row in db.execute("SELECT id, name FROM turbo_makes").fetchall(): if _norm(row["name"]) == n: return row["id"] return None def resolve_model_id(db, make_id, model_name): """Resolve an inventory model name to a turbo_models id for the given make. Exact normalized match first. If that fails, fall back to a PREFIX match: pick the longest AutoTrader model whose normalized name is a prefix of the inventory model's normalized name. AutoTrader's models are base names (e.g. "Civic", "Wrangler", "Range Rover"); inventory often has variants ("Civic Hatchback", "Wrangler 4xe", "Range Rover Sport") that would otherwise never resolve. The longest-prefix rule prefers "Range Rover" over "Range" when both could match. Returns int id or None.""" n = _norm(model_name) if not n or make_id is None: return None best_id = None best_len = 0 for row in db.execute("SELECT id, name FROM turbo_models WHERE make_id=?", (make_id,)).fetchall(): rn = _norm(row["name"]) if not rn: continue if rn == n: return row["id"] # exact match always wins if n.startswith(rn) and len(rn) > best_len: best_id = row["id"] best_len = len(rn) return best_id def resolve_body_type_id(body_style): """Resolve our body_style text to an AutoTrader bodyType id (or None).""" n = _norm(body_style) if not n: return None if n in BODY_TYPE_SYNONYMS: return BODY_TYPE_SYNONYMS[n] # try the seeded body-type labels directly label_map = {"convertible": 2, "coupe": 3, "hatchback": 1, "minivan": 12, "pickuptruck": 15, "sedan": 6, "suv": 14, "wagon": 5, "others": 7} return label_map.get(n) def vin_for_payload(include_vin, vin_replacement, inventory_vin): """Turbo Listing VIN logic. Case 1: include + no replacement -> inventory VIN. Case 2: not include + no replacement -> None. Case 3: replacement present -> replacement wins (trimmed, uppercased). Empty inventory VIN + no replacement -> None.""" rep = (vin_replacement or "").strip().upper() if rep: return rep if include_vin and (inventory_vin or "").strip(): return (inventory_vin or "").strip().upper() return None def override_number(value, inventory_value): """Return override if it's a valid number, else the inventory value (as number or None).""" if value is None or str(value).strip() == "": return _to_number(inventory_value) n = _to_number(value) return n if n is not None else _to_number(inventory_value) def _to_number(value): try: return int(float(str(value).replace(",", "").replace("$", "").strip())) except (ValueError, TypeError): return None def _g(row, name, default=None): try: v = row[name] if name in row.keys() else None except (KeyError, AttributeError, TypeError): v = None return v if v not in (None, "") else default def _map_fuel_category(fuel): n = _norm(fuel) if not n: return None if "electric" in n and "plug" not in n and "hybrid" not in n: return "E" if "plugin" in n or (n.startswith("plug")): return "P" if "hybrid" in n: return "P" if "diesel" in n: return "D" if "gas" in n or "petrol" in n: return "B" return None def _map_transmission(value): n = _norm(value) if not n: return None if "auto" in n or "cvt" in n or "dct" in n or "tiptronic" in n: return "A" if "manual" in n or "stick" in n: return "M" return None def _map_drivetrain(value): n = _norm(value) if not n: return None if "awd" in n or "4wd" in n or "4x4" in n or "allwheel" in n or "fourwheel" in n: return "A" if "fwd" in n or "front" in n: return "F" if "rwd" in n or "rear" in n: return "R" return None def build_payload(listing, overrides, db, description=""): """Build the full AutoTrader listing payload from an inventory row + overrides. trim is always null (integer-only field); the trim text goes in modelVersion. Raises RuntimeError if make/model can't be resolved.""" make_id = resolve_make_id(db, _g(listing, "make")) model_id = resolve_model_id(db, make_id, _g(listing, "model")) if make_id else None if make_id is None: raise RuntimeError(f"Could not resolve AutoTrader make for '{_g(listing, 'make')}'.") if model_id is None: raise RuntimeError(f"Could not resolve AutoTrader model for '{_g(listing, 'model')}' under make '{_g(listing, 'make')}'. Run Refresh taxonomy.") include_vin = bool(overrides.get("include_vin", True)) vin = vin_for_payload(include_vin, overrides.get("vin_replacement", ""), _g(listing, "vin")) mileage = override_number(overrides.get("mileage_override"), _g(listing, "odometer")) price = override_number(overrides.get("price_override"), _g(listing, "price")) try: model_year = int(float(_g(listing, "year", 0) or 0)) or None except (ValueError, TypeError): model_year = None try: doors = int(float(_g(listing, "doors", 0) or 0)) or None except (ValueError, TypeError): doors = None trim_text = (_g(listing, "trim") or "").strip() or None return { "vehicleType": "C", "vehicleData": { "make": make_id, "model": model_id, "modelVersion": trim_text, "modelName": None, "vin": vin, "licencePlate": None, "eCode": None, "natCode": None, "modelGenerationId": None, "modelVariantId": None, "motorTypeId": None, "trim": None, "modelYear": model_year, "inventoryTags": None, }, "characteristics": { "bodyType": resolve_body_type_id(_g(listing, "body_style")), "bodyColor": None, "seatCount": None, "doorCount": doors, "isMetallic": None, "upholsteryType": None, "upholsteryColor": None, "axleCount": None, "bedCount": None, "slideOutCount": None, "exteriorMaterial": None, "flooringMaterial": None, "fridgePowerType": None, "chassisBedLength": None, "grossVehicleWeight": None, "grossVehicleWeightGrams": None, "hasCarRegistration": None, "loadHeight": None, "loadLength": None, "loadVolume": None, "loadWidth": None, "maximumTowingWeight": None, "payload": None, "totalHeight": None, "totalLength": None, "totalWidth": None, "emptyWeight": None, "emptyWeightGrams": None, "fridgeCapacity": None, "fridgeCapacityMilliliters": None, "wasteTankCapacity": None, "wasteTankCapacityMilliliters": None, "waterTankCapacity": None, "waterTankCapacityMilliliters": None, "chassisManufacturerName": None, "extendedLength": None, "skiStanceWidth": None, "autoTraderTrackDimension": None, "autoTraderVehicleLocation": None, }, "motor": { "power": None, "drivetrain": _map_drivetrain(_g(listing, "drive_line")), "transmission": _map_transmission(_g(listing, "transmission")), "gearCount": None, "cylinderCount": None, "cylinderCapacity": None, "wheelbase": None, "engineManufacturerName": None, "engineCoolingSystem": None, "fuelDeliveryType": None, "axleSpread": None, "cabinAxleDistance": None, "autoTraderFrameType": None, "autoTraderEngineControlModuleName": None, "autoTraderEngineMarketingDescription": None, "fuelCapacity": None, "fuelCapacityMilliliters": None, "oilCapacity": None, "oilCapacityMilliliters": None, "frontAxleWeightRating": None, "frontAxleWeightRatingGrams": None, "rearAxleWeightRating": None, "rearAxleWeightRatingGrams": None, "differentialRatio": None, "brakeType": None, "headCount": None, "steeringType": None, "auxiliaryPower": None, "hullMaterial": None, "engineMountingType": None, "engineCount": None, }, "fuel": {"category": _map_fuel_category(_g(listing, "fuel_type")), "primaryType": None, "additionalTypes": [], "consumption": {"combined": None, "electricCombined": None}, "isPluginHybrid": None}, "emissions": {"co2": None, "efficiencyClass": None, "electricRange": None, "euStandard": None, "germanSticker": None, "hasParticleFilter": None}, "wltp": {"co2Class": None, "co2ClassDischarged": None, "co2EmissionsCombined": None, "co2EmissionsCombinedWeighted": None, "consumptionCombined": None, "consumptionCombinedDischarged": None, "consumptionCombinedWeighted": None, "consumptionElectricCombined": None, "consumptionElectricCombinedWeighted": None, "co2EmissionsDischarged": None, "consumptionCity": None, "consumptionSuburban": None, "consumptionRural": None, "consumptionHighway": None, "consumptionElectricCity": None, "consumptionElectricRural": None, "consumptionElectricSuburban": None, "consumptionElectricHighway": None, "consumptionCityDischarged": None, "consumptionSuburbanDischarged": None, "consumptionRuralDischarged": None, "consumptionHighwayDischarged": None}, "battery": {"ownershipType": None, "chargingTime10to80Minutes": None}, "condition": {"vehicleOfferTypeId": "U", "mileage": mileage, "engineHours": None, "firstRegistration": None, "previousOwnerCount": None, "hadAccident": False, "isDamaged": None, "roadworthy": None, "isReserved": None, "hasFullServiceHistory": None, "isNonSmoking": None, "isCabinClean": None, "lastCamBeltServiceDate": None, "lastTechnicalServiceDate": None, "nextInspectionDate": None}, "equipment": {"equipment": [], "alloyWheelSize": None}, "description": description or "", "media": {"images": []}, "publish": True, "userType": "P", "testMode": False, "prices": {"public": {"price": price, "currency": "CAD", "isNegotiable": None, "isTaxDeductible": None, "netPrice": None, "vatRate": None, "germanEnvironmentalGrant": None}}, "financingOffer": None, } def validate_request(parsed, payload, proxy=None): """Return (ok, errors[]).""" errors = [] try: host = urllib.parse.urlparse(parsed["url"]).hostname or "" except Exception: host = "" if host not in AUTOTRADER_HOSTS: errors.append("URL host must be autotrader.ca.") if not parsed.get("cookies"): errors.append("No session cookies parsed.") if not parsed.get("user_agent"): errors.append("No User-Agent parsed.") if "application/json" not in (parsed.get("content_type") or "").lower(): errors.append("Content-Type must be application/json.") try: json.dumps(payload) except Exception as e: errors.append(f"Payload is not valid JSON: {e}") vd = payload.get("vehicleData", {}) if not isinstance(vd.get("make"), int): errors.append("make must be an integer.") if not isinstance(vd.get("model"), int): errors.append("model must be an integer.") if vd.get("trim") is not None and not isinstance(vd["trim"], int): errors.append("trim must be null or an integer (never a string).") if vd.get("modelYear") is not None and not isinstance(vd["modelYear"], int): errors.append("modelYear must be an integer.") cond = payload.get("condition", {}) if cond.get("mileage") is not None and not isinstance(cond["mileage"], int): errors.append("mileage must be numeric or null.") price = (payload.get("prices", {}) or {}).get("public", {}).get("price") if price is None or not isinstance(price, int): errors.append("price must be a number.") if vd.get("vin") is not None and not isinstance(vd["vin"], str): errors.append("vin must be null or a string.") if proxy: try: normalize_proxy(proxy) except ValueError as e: errors.append(str(e)) return (len(errors) == 0), errors def _summarize_error(data, text, status): if isinstance(data, dict) and data.get("errors"): errs = data["errors"] if isinstance(errs, list) and errs: first = errs[0] return f"AutoTrader {status}: {first.get('message') or first}" return f"AutoTrader {status}: {errs}" if status in (401, 403): return f"AutoTrader {status}: session expired or blocked (refresh the browser session and re-paste)." snippet = re.sub(r'\s+', ' ', text[:200]).strip() return f"AutoTrader {status}: {snippet}" def parse_response(resp): """Parse an AutoTrader HTTP response into a result dict.""" text = resp.text or "" try: data = resp.json() except Exception: data = None is_html_waf = data is None and ( " 0: break # got new cookies, stop trying except Exception: continue new_total = len(merged) orig_total = len(parsed.get("cookies") or []) if new_total <= orig_total: return None, f"No new cookies in response (tried {len(urls_to_try)} URL(s), original={orig_total}, merged={new_total})." # Build refreshed PowerShell text cookie_lines = [] for name, value in merged.items(): esc_val = value.replace('"', '`"') cookie_lines.append(f'$session.Cookies.Add((New-Object System.Net.Cookie("{name}", "{esc_val}", "/", "www.autotrader.ca")))') ua = parsed.get("user_agent") or "" refresh_ps = "$session = New-Object Microsoft.PowerShell.Commands.WebRequestSession\n" if ua: refresh_ps += f'$session.UserAgent = "{ua}"\n' refresh_ps += "\n".join(cookie_lines) refresh_ps += '\nInvoke-WebRequest -UseBasicParsing -Uri "https://www.autotrader.ca/manual-listing-creation/api/private/listings" -Method "POST" -WebSession $session -ContentType "application/json" -Body "{}"' return refresh_ps, None def retrieve_listings(raw_ps, proxy=None, timeout=30): """GET the current active listings for this account via curl_cffi. Returns (list_of_listing_dicts, error_or_None). Each dict has: uuid, status.""" if crequests is None: return [], "curl_cffi is required. Install: pip install curl_cffi" try: parsed = parse_powershell_request(raw_ps) except ValueError as e: return [], f"Session: {e}" headers = {} for k, v in (parsed.get("headers") or {}).items(): headers[k] = v if parsed.get("user_agent"): headers["User-Agent"] = parsed["user_agent"] if parsed.get("cookies"): headers["Cookie"] = "; ".join(f'{c["name"]}={c["value"]}' for c in parsed["cookies"]) headers["Referer"] = "https://www.autotrader.ca/account/listings" headers["Accept"] = "application/json, text/plain, */*" for k in list(headers): if k.lower() in ("content-length", "host", "connection", "authority", "method", "path", "scheme", "content-type"): headers.pop(k, None) proxies = None if proxy: try: proxy_url = proxy if str(proxy).startswith("socks5h://") else normalize_proxy(proxy) proxies = {"http": proxy_url, "https": proxy_url} except ValueError as e: return [], str(e) try: resp = crequests.request(method="GET", url=AUTOTRADER_RETRIEVE_URL, headers=headers, proxies=proxies, timeout=timeout, allow_redirects=False) if not resp.ok: return [], f"HTTP {resp.status_code}: {resp.text[:200]}" data = resp.json() items = data if isinstance(data, list) else (data.get("data") or data.get("listings") or data.get("items") or []) listings = [] for item in items: if isinstance(item, dict): uid = item.get("listingId") or item.get("id") or item.get("uuid") status = (item.get("status") or item.get("listingStatus") or "").strip().upper() if uid: # Keep the raw item so callers can capture WHY a listing is INACTIVE # (AutoTrader's reason/rejection fields) instead of only the status. listings.append({"uuid": str(uid), "status": status, "raw": item}) return listings, None except Exception as e: return [], f"Retrieve failed: {type(e).__name__}: {e}" def at_listing_url(listing_id): """Public offer URL for an AutoTrader listing id (works for both the bare UUID our posts return and the longer slugs the site sometimes uses).""" return f"{AUTOTRADER_OFFER_BASE_URL}{listing_id}" def check_listing_live(raw_ps, listing_id, proxy=None, timeout=30): """Fetch one AutoTrader offer page through the profile's session (cookies + headers + SOCKS, same as every other AutoTrader request) and decide whether the listing is still live. A removed listing serves HTTP 410 Gone with a page containing 'This listing no longer exists.' (both verified against a real removed listing). Returns: {"ok": bool, "live": True|False|None, "http_status": int|None, "error": str|None} live=None means the check itself could not decide (blocked/timeout/unclear).""" if crequests is None: return {"ok": False, "live": None, "http_status": None, "error": "curl_cffi is required. Install: pip install curl_cffi"} if not (listing_id or "").strip(): return {"ok": False, "live": None, "http_status": None, "error": "No listing id."} try: parsed = parse_powershell_request(raw_ps) except ValueError as e: return {"ok": False, "live": None, "http_status": None, "error": f"Session: {e}"} headers = {} for k, v in (parsed.get("headers") or {}).items(): headers[k] = v if parsed.get("user_agent"): headers["User-Agent"] = parsed["user_agent"] if parsed.get("cookies"): headers["Cookie"] = "; ".join(f'{c["name"]}={c["value"]}' for c in parsed["cookies"]) headers["Accept"] = "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8" headers["Referer"] = "https://www.autotrader.ca/" for k in list(headers): if k.lower() in ("content-length", "host", "connection", "authority", "method", "path", "scheme", "content-type"): headers.pop(k, None) proxies = None if proxy: try: proxy_url = proxy if str(proxy).startswith("socks5h://") else normalize_proxy(proxy) proxies = {"http": proxy_url, "https": proxy_url} except ValueError as e: return {"ok": False, "live": None, "http_status": None, "error": str(e)} try: resp = crequests.request( method="GET", url=at_listing_url(listing_id), headers=headers, proxies=proxies, timeout=timeout, allow_redirects=True, ) status = resp.status_code if status in (404, 410): return {"ok": True, "live": False, "http_status": status, "error": None} try: body = resp.text or "" except Exception: body = "" if AUTOTRADER_REMOVED_MARKER in body or AUTOTRADER_REMOVED_MARKER.lower() in body.lower(): return {"ok": True, "live": False, "http_status": status, "error": None} if status == 200: return {"ok": True, "live": True, "http_status": status, "error": None} return {"ok": False, "live": None, "http_status": status, "error": f"Unclear response (HTTP {status}) — possibly anti-bot; use the manual open-in-browser check."} except Exception as e: return {"ok": False, "live": None, "http_status": None, "error": f"{type(e).__name__}: {e}"} def submit_listing(raw_ps, payload, proxy=None, timeout=60): """Submit the listing via curl_cffi (browser TLS fingerprint that passes WAF), with optional SOCKS proxy. Reconstructs headers from the parsed PowerShell session — cookies go into a Cookie header to preserve the exact browser format. No powershell.exe needed — works on any OS.""" if crequests is None: return {"ok": False, "http_status": None, "body": "", "json": None, "at_request_id": None, "is_html_waf": False, "error": "curl_cffi is required for submitting (browser TLS fingerprint). Install: pip install curl_cffi"} try: parsed = parse_powershell_request(raw_ps) except ValueError as e: return {"ok": False, "http_status": None, "body": "", "json": None, "at_request_id": None, "is_html_waf": False, "error": f"Session: {e}"} url = parsed["url"] host = urllib.parse.urlparse(url).hostname or "" if host not in AUTOTRADER_HOSTS: return {"ok": False, "http_status": None, "body": "", "json": None, "at_request_id": None, "is_html_waf": False, "error": f"Refusing to submit to non-AutoTrader host: {host}"} # Reconstruct headers (mirrors the working script's reconstructed_headers). headers = {} for k, v in (parsed.get("headers") or {}).items(): headers[k] = v if parsed.get("user_agent"): headers["User-Agent"] = parsed["user_agent"] if parsed.get("content_type"): headers["Content-Type"] = parsed["content_type"] if parsed.get("cookies"): headers["Cookie"] = "; ".join(f'{c["name"]}={c["value"]}' for c in parsed["cookies"]) # Strip transport-managed headers + HTTP/2 pseudo-headers (Chrome's "Copy as # PowerShell" includes :authority/:method/:path/:scheme without the colon — # sending them as regular headers causes WAF 403). for k in list(headers): if k.lower() in ("content-length", "host", "connection", "authority", "method", "path", "scheme"): headers.pop(k, None) body_json = json.dumps(payload) proxies = None if proxy: try: proxy_url = proxy if str(proxy).startswith("socks5h://") else normalize_proxy(proxy) except ValueError as e: return {"ok": False, "http_status": None, "body": "", "json": None, "at_request_id": None, "is_html_waf": False, "error": str(e)} proxies = {"http": proxy_url, "https": proxy_url} try: resp = crequests.request( method=parsed.get("method", "POST"), url=url, headers=headers, data=body_json.encode("utf-8"), proxies=proxies, timeout=timeout, allow_redirects=False, ) return parse_response(resp) except Exception as e: return {"ok": False, "http_status": None, "body": "", "json": None, "at_request_id": None, "is_html_waf": False, "error": f"Network error (no listing created; do not auto-retry): {type(e).__name__}: {e}"} def extract_models(data): """Pull the models list out of an AutoTrader models-endpoint response, which arrives as {"data":{"models":[...]}} (also tolerates a bare list or top-level key).""" if isinstance(data, list): return data if isinstance(data, dict): nested = data.get("data") if isinstance(nested, dict): got = nested.get("models") or nested.get("items") if got: return got return data.get("models") or data.get("items") or [] return [] def normalize_proxy(text): """Normalize a SOCKS proxy to socks5h://user:pass@ip:port (or None).""" if not text: return None s = text.strip() if not s: return None if s.startswith("socks5://") or s.startswith("socks5h://"): rest = s.split("://", 1)[1] if "@" in rest: userinfo, hostport = rest.rsplit("@", 1) if ":" in userinfo: user, pwd = userinfo.split(":", 1) else: user, pwd = userinfo, "" return f"socks5h://{urllib.parse.quote(user, safe='')}:{urllib.parse.quote(pwd, safe='')}@{hostport}" return "socks5h://" + rest # IP:PORT:USER:PASSWORD, with escaped-colon variants (\: normalized to : before splitting) parts = s.replace(r'\:', ':').split(':') if len(parts) == 4: ip, port, user, pwd = parts return f"socks5h://{urllib.parse.quote(user, safe='')}:{urllib.parse.quote(pwd, safe='')}@{ip}:{port}" if len(parts) == 2: return f"socks5h://{parts[0]}:{parts[1]}" raise ValueError(f"Unrecognized proxy format: {text!r}")