13ft Ladder

skeleton, so

# on its own is not proof that the post was rendered. FREEDIUM_CONTENT_MARKERS = ("main-content", "= 500: html_text = "" set_step(job_id, 'detect') if not html_text or is_challenge_page(html_text): recovered = False archived_html, archived_url = fetch_via_archive_org(url, job_id) if archived_html and not is_challenge_page(archived_html): html_text = archived_html final_url = archived_url recovered = True if not recovered: archived_html, archived_url = fetch_via_archive_ph(url, job_id) if archived_html and not is_challenge_page(archived_html): html_text = archived_html final_url = archived_url recovered = True if not recovered: if medium: msg = strings["medium_challenge_error"] else: msg = strings["challenge_error"] raise UserFacingError(msg) set_step(job_id, 'process') result = process_html_document(html_text, final_url) set_step(job_id, 'cleanup') return result def normalize_cache_url(url): if "://" in url: return "https://" + url.split("://", 1)[1] return "https://" + url def clean_page_cache(): now = time.time() expired = [k for k, v in page_cache.items() if now - v['timestamp'] > PAGE_CACHE_TTL] for k in expired: del page_cache[k] def fetch_worker(job_id, url, strings): try: result = bypass_paywall(url, strings, job_id) # Never cache challenge/error pages — only store clean article HTML. if not is_challenge_page(result): cache_key = normalize_cache_url(url) with page_cache_lock: clean_page_cache() page_cache[cache_key] = {'html': result, 'timestamp': time.time()} else: cache_key = normalize_cache_url(url) with jobs_lock: jobs[job_id]['result'] = cache_key jobs[job_id]['step'] = 'done' except requests.exceptions.Timeout: with jobs_lock: jobs[job_id]['error'] = strings["timeout_error"] jobs[job_id]['step'] = 'error' except requests.exceptions.ConnectionError: with jobs_lock: jobs[job_id]['error'] = strings["connection_error"] jobs[job_id]['step'] = 'error' except requests.exceptions.RequestException as e: with jobs_lock: jobs[job_id]['error'] = strings["request_error"] jobs[job_id]['step'] = 'error' except UserFacingError as error: with jobs_lock: jobs[job_id]['error'] = error.user_message jobs[job_id]['step'] = 'error' except Exception as e: with jobs_lock: jobs[job_id]['error'] = strings["unexpected_error"] jobs[job_id]['step'] = 'error' @app.route("/favicon.ico") def favicon(): logo_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "logo.png") if os.path.exists(logo_path): with open(logo_path, "rb") as f: data = f.read() return Response(data, mimetype="image/png", headers={"Cache-Control": "public, max-age=86400"}) return "", 404 @app.route("/") def main_page(): return render_main_page() @app.route("/status") def status_stream(): raw_url = request.args.get("url", "") url = unquote(raw_url) if not url: return "Missing URL", 400 strings = load_strings() job_id = str(uuid.uuid4()) with jobs_lock: jobs[job_id] = {'step': 'queued', 'result': None, 'error': None} thread = threading.Thread(target=fetch_worker, args=(job_id, url, strings)) thread.daemon = True thread.start() def generate(): last_step = None try: while True: with jobs_lock: job = dict(jobs[job_id]) if job_id in jobs else None if not job: break current = job['step'] if current != last_step: last_step = current if current == 'done': yield f"event: step\ndata: {json.dumps({'step': 'done', 'redirect': '/' + job['result']})}\n\n" break elif current == 'error': yield f"event: error_msg\ndata: {json.dumps({'message': job['error']})}\n\n" break else: yield f"event: step\ndata: {json.dumps({'step': current})}\n\n" time.sleep(0.2) finally: with jobs_lock: jobs.pop(job_id, None) return Response(generate(), mimetype='text/event-stream', headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}) @app.route("/article", methods=["POST"]) def show_article(): link = flask.request.form["link"] strings = load_strings() try: return bypass_paywall(link, strings) except requests.exceptions.Timeout: return strings["timeout_error"], 400 except requests.exceptions.ConnectionError: return strings["connection_error"], 400 except requests.exceptions.RequestException: return strings["request_error"], 400 except UserFacingError as error: return error.user_message, 400 except Exception: return strings["unexpected_error"], 500 @app.route("/", defaults={"path": ""}) @app.route("/", methods=["GET"]) def get_article(path): strings = load_strings() full_url = request.url parts = full_url.split("/", 4) if len(parts) >= 5: actual_url = "https://" + parts[4].lstrip("/") with page_cache_lock: cached = page_cache.get(actual_url) if cached: return cached['html'] try: return bypass_paywall(actual_url, strings) except requests.exceptions.Timeout: return strings["timeout_error"], 400 except requests.exceptions.ConnectionError: return strings["connection_error"], 400 except requests.exceptions.RequestException: return strings["request_error"], 400 except UserFacingError as error: return error.user_message, 400 except Exception: return strings["unexpected_error"], 500 else: return strings["invalid_url"], 400 if __name__ == "__main__": app.run(host="0.0.0.0", port=os.getenv("PORT") or 5000, debug=False)

{{ heading }}

{{ status_heading }}