tonym/13ft
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #44 from wasi-master/copilot/apply-review-comments

Browse Source 
Fix review issues from PR #43: XSS, duplicate imports, dead code, URL encoding, threading, deduplication
This commit is contained in:
Wasi Master
2026-05-12 00:29:53 +06:00
committed by GitHub
parent e1dd3925b2 fc9559ad8b
commit 3d4db07d97
1 changed files with 67 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/portable.py
+67 -68
View File
@@ -1,15 +1,13 @@
import os import os
from urllib.parse import urljoin, urlparse
import flask import flask
import re import re
import requests import requests
import uuid import uuid
import threading import threading
import time import time
from urllib.parse import urljoin, urlparse, quote
from flask import request, Response from flask import request, Response
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
from urllib.parse import urlparse, urljoin, quote
app = flask.Flask(__name__) app = flask.Flask(__name__)
googlebot_headers = { googlebot_headers = {
@@ -17,6 +15,7 @@ googlebot_headers = {
} }
jobs = {} jobs = {}
jobs_lock = threading.Lock()
html = """ html = """
<!DOCTYPE html> <!DOCTYPE html>
@@ -340,9 +339,19 @@ html = """
function showError(message) { function showError(message) {
const area = document.getElementById('error-area'); const area = document.getElementById('error-area');
area.innerHTML = '<div class="error-box"><strong>Something went wrong</strong>' + area.innerHTML = '';
message + '</div>' + const box = document.createElement('div');
'<a href="/" class="retry-btn">Try Another URL</a>'; box.className = 'error-box';
const strong = document.createElement('strong');
strong.textContent = 'Something went wrong';
box.appendChild(strong);
box.appendChild(document.createTextNode(message));
area.appendChild(box);
const retry = document.createElement('a');
retry.href = '/';
retry.className = 'retry-btn';
retry.textContent = 'Try Another URL';
area.appendChild(retry);
} }
document.getElementById('url-form').addEventListener('submit', function(e) { document.getElementById('url-form').addEventListener('submit', function(e) {
@@ -360,17 +369,12 @@ html = """
elapsedEl.textContent = sec + 's elapsed'; elapsedEl.textContent = sec + 's elapsed';
}, 500); }, 500);
const formData = new FormData();
formData.append('link', link);
const evtSource = new EventSource('/status/' + encodeURIComponent(link)); const evtSource = new EventSource('/status/' + encodeURIComponent(link));
let currentStep = 0;
evtSource.addEventListener('step', function(e) { evtSource.addEventListener('step', function(e) {
const data = JSON.parse(e.data); const data = JSON.parse(e.data);
const idx = STEPS.findIndex(s => s.id === data.step); const idx = STEPS.findIndex(s => s.id === data.step);
if (idx >= 0) { if (idx >= 0) {
currentStep = idx;
renderSteps(idx); renderSteps(idx);
const pct = Math.min(((idx + 1) / STEPS.length) * 100, 95); const pct = Math.min(((idx + 1) / STEPS.length) * 100, 95);
progressEl.style.width = pct + '%'; progressEl.style.width = pct + '%';
@@ -507,9 +511,9 @@ def fetch_via_freedium(url, job_id=None):
def fetch_via_archive_org(url, job_id=None): def fetch_via_archive_org(url, job_id=None):
set_step(job_id, 'fallback_org') set_step(job_id, 'fallback_org')
wayback_api = f"https://archive.org/wayback/available?url={url}" wayback_api = "https://archive.org/wayback/available"
try: try:
meta = requests.get(wayback_api, timeout=15).json() meta = requests.get(wayback_api, params={"url": url}, timeout=15).json()
snapshot = meta.get("archived_snapshots", {}).get("closest", {}) snapshot = meta.get("archived_snapshots", {}).get("closest", {})
if not snapshot.get("available"): if not snapshot.get("available"):
return None, None return None, None
@@ -517,7 +521,6 @@ def fetch_via_archive_org(url, job_id=None):
if archived_url.startswith("http://"): if archived_url.startswith("http://"):
archived_url = "https://" + archived_url[len("http://"):] archived_url = "https://" + archived_url[len("http://"):]
if "/web/" in archived_url and "id_/" not in archived_url: 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 = archived_url.replace(
archived_url.split("/web/")[1].split("/")[0], archived_url.split("/web/")[1].split("/")[0],
archived_url.split("/web/")[1].split("/")[0] + "id_", archived_url.split("/web/")[1].split("/")[0] + "id_",
@@ -535,7 +538,7 @@ def fetch_via_archive_ph(url, job_id=None):
set_step(job_id, 'fallback_ph') set_step(job_id, 'fallback_ph')
for mirror in ARCHIVE_PH_MIRRORS: for mirror in ARCHIVE_PH_MIRRORS:
try: try:
newest_url = f"https://{mirror}/newest/{url}" newest_url = f"https://{mirror}/newest/{quote(url, safe=':/')}"
resp = requests.get( resp = requests.get(
newest_url, newest_url,
headers=REAL_BROWSER_HEADERS, headers=REAL_BROWSER_HEADERS,
@@ -593,33 +596,19 @@ def bypass_paywall(url, job_id=None):
if not html_text or is_challenge_page(html_text): if not html_text or is_challenge_page(html_text):
recovered = False recovered = False
if not medium: archived_html, archived_url = fetch_via_archive_org(url, job_id)
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): if archived_html and not is_challenge_page(archived_html):
html_text = archived_html html_text = archived_html
final_url = archived_url final_url = archived_url
recovered = True 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 not recovered:
if medium: if medium:
msg = ( msg = (
@@ -645,23 +634,29 @@ def bypass_paywall(url, job_id=None):
def fetch_worker(job_id, url): def fetch_worker(job_id, url):
try: try:
result = bypass_paywall(url, job_id) result = bypass_paywall(url, job_id)
jobs[job_id]['result'] = result with jobs_lock:
jobs[job_id]['step'] = 'done' jobs[job_id]['result'] = result
jobs[job_id]['step'] = 'done'
except requests.exceptions.Timeout: except requests.exceptions.Timeout:
jobs[job_id]['error'] = 'The website took too long to respond (30s timeout). Try again later.' with jobs_lock:
jobs[job_id]['step'] = 'error' 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: except requests.exceptions.ConnectionError:
jobs[job_id]['error'] = 'Could not connect to the website. Check the URL and try again.' with jobs_lock:
jobs[job_id]['step'] = 'error' 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: except requests.exceptions.RequestException as e:
jobs[job_id]['error'] = f'Failed to fetch the page: {e}' with jobs_lock:
jobs[job_id]['step'] = 'error' jobs[job_id]['error'] = f'Failed to fetch the page: {e}'
jobs[job_id]['step'] = 'error'
except RuntimeError as e: except RuntimeError as e:
jobs[job_id]['error'] = str(e) with jobs_lock:
jobs[job_id]['step'] = 'error' jobs[job_id]['error'] = str(e)
jobs[job_id]['step'] = 'error'
except Exception as e: except Exception as e:
jobs[job_id]['error'] = f'Unexpected error: {e}' with jobs_lock:
jobs[job_id]['step'] = 'error' jobs[job_id]['error'] = f'Unexpected error: {e}'
jobs[job_id]['step'] = 'error'
@app.route("/") @app.route("/")
@@ -674,7 +669,8 @@ def status_stream(url):
import json import json
job_id = str(uuid.uuid4()) job_id = str(uuid.uuid4())
jobs[job_id] = {'step': 'queued', 'result': None, 'error': None} with jobs_lock:
jobs[job_id] = {'step': 'queued', 'result': None, 'error': None}
thread = threading.Thread(target=fetch_worker, args=(job_id, url)) thread = threading.Thread(target=fetch_worker, args=(job_id, url))
thread.daemon = True thread.daemon = True
@@ -682,27 +678,30 @@ def status_stream(url):
def generate(): def generate():
last_step = None last_step = None
while True: try:
job = jobs.get(job_id) while True:
if not job: with jobs_lock:
break job = dict(jobs[job_id]) if job_id in jobs else None
if not job:
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 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) 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"
jobs.pop(job_id, None) time.sleep(0.2)
finally:
with jobs_lock:
jobs.pop(job_id, None)
return Response(generate(), mimetype='text/event-stream', return Response(generate(), mimetype='text/event-stream',
headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}) headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'})