import json
import os
import flask
import re
import requests
import uuid
import threading
import time
from urllib.parse import urljoin, urlparse, quote
from flask import request, Response
from bs4 import BeautifulSoup
app = flask.Flask(__name__)
_DEFAULT_STRINGS = {
"heading": "Enter Website Link",
"label": "Link of the website you want to remove paywall for:",
"submit": "Submit",
"toggle_dark_mode": "Toggle Dark Mode",
"status_heading": "Fetching Article",
"error_title": "Something went wrong",
"retry_button": "Try Another URL",
"elapsed_template": "{seconds}s elapsed",
"connection_lost_error": "Connection to server lost. The page may be taking too long to load.",
"step_connect": "Connecting to website...",
"step_fetch": "Downloading page content...",
"step_detect": "Checking for anti-bot challenges...",
"step_fallback_freedium": "Trying Freedium (Medium bypass)...",
"step_fallback_org": "Trying archive.org snapshot...",
"step_fallback_ph": "Trying archive.today / archive.ph snapshot...",
"step_process": "Processing article...",
"step_cleanup": "Cleaning up & preparing view...",
"step_done": "Done!",
"invalid_url": "Invalid URL",
"medium_challenge_error": (
"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."
),
"challenge_error": (
"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."
),
"timeout_error": "The website took too long to respond (30s timeout). Try again later.",
"connection_error": "Could not connect to the website. Check the URL and try again.",
"request_error": "Failed to fetch the page: {error}",
"unexpected_error": "Unexpected error: {error}",
}
def load_strings():
locale = os.environ.get("LOCALE", "en")
if not re.match(r"^[a-zA-Z0-9_-]+$", locale):
locale = "en"
strings = dict(_DEFAULT_STRINGS)
locale_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "locales")
locale_file = os.path.join(locale_dir, f"{locale}.json")
if locale != "en":
try:
with open(locale_file, "r", encoding="utf-8") as file:
strings.update(json.load(file))
except (FileNotFoundError, json.JSONDecodeError, OSError):
locale = "en"
strings["lang"] = locale
return strings
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 = {}
jobs_lock = threading.Lock()
html = """
13ft Ladder
{{ status_heading }}
"""
def build_steps(strings):
return [
{"id": "connect", "label": strings["step_connect"]},
{"id": "fetch", "label": strings["step_fetch"]},
{"id": "detect", "label": strings["step_detect"]},
{"id": "fallback_freedium", "label": strings["step_fallback_freedium"]},
{"id": "fallback_org", "label": strings["step_fallback_org"]},
{"id": "fallback_ph", "label": strings["step_fallback_ph"]},
{"id": "process", "label": strings["step_process"]},
{"id": "cleanup", "label": strings["step_cleanup"]},
{"id": "done", "label": strings["step_done"]},
]
def render_main_page():
strings = load_strings()
ui_strings = {
"error_title": strings["error_title"],
"retry_button": strings["retry_button"],
"elapsed_template": strings["elapsed_template"],
"connection_lost_error": strings["connection_lost_error"],
}
return flask.render_template_string(
html,
steps=build_steps(strings),
ui_strings=ui_strings,
**strings,
)
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 = "https://archive.org/wayback/available"
try:
meta = requests.get(wayback_api, params={"url": url}, 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(
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/{quote(url, safe=':/')}"
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, strings, 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
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 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, strings):
try:
result = bypass_paywall(url, strings, job_id)
with jobs_lock:
jobs[job_id]['result'] = result
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"].format(error=e)
jobs[job_id]['step'] = 'error'
except RuntimeError as e:
with jobs_lock:
jobs[job_id]['error'] = str(e)
jobs[job_id]['step'] = 'error'
except Exception as e:
with jobs_lock:
jobs[job_id]['error'] = strings["unexpected_error"].format(error=e)
jobs[job_id]['step'] = 'error'
@app.route("/")
def main_page():
return render_main_page()
@app.route("/status/")
def status_stream(url):
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'})}\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)
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.RequestException as e:
return strings["request_error"].format(error=e), 400
except Exception as e:
raise e
@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("/")
try:
return bypass_paywall(actual_url, strings)
except requests.exceptions.RequestException as e:
return strings["request_error"].format(error=e), 400
except Exception as e:
raise e
else:
return strings["invalid_url"], 400
if __name__ == "__main__":
app.run(host="0.0.0.0", port=os.getenv("PORT") or 5000, debug=False)