210 lines
8.7 KiB
Python
210 lines
8.7 KiB
Python
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import URLError, HTTPError
|
|
import re
|
|
from html import unescape
|
|
from urllib.parse import urlparse, parse_qs, unquote
|
|
from html.parser import HTMLParser
|
|
|
|
TARGET_URL = "https://m.slrclub.com/l/free"
|
|
|
|
class ProxyHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
# parse query params for skip/limit and optional path override
|
|
parsed = urlparse(self.path)
|
|
qs = parse_qs(parsed.query)
|
|
try:
|
|
skip = int(qs.get('skip', ['0'])[0])
|
|
except Exception:
|
|
skip = 0
|
|
try:
|
|
limit = int(qs.get('limit', ['10'])[0])
|
|
except Exception:
|
|
limit = 10
|
|
|
|
# Path to fetch on the remote site. If `path` query provided, use it; otherwise use request path or default.
|
|
requested_path = qs.get('path', [unquote(parsed.path or '')])[0]
|
|
if requested_path in ('', '/', None):
|
|
requested_path = '/l/free'
|
|
|
|
target_url = f"https://m.slrclub.com{requested_path}"
|
|
|
|
try:
|
|
req = Request(
|
|
target_url,
|
|
headers={
|
|
"User-Agent": "Mozilla/5.0 (compatible; Python crawler/1.0)",
|
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
|
},
|
|
)
|
|
with urlopen(req, timeout=15) as response:
|
|
content_type = response.headers.get_content_type()
|
|
charset = response.headers.get_content_charset(failobj="utf-8")
|
|
raw_html = response.read().decode(charset, errors="replace")
|
|
|
|
# Extract links using HTMLParser to better preserve anchor text, titles, and img alts
|
|
class LinkExtractor(HTMLParser):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.stack = []
|
|
self.anchors = []
|
|
self.current = None
|
|
|
|
def handle_starttag(self, tag, attrs):
|
|
attrs = dict(attrs)
|
|
self.stack.append((tag, attrs))
|
|
if tag == 'a':
|
|
self.current = {
|
|
'href': attrs.get('href', '').strip(),
|
|
'title': attrs.get('title', '').strip(),
|
|
'text': '',
|
|
'ancestors': [t for t, a in self.stack[:-1]],
|
|
}
|
|
if tag == 'img' and self.current is not None:
|
|
alt = attrs.get('alt') or attrs.get('title') or ''
|
|
if alt:
|
|
self.current.setdefault('img_alt', alt.strip())
|
|
|
|
def handle_data(self, data):
|
|
if self.current is not None:
|
|
self.current['text'] += data
|
|
|
|
def handle_endtag(self, tag):
|
|
if not self.stack:
|
|
return
|
|
top_tag, top_attrs = self.stack.pop()
|
|
if tag == 'a' and self.current is not None:
|
|
href = (self.current.get('href') or '').strip()
|
|
text = (self.current.get('text') or '').strip()
|
|
title = (self.current.get('title') or '').strip()
|
|
img_alt = (self.current.get('img_alt') or '').strip()
|
|
final = text or title or img_alt
|
|
ancestors = self.current.get('ancestors', [])
|
|
self.anchors.append({'href': href, 'text': unescape(final), 'ancestors': ancestors})
|
|
self.current = None
|
|
|
|
parser = LinkExtractor()
|
|
parser.feed(raw_html)
|
|
|
|
# Prepare li blocks to help extract nearby text when anchor text is empty
|
|
li_blocks = re.findall(r"<li[^>]*>(.*?)</li>", raw_html, flags=re.DOTALL | re.IGNORECASE)
|
|
|
|
items = []
|
|
for a in parser.anchors:
|
|
href = a['href']
|
|
txt = (a['text'] or '').strip()
|
|
if not href:
|
|
continue
|
|
if href.lower().startswith('javascript:'):
|
|
continue
|
|
# skip anchors inside nav/header/footer elements
|
|
anc = ' '.join(a.get('ancestors', [])).lower()
|
|
if any(k in anc for k in ('nav', 'header', 'footer', 'menu', 'notice', 'gnb', 'util')):
|
|
continue
|
|
|
|
# If anchor text is empty, try to find text inside the same <li> block
|
|
if not txt:
|
|
for block in li_blocks:
|
|
if href in block:
|
|
cand = re.sub(r"<[^>]+>", "", block).strip()
|
|
# pick the first non-empty line as title
|
|
for line in cand.splitlines():
|
|
line = line.strip()
|
|
if line:
|
|
txt = unescape(line)
|
|
break
|
|
if txt:
|
|
break
|
|
|
|
# still empty? skip
|
|
if not txt:
|
|
continue
|
|
|
|
items.append((href, txt))
|
|
|
|
# Deduplicate while preserving order
|
|
seen = set()
|
|
filtered = []
|
|
for href, txt in items:
|
|
key = (href, txt)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
# ignore trivial/empty titles
|
|
if not txt:
|
|
continue
|
|
if href in ('#', '/', '', '/l/free'):
|
|
continue
|
|
filtered.append((href, txt))
|
|
|
|
# Apply skip/limit to all filtered items (including notices)
|
|
posts = filtered[skip:skip+limit]
|
|
|
|
# Build simple HTML list for posts only
|
|
host = self.headers.get("Host", f"localhost:{self.server.server_port}")
|
|
out_lines = [
|
|
'<!doctype html>',
|
|
'<html><head><meta charset="utf-8"><title>SLRClub - 게시글 목록</title></head><body>',
|
|
f'<h1>게시글 (skip={skip}, limit={limit})</h1>',
|
|
'<ul>'
|
|
]
|
|
for href, txt in posts:
|
|
# Build absolute URL to original site so clicks open the real page
|
|
abs_url = href
|
|
if href.startswith("/"):
|
|
abs_url = "https://m.slrclub.com" + href
|
|
elif href.startswith("//"):
|
|
abs_url = "https:" + href
|
|
elif not href.startswith("http"):
|
|
# relative path without leading slash
|
|
abs_url = "https://m.slrclub.com/" + href.lstrip("/")
|
|
|
|
# Escape minimal characters in title
|
|
safe_txt = txt.replace("<", "<").replace(">", ">")
|
|
out_lines.append(f'<li><a href="{abs_url}" target="_blank" rel="noopener noreferrer">{safe_txt}</a></li>')
|
|
|
|
out_lines.extend(["</ul>", "</body></html>"])
|
|
out = "\n".join(out_lines)
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(out.encode("utf-8", errors="replace"))
|
|
|
|
except HTTPError as e:
|
|
self.send_response(e.code)
|
|
self.send_header("Content-Type", "text/plain; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(f"HTTP Error: {e.code} {e.reason}".encode("utf-8"))
|
|
except URLError as e:
|
|
self.send_response(502)
|
|
self.send_header("Content-Type", "text/plain; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(f"Failed to fetch target page: {e.reason}".encode("utf-8"))
|
|
except Exception as e:
|
|
self.send_response(500)
|
|
self.send_header("Content-Type", "text/plain; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(f"Server error: {e}".encode("utf-8"))
|
|
|
|
|
|
def run(server_class=HTTPServer, handler_class=ProxyHandler, port=8000):
|
|
server_address = ("", port)
|
|
try:
|
|
httpd = server_class(server_address, handler_class)
|
|
except OSError as exc:
|
|
if exc.errno == 98:
|
|
raise SystemExit(f"Port {port} is already in use. Run with a different port using --port.") from exc
|
|
raise
|
|
print(f"Serving crawler homepage at http://localhost:{port}/")
|
|
httpd.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Proxy crawler for https://m.slrclub.com/l/free")
|
|
parser.add_argument("--port", type=int, default=8000, help="Port to listen on")
|
|
args = parser.parse_args()
|
|
run(port=args.port)
|
|
|