diff --git a/app/portable.py b/app/portable.py index 7449035..b4c2621 100644 --- a/app/portable.py +++ b/app/portable.py @@ -1,13 +1,20 @@ import flask +import re import requests -from flask import request +import uuid +import threading +import time +from flask import request, Response from bs4 import BeautifulSoup -from urllib.parse import urlparse, urljoin +from urllib.parse import urlparse, urljoin, quote app = flask.Flask(__name__) googlebot_headers = { "User-Agent": "Mozilla/5.0 (Linux; Android 6.0.1; Nexus 5X Build/MMB29P) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.6533.119 Mobile Safari/537.36 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)" } + +jobs = {} + html = """ @@ -17,8 +24,9 @@ html = """ 13ft Ladder - + @@ -156,12 +249,28 @@ html = """ -
-

Enter Website Link

- - - -
+ +
+
+

Enter Website Link

+ + + +
+
+ + """ + def add_base_tag(html_content, original_url): soup = BeautifulSoup(html_content, 'html.parser') parsed_url = urlparse(original_url) base_url = f"{parsed_url.scheme}://{parsed_url.netloc}/" - - # Handle paths that are not root, e.g., "https://x.com/some/path/w.html" + if parsed_url.path and not parsed_url.path.endswith('/'): base_url = urljoin(base_url, parsed_url.path.rsplit('/', 1)[0] + '/') base_tag = soup.find('base') - - print(base_url) + if not base_tag: new_base_tag = soup.new_tag('base', href=base_url) if soup.head: @@ -206,22 +424,240 @@ def add_base_tag(html_content, original_url): head_tag = soup.new_tag('head') head_tag.insert(0, new_base_tag) soup.insert(0, head_tag) - + return str(soup) -def bypass_paywall(url): - """ - Bypass paywall for a given url - """ - if url.startswith("http"): - response = requests.get(url, headers=googlebot_headers) - response.encoding = response.apparent_encoding - return add_base_tag(response.text, response.url) +CHALLENGE_SIGNATURES = [ + "just a moment", + "checking your browser", + "cloudflare", + "cf-challenge", + "ddos protection", + "attention required", + "enable javascript and cookies to continue", + "please verify you are a human", +] + + +def is_challenge_page(html_text): + if not html_text: + return False + lower = html_text[:8000].lower() + hits = sum(1 for sig in CHALLENGE_SIGNATURES if sig in lower) + if hits >= 1 and len(html_text) < 20000: + return True + return hits >= 2 + + +def set_step(job_id, step): + if job_id and job_id in jobs: + jobs[job_id]['step'] = step + + +REAL_BROWSER_HEADERS = { + "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "en-US,en;q=0.9", +} + +ARCHIVE_PH_MIRRORS = ["archive.ph", "archive.today", "archive.is", "archive.li"] + +FREEDIUM_MIRRORS = ["freedium-mirror.cfd", "freedium.cfd"] + + +def is_medium_url(url): try: - return bypass_paywall("https://" + url) + host = (urlparse(url).hostname or "").lower() + return host == "medium.com" or host.endswith(".medium.com") + except Exception: + return False + + +def fetch_via_freedium(url, job_id=None): + """Freedium is a Medium-specific paywall bypass service.""" + set_step(job_id, 'fallback_freedium') + clean_url = url.split('?')[0] + for mirror in FREEDIUM_MIRRORS: + try: + freedium_url = f"https://{mirror}/{clean_url}" + resp = requests.get( + freedium_url, + headers=REAL_BROWSER_HEADERS, + timeout=30, + allow_redirects=True, + ) + if resp.status_code != 200: + continue + resp.encoding = resp.apparent_encoding + text = resp.text + if is_challenge_page(text): + continue + if 'main-content' not in text or len(text) < 2000: + continue + return text, resp.url + except Exception: + continue + return None, None + + +def fetch_via_archive_org(url, job_id=None): + set_step(job_id, 'fallback_org') + wayback_api = f"https://archive.org/wayback/available?url={url}" + try: + meta = requests.get(wayback_api, timeout=15).json() + snapshot = meta.get("archived_snapshots", {}).get("closest", {}) + if not snapshot.get("available"): + return None, None + archived_url = snapshot["url"] + if archived_url.startswith("http://"): + archived_url = "https://" + archived_url[len("http://"):] + if "/web/" in archived_url and "id_/" not in archived_url: + archived_url = archived_url.replace("/web/", "/web/", 1) + archived_url = archived_url.replace( + archived_url.split("/web/")[1].split("/")[0], + archived_url.split("/web/")[1].split("/")[0] + "id_", + 1, + ) + resp = requests.get(archived_url, headers=googlebot_headers, timeout=30) + resp.encoding = resp.apparent_encoding + return resp.text, resp.url + except Exception: + return None, None + + +def fetch_via_archive_ph(url, job_id=None): + """Try archive.today / archive.ph / archive.is — all mirrors of the same service.""" + set_step(job_id, 'fallback_ph') + for mirror in ARCHIVE_PH_MIRRORS: + try: + newest_url = f"https://{mirror}/newest/{url}" + resp = requests.get( + newest_url, + headers=REAL_BROWSER_HEADERS, + timeout=20, + allow_redirects=True, + ) + if resp.status_code != 200: + continue + + final = resp.url + if not re.search(r"https?://archive\.(ph|today|is|li)/[A-Za-z0-9]{4,}", final): + continue + if re.search(r"archive\.(ph|today|is|li)/newest/", final): + continue + + text = resp.text + if is_challenge_page(text): + continue + + lower_head = text[:5000].lower() + if "no results" in lower_head or "0 captures" in lower_head: + continue + + return text, final + except Exception: + continue + return None, None + + +def bypass_paywall(url, job_id=None): + set_step(job_id, 'connect') + + if not url.startswith("http"): + url = "https://" + url + + medium = is_medium_url(url) + html_text = "" + final_url = url + + if medium: + freedium_html, freedium_final = fetch_via_freedium(url, job_id) + if freedium_html: + set_step(job_id, 'process') + result = add_base_tag(freedium_html, freedium_final) + set_step(job_id, 'cleanup') + return result + else: + set_step(job_id, 'fetch') + response = requests.get(url, headers=googlebot_headers, timeout=30) + response.encoding = response.apparent_encoding + html_text = response.text + final_url = response.url + + set_step(job_id, 'detect') + if not html_text or is_challenge_page(html_text): + recovered = False + + if not medium: + 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 + else: + 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 = ( + "Medium served an anti-bot challenge and Freedium / archive.org / " + "archive.today all came back empty. The article may be too new to " + "have been archived yet." + ) + else: + msg = ( + "The site served an anti-bot challenge (Cloudflare or similar) " + "and no usable snapshot was found on archive.org or archive.today. " + "This site is actively blocking automated requests." + ) + raise RuntimeError(msg) + + set_step(job_id, 'process') + result = add_base_tag(html_text, final_url) + + set_step(job_id, 'cleanup') + return result + + +def fetch_worker(job_id, url): + try: + result = bypass_paywall(url, job_id) + jobs[job_id]['result'] = result + jobs[job_id]['step'] = 'done' + except requests.exceptions.Timeout: + jobs[job_id]['error'] = 'The website took too long to respond (30s timeout). Try again later.' + jobs[job_id]['step'] = 'error' + except requests.exceptions.ConnectionError: + jobs[job_id]['error'] = 'Could not connect to the website. Check the URL and try again.' + jobs[job_id]['step'] = 'error' except requests.exceptions.RequestException as e: - return bypass_paywall("http://" + url) + jobs[job_id]['error'] = f'Failed to fetch the page: {e}' + jobs[job_id]['step'] = 'error' + except RuntimeError as e: + jobs[job_id]['error'] = str(e) + jobs[job_id]['step'] = 'error' + except Exception as e: + jobs[job_id]['error'] = f'Unexpected error: {e}' + jobs[job_id]['step'] = 'error' @app.route("/") @@ -229,6 +665,45 @@ def main_page(): return html +@app.route("/status/") +def status_stream(url): + import json + + job_id = str(uuid.uuid4()) + jobs[job_id] = {'step': 'queued', 'result': None, 'error': None} + + thread = threading.Thread(target=fetch_worker, args=(job_id, url)) + thread.daemon = True + thread.start() + + def generate(): + last_step = None + while True: + job = jobs.get(job_id) + 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'})}\n\n" + yield f"event: done\ndata: {json.dumps({'html': 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) + + 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"] @@ -236,8 +711,8 @@ def show_article(): return bypass_paywall(link) except requests.exceptions.RequestException as e: return str(e), 400 - except e: - raise e + except Exception as exc: + raise exc @app.route("/", defaults={"path": ""}) @@ -251,7 +726,7 @@ def get_article(path): return bypass_paywall(actual_url) except requests.exceptions.RequestException as e: return str(e), 400 - except e: + except Exception as e: raise e else: return "Invalid URL", 400