import os
from urllib.parse import urljoin, urlparse
import flask
import re
import requests
import uuid
import threading
import time
from flask import request, Response
from bs4 import BeautifulSoup
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 = """
13ft Ladder
"""
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")
if not base_tag:
new_base_tag = soup.new_tag("base", href=base_url)
if soup.head:
soup.head.insert(0, new_base_tag)
else:
head_tag = soup.new_tag("head")
head_tag.insert(0, new_base_tag)
soup.insert(0, head_tag)
return str(soup)
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:
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:
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("/")
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"]
try:
return bypass_paywall(link)
except requests.exceptions.RequestException as e:
return str(e), 400
except Exception as e:
raise e
@app.route("/", defaults={"path": ""})
@app.route("/", methods=["GET"])
def get_article(path):
full_url = request.url
parts = full_url.split("/", 4)
if len(parts) >= 5:
actual_url = "https://" + parts[4].lstrip("/")
try:
return bypass_paywall(actual_url)
except requests.exceptions.RequestException as e:
return str(e), 400
except Exception as e:
raise e
else:
return "Invalid URL", 400
if __name__ == "__main__":
app.run(host="0.0.0.0", port=os.getenv("PORT") or 5000, debug=False)