#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ hguaw_ios.py · 黄瓜网采集工具(iOS 版) ========================================= 纯标准库实现,零第三方依赖 —— iPhone / iPad 上直接能跑。 适配环境:Pythonista 3 / Pyto / a-Shell / iSH / 桌面 Python 3.8+ 功能: 1) 采集 文章正文 / 标题 / 分类 / 作者 / 日期 / 标签 / 封面 2) 下图 封面 + 正文配图,sha1 去重,断点续传 3) 处理 入库去重 / 导出 json·csv·md / 全文搜索 / 统计 / 清理 4) 定时 前台轮询监控,有新文自动抓 + 推送 TG/Webhook 零依赖说明:没用 requests,没用 bs4 —— 内置 urllib + 手写 HTML 解析器。 iOS 上的 Python 环境装不了第三方包,所以这版必须自给自足。 常用命令: python3 hguaw_ios.py info 环境自检 python3 hguaw_ios.py sync 刷新文章索引 python3 hguaw_ios.py crawl -n 20 --with-img 抓最新 20 篇 + 下图 python3 hguaw_ios.py crawl --since 2026-10-01 抓某日期之后的 python3 hguaw_ios.py imgs 给缺图的文章补图 python3 hguaw_ios.py export --fmt md 导出 markdown python3 hguaw_ios.py stats 看进度 python3 hguaw_ios.py search 关键词 全文搜索 python3 hguaw_ios.py clean --dry-run 查重/清孤儿 python3 hguaw_ios.py watch --interval 600 每 10 分钟监控一次 """ import argparse import concurrent.futures import csv import gzip import hashlib import json import os import re import shutil import sqlite3 import ssl import sys import threading import time import urllib.error import urllib.parse import urllib.request import zlib from html.parser import HTMLParser VERSION = "1.0.0" BASE = "https://www.hguaw.com" DEFAULT_REFERER = BASE + "/" DEFAULT_UA = ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/124.0 Safari/537.36" ) UA_IOS = ( "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) " "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1" ) BLOCK_TAGS = { "p", "div", "br", "li", "tr", "td", "th", "h1", "h2", "h3", "h4", "h5", "h6", "section", "article", "header", "footer", "blockquote", "pre", "figure", "figcaption", "ul", "ol", "table", "hr", "dd", "dt", "dl", } SKIP_TAGS = {"script", "style", "noscript", "svg", "iframe"} VOID_TAGS = { "area", "base", "br", "col", "embed", "hr", "img", "input", "link", "meta", "param", "source", "track", "wbr", } IMG_ATTRS = ("data-src", "data-original", "data-lazy-src", "data-echo", "src") IMG_BAD = ("yandex", ".svg", "/static/brand/", "placeholder", "avatar", "logo", "1x1", "spacer") # ───────────────────────────────────────────────────────────── # 输出小工具 # ───────────────────────────────────────────────────────────── def out(msg="", **kw): try: print(msg, flush=True, **kw) except UnicodeEncodeError: print(msg.encode("utf-8", "replace").decode("utf-8", "replace"), flush=True) except BrokenPipeError: try: sys.stdout.close() except Exception: pass def hr(title=""): if title: out("\n" + "─" * 4 + " " + title + " " + "─" * max(2, 46 - len(title))) else: out("─" * 52) def human_size(n): n = float(n or 0) for unit in ("B", "KB", "MB", "GB"): if n < 1024 or unit == "GB": return ("%.0f %s" % (n, unit)) if unit == "B" else ("%.1f %s" % (n, unit)) n /= 1024.0 def now_str(): return time.strftime("%Y-%m-%d %H:%M:%S") def today(): return time.strftime("%Y-%m-%d") def sha1(s): if isinstance(s, str): s = s.encode("utf-8", "replace") return hashlib.sha1(s).hexdigest() # ───────────────────────────────────────────────────────────── # iOS 环境识别 & 数据目录 # ───────────────────────────────────────────────────────────── def ios_kind(): """返回当前运行环境:pythonista / pyto / ash / ish / desktop""" if "pythonista" in sys.modules: return "pythonista" if "pyto" in sys.modules: return "pyto" try: if "pythonista" in sys.executable.lower(): return "pythonista" except Exception: pass plat = sys.platform if plat == "ios": shell = os.environ.get("SHELL", "") return "ash" if "ash" in shell.lower() else "ish" return "desktop" IS_IOS = ios_kind() != "desktop" def default_datadir(): """ 找一个能写、且在 iOS 上不会被清掉的目录。 优先 iOS 文档目录(能在「文件」App 里看到),退回脚本同级目录。 """ home = os.path.expanduser("~") cands = [] kind = ios_kind() if kind in ("pythonista", "pyto"): cands.append(os.path.join(home, "Documents", "hguaw_data")) cands.append(os.path.join(home, "hguaw_data")) else: cands.append(os.path.join(home, "Documents", "hguaw_data")) try: cands.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), "hguaw_data")) except NameError: pass cands.append(os.path.join(os.getcwd(), "hguaw_data")) for c in cands: parent = os.path.dirname(c) or "." if not os.path.isdir(parent): continue if not os.access(parent, os.W_OK): continue try: os.makedirs(c, exist_ok=True) probe = os.path.join(c, ".w_test") with open(probe, "w") as f: f.write("1") os.remove(probe) return c except Exception: continue # 最后兜底:当前目录硬建 c = os.path.join(os.getcwd(), "hguaw_data") os.makedirs(c, exist_ok=True) return c # ───────────────────────────────────────────────────────────── # HTTP # ───────────────────────────────────────────────────────────── _SSL_CTX_LAX = None def ssl_ctx(insecure=False): global _SSL_CTX_LAX if insecure: if _SSL_CTX_LAX is None: _SSL_CTX_LAX = ssl.create_default_context() _SSL_CTX_LAX.check_hostname = False _SSL_CTX_LAX.verify_mode = ssl.CERT_NONE return _SSL_CTX_LAX return None class Net: def __init__(self, ua=DEFAULT_UA, referer=DEFAULT_REFERER, timeout=30, retries=2, insecure=False, delay=0.0, quiet=False): self.ua = ua self.referer = referer self.timeout = timeout self.retries = retries self.insecure = insecure self.delay = delay self.quiet = quiet self._last = 0.0 def _headers(self, referer=None, extra=None): h = { "User-Agent": self.ua, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", } ref = referer or self.referer if ref: h["Referer"] = ref if extra: h.update(extra) return h def _throttle(self): if self.delay > 0: gap = time.time() - self._last if gap < self.delay: time.sleep(self.delay - gap) self._last = time.time() def get(self, url, referer=None, timeout=None, binary=False): """取回 bytes(失败重试)。""" last = None for i in range(self.retries + 1): try: self._throttle() req = urllib.request.Request(url, headers=self._headers(referer)) with urllib.request.urlopen(req, timeout=timeout or self.timeout, context=ssl_ctx(self.insecure)) as r: data = r.read() enc = (r.headers.get("Content-Encoding") or "").lower() if enc == "gzip": data = gzip.decompress(data) elif enc == "deflate": try: data = zlib.decompress(data) except zlib.error: data = zlib.decompress(data, -zlib.MAX_WBITS) return data except Exception as e: last = e if i < self.retries: time.sleep(1.2 * (i + 1)) raise last if last else RuntimeError("request failed") def text(self, url, referer=None, timeout=None): return decode_html(self.get(url, referer=referer, timeout=timeout)) def download(self, url, dest, referer=None, timeout=None): """流式下载到 dest,返回 (字节数, content_type)。""" last = None os.makedirs(os.path.dirname(dest), exist_ok=True) tmp = dest + ".part" for i in range(self.retries + 1): try: self._throttle() req = urllib.request.Request(url, headers=self._headers(referer)) total = 0 with urllib.request.urlopen(req, timeout=timeout or self.timeout, context=ssl_ctx(self.insecure)) as r: ctype = (r.headers.get("Content-Type") or "").split(";")[0].strip() with open(tmp, "wb") as f: while True: chunk = r.read(65536) if not chunk: break f.write(chunk) total += len(chunk) os.replace(tmp, dest) return total, ctype except Exception as e: last = e if os.path.exists(tmp): try: os.remove(tmp) except Exception: pass if i < self.retries: time.sleep(1.2 * (i + 1)) raise last if last else RuntimeError("download failed") def decode_html(data): if isinstance(data, str): return data for enc in ("utf-8", "gb18030"): try: return data.decode(enc) except (UnicodeDecodeError, LookupError): continue # 从 meta 里找 charset head = data[:2048].decode("latin-1", "replace") m = re.search(r'charset=["\']?([\w-]+)', head, re.I) if m: try: return data.decode(m.group(1)) except Exception: pass return data.decode("utf-8", "replace") # ───────────────────────────────────────────────────────────── # 迷你 HTML 解析器(不依赖 bs4) # ───────────────────────────────────────────────────────────── class Node: __slots__ = ("tag", "attrs", "children", "parent", "tag_start", "tag_end", "end_start") def __init__(self, tag, attrs=None, parent=None, tag_start=0, tag_end=0): self.tag = tag self.attrs = attrs or {} self.children = [] self.parent = parent self.tag_start = tag_start self.tag_end = tag_end self.end_start = None def cls(self): return (self.attrs.get("class") or "").split() def has_cls(self, name): return name in self.cls() def get(self, key, default=""): return self.attrs.get(key, default) def __repr__(self): c = " ".join(self.cls()) return "<%s%s>" % (self.tag, ("." + c) if c else "") class DomBuilder(HTMLParser): def __init__(self, raw): HTMLParser.__init__(self, convert_charrefs=True) self.raw = raw self.line_starts = [0] pos = raw.find("\n") while pos != -1: self.line_starts.append(pos + 1) pos = raw.find("\n", pos + 1) self.root = Node("#root") self.stack = [self.root] def _off(self): try: line, col = self.getpos() except Exception: return len(self.raw) if 1 <= line <= len(self.line_starts): return min(self.line_starts[line - 1] + col, len(self.raw)) return len(self.raw) def _append(self, node): node.parent = self.stack[-1] self.stack[-1].children.append(node) def handle_starttag(self, tag, attrs): start = self._off() gt = self.raw.find(">", start) end = gt + 1 if gt != -1 else start node = Node(tag, dict(attrs), tag_start=start, tag_end=end) self._append(node) if tag not in VOID_TAGS: self.stack.append(node) def handle_startendtag(self, tag, attrs): start = self._off() gt = self.raw.find(">", start) end = gt + 1 if gt != -1 else start self._append(Node(tag, dict(attrs), tag_start=start, tag_end=end)) def handle_endtag(self, tag): off = self._off() for i in range(len(self.stack) - 1, 0, -1): if self.stack[i].tag == tag: self.stack[i].end_start = off del self.stack[i:] return def handle_data(self, data): if data and data.strip(): self.stack[-1].children.append(data) def close(self): HTMLParser.close(self) for n in self.stack[1:]: if n.end_start is None: n.end_start = len(self.raw) class Doc: """解析后的文档,提供查询接口。""" def __init__(self, raw): self.raw = raw b = DomBuilder(raw) b.feed(raw) b.close() self.root = b.root def iter(self, node=None): node = node or self.root for ch in node.children: if isinstance(ch, Node): yield ch for sub in self.iter(ch): yield sub def find_all(self, tag=None, cls=None, node=None): res = [] for n in self.iter(node): if tag and n.tag != tag: continue if cls and not n.has_cls(cls): continue res.append(n) return res def first(self, tag=None, cls=None, node=None): r = self.find_all(tag, cls, node) return r[0] if r else None def inner_html(self, node): if node is None: return "" end = node.end_start if node.end_start is not None else len(self.raw) return self.raw[node.tag_end:end] def text(self, node, strip_skip=True): if node is None: return "" buf = [] def walk(n, depth): for ch in n.children: if isinstance(ch, str): buf.append(ch) continue if strip_skip and ch.tag in SKIP_TAGS: continue if ch.tag in BLOCK_TAGS: buf.append("\n") walk(ch, depth + 1) if ch.tag in BLOCK_TAGS: buf.append("\n") walk(node, 0) return clean_text("".join(buf)) def clean_text(t): t = t.replace("\u00a0", " ").replace("\u200b", "") lines = [] for raw_line in t.split("\n"): line = re.sub(r"[ \t\r\f\v]+", " ", raw_line).strip() if not line: if lines and lines[-1] != "": lines.append("") continue lines.append(line) while lines and lines[-1] == "": lines.pop() return "\n".join(lines).strip() # ───────────────────────────────────────────────────────────── # 文章解析 # ───────────────────────────────────────────────────────────── def abs_url(u, base=BASE): if not u: return "" u = u.strip() if u.startswith("//"): return "https:" + u if u.startswith("/"): return base + u return u def is_content_image(u): if not u: return False low = u.lower() if low.startswith("data:"): return False for bad in IMG_BAD: if bad in low: return False return low.startswith("http") or low.startswith("//") or low.startswith("/") def parse_article(raw, url): doc = Doc(raw) r = doc.root # 标题 title = "" h1 = doc.first("h1", "post-title") if h1 is None: h1 = doc.first("h1") if h1 is not None: title = doc.text(h1) if not title: m = re.search(r']+property=["\']og:title["\'][^>]+content=["\']([^"\']+)', raw) if m: title = m.group(1).strip() if not title: m = re.search(r"(.*?)", raw, re.S) if m: title = re.sub(r"\s*[-|｜]\s*黄瓜网.*$", "", m.group(1)).strip() # 封面 cover = "" for m in re.finditer(r']+(?:property|itemprop)=["\'](?:og:image|image)["\'][^>]*>', raw): mm = re.search(r'content=["\']([^"\']+)', m.group(0)) if mm: cover = abs_url(mm.group(1)) break if not cover: img = doc.first("img", "post-cover") if img is not None: cover = abs_url(img.get("src") or img.get("data-src")) # 日期 pubdate = "" t = None for cand in doc.find_all("time"): if cand.get("itemprop") == "datePublished": t = cand break if t is None: for cand in doc.find_all("time"): if cand.get("datetime"): t = cand break if t is not None and t.get("datetime"): pubdate = t.get("datetime") if not pubdate: m = re.search(r'"datePublished"\s*:\s*"([^"]+)"', raw) if m: pubdate = m.group(1) # 作者 author = author_url = "" pa = doc.first("span", "post-author") if pa is not None: author = doc.text(pa) a = doc.first("a", None, pa) if a is not None: author_url = abs_url(a.get("href")) # 分类 —— 取面包屑里最后一个 /category/ 链接(不是导航里那个) category = cat_url = "" for nav in doc.find_all("nav"): if "首页" not in doc.text(nav): continue cats = [x for x in doc.find_all("a", node=nav) if "/category/" in (x.get("href") or "")] if cats: category = doc.text(cats[-1]) cat_url = abs_url(cats[-1].get("href")) break if not category: m = re.search(r'"articleSection"\s*:\s*"([^"]+)"', raw) if m: category = m.group(1) # 标签 tags = [] sec = doc.first("section", "post-tags") if sec is not None: tags = [doc.text(a) for a in doc.find_all("a", None, sec)] tags = [x for x in tags if x] # 正文容器 content_node = doc.first("div", "post-content") content_html = doc.inner_html(content_node) if content_node is not None else "" content_text = doc.text(content_node) if content_node is not None else "" # 正文配图 images = [] if content_node is not None: for img in doc.find_all("img", node=content_node): src = "" for k in IMG_ATTRS: if img.get(k): src = img.get(k) break src = abs_url(src) if is_content_image(src) and src not in images: images.append(src) if cover and cover not in images: images.insert(0, cover) aid = "" m = re.search(r"/archives/(\d+)", url) if m: aid = m.group(1) return { "url": url, "aid": aid, "title": clean_text(title), "category": category, "cat_url": cat_url, "author": author, "author_url": author_url, "pubdate": pubdate, "tags": tags, "cover": cover, "content_text": content_text, "content_html": content_html, "content_len": len(content_text), "images": images, } # ───────────────────────────────────────────────────────────── # 存储 # ───────────────────────────────────────────────────────────── SCHEMA = """ CREATE TABLE IF NOT EXISTS posts ( url TEXT PRIMARY KEY, aid TEXT, title TEXT, category TEXT, cat_url TEXT, author TEXT, author_url TEXT, pubdate TEXT, tags TEXT, cover TEXT, content_text TEXT, content_html TEXT, content_len INTEGER DEFAULT 0, img_total INTEGER DEFAULT 0, img_ok INTEGER DEFAULT 0, fetched_at TEXT, updated_at TEXT ); CREATE TABLE IF NOT EXISTS images ( post_url TEXT, url TEXT, sha1 TEXT, path TEXT, size INTEGER DEFAULT 0, ok INTEGER DEFAULT 0, err TEXT, created_at TEXT, PRIMARY KEY (post_url, url) ); CREATE TABLE IF NOT EXISTS idx ( url TEXT PRIMARY KEY, lastmod TEXT, fetched INTEGER DEFAULT 0, added_at TEXT, fetch_at TEXT ); CREATE INDEX IF NOT EXISTS idx_pubdate ON posts(pubdate DESC); CREATE INDEX IF NOT EXISTS idx_category ON posts(category); CREATE INDEX IF NOT EXISTS idx_idx_fetched ON idx(fetched, lastmod DESC); """ class Store: def __init__(self, datadir): self.dir = datadir self.img_dir = os.path.join(datadir, "images") self.md_dir = os.path.join(datadir, "md") self.db_path = os.path.join(datadir, "hguaw.db") os.makedirs(self.img_dir, exist_ok=True) self.conn = sqlite3.connect(self.db_path, timeout=30, check_same_thread=False) self.conn.row_factory = sqlite3.Row self._lock = threading.RLock() with self._lock: self.conn.executescript(SCHEMA) self.conn.commit() def close(self): try: self.conn.commit() self.conn.close() except Exception: pass # ---- 索引 ---- def add_index(self, rows): """rows: [(url, lastmod), ...] 返回新增数""" cur = self.conn.cursor() added = 0 ts = now_str() with self._lock: for url, lm in rows: cur.execute("INSERT OR IGNORE INTO idx(url,lastmod,added_at) VALUES(?,?,?)", (url, lm or "", ts)) if cur.rowcount: added += 1 self.conn.commit() return added def index_total(self): return self.conn.execute("SELECT COUNT(*) FROM idx").fetchone()[0] def pending(self, limit=0, since="", force=False, order="new"): sql = "SELECT url,lastmod FROM idx WHERE 1=1" args = [] if not force: sql += " AND fetched=0" if since: sql += " AND lastmod>=?" args.append(since) sql += " ORDER BY lastmod DESC, url DESC" if order == "new" else " ORDER BY lastmod ASC, url ASC" if limit and limit > 0: sql += " LIMIT ?" args.append(limit) return [(r["url"], r["lastmod"]) for r in self.conn.execute(sql, args)] def mark_fetched(self, url, ok=True): with self._lock: self.conn.execute("UPDATE idx SET fetched=?, fetch_at=? WHERE url=?", (1 if ok else -1, now_str(), url)) self.conn.commit() # ---- 文章 ---- def save_post(self, d): ts = now_str() sql = """INSERT INTO posts(url,aid,title,category,cat_url,author,author_url,pubdate, tags,cover,content_text,content_html,content_len, img_total,img_ok,fetched_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(url) DO UPDATE SET aid=excluded.aid, title=excluded.title, category=excluded.category, cat_url=excluded.cat_url, author=excluded.author, author_url=excluded.author_url, pubdate=excluded.pubdate, tags=excluded.tags, cover=excluded.cover, content_text=excluded.content_text, content_html=excluded.content_html, content_len=excluded.content_len, img_total=excluded.img_total, updated_at=excluded.updated_at""" args = (d["url"], d["aid"], d["title"], d["category"], d.get("cat_url", ""), d["author"], d.get("author_url", ""), d["pubdate"], ", ".join(d["tags"]), d["cover"], d["content_text"], d["content_html"], d["content_len"], len(d["images"]), self.count_ok_images(d["url"]), ts, ts) with self._lock: self.conn.execute(sql, args) self.conn.commit() def count_ok_images(self, post_url): return self.conn.execute( "SELECT COUNT(*) FROM images WHERE post_url=? AND ok=1", (post_url,)).fetchone()[0] def post_exists(self, url): return self.conn.execute("SELECT 1 FROM posts WHERE url=?", (url,)).fetchone() is not None def get_post(self, url): return self.conn.execute("SELECT * FROM posts WHERE url=?", (url,)).fetchone() def posts_missing_images(self, limit=0): sql = """SELECT p.url FROM posts p WHERE p.img_total > IFNULL((SELECT COUNT(*) FROM images i WHERE i.post_url=p.url AND i.ok=1),0) ORDER BY p.pubdate DESC""" if limit: sql += " LIMIT %d" % int(limit) return [r[0] for r in self.conn.execute(sql)] def all_posts(self, limit=0): sql = "SELECT * FROM posts ORDER BY pubdate DESC" if limit: sql += " LIMIT %d" % int(limit) return list(self.conn.execute(sql)) # ---- 图片 ---- def image_known(self, post_url, img_url): r = self.conn.execute("SELECT ok FROM images WHERE post_url=? AND url=?", (post_url, img_url)).fetchone() return bool(r and r["ok"] == 1) def save_image(self, post_url, img_url, path, size, ok, err=""): with self._lock: self.conn.execute( """INSERT INTO images(post_url,url,sha1,path,size,ok,err,created_at) VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(post_url,url) DO UPDATE SET sha1=excluded.sha1, path=excluded.path, size=excluded.size, ok=excluded.ok, err=excluded.err, created_at=excluded.created_at""", (post_url, img_url, sha1(img_url), path, size, 1 if ok else 0, err, now_str()), ) self.conn.execute("UPDATE posts SET img_ok=? WHERE url=?", (self.count_ok_images(post_url), post_url)) self.conn.commit() def images_of(self, post_url): return list(self.conn.execute("SELECT * FROM images WHERE post_url=? ORDER BY url", (post_url,))) def image_path_for(self, img_url, ext=".jpg"): h = sha1(img_url) sub = os.path.join(self.img_dir, h[:2], h[2:4]) os.makedirs(sub, exist_ok=True) return os.path.join(sub, h + ext) def existing_image(self, img_url): """按 sha1 找已下好的文件(不管扩展名)。""" h = sha1(img_url) sub = os.path.join(self.img_dir, h[:2], h[2:4]) if not os.path.isdir(sub): return None try: for fn in os.listdir(sub): p = os.path.join(sub, fn) if fn.startswith(h) and not fn.endswith((".part", ".tmp_dl")) \ and os.path.getsize(p) > 0: return p except Exception: return None return None def ext_of(url, ctype=""): ctype = (ctype or "").lower() if "webp" in ctype: return ".webp" if "png" in ctype: return ".png" if "gif" in ctype: return ".gif" if "jpeg" in ctype or "jpg" in ctype: return ".jpg" path = urllib.parse.urlparse(url).path ext = os.path.splitext(path)[1].lower() if ext in (".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".avif"): return ext return ".jpg" # ───────────────────────────────────────────────────────────── # 索引同步(走 sitemap) # ───────────────────────────────────────────────────────────── def sync_index(net, store, quiet=False): idx_xml = net.text(BASE + "/sitemap.xml") shards = re.findall(r"\s*([^<\s]*sitemap-content[^<\s]*)\s*", idx_xml) if not shards: shards = [BASE + "/sitemap.xml"] if not quiet: out(" 发现 %d 个内容分片" % len(shards)) rows = [] for s in shards: try: xml = net.text(s) except Exception as e: if not quiet: out(" ! 分片拉取失败 %s (%s)" % (s.rsplit("/", 1)[-1], e)) continue for m in re.finditer(r"\s*\s*([^<\s]+)\s*\s*(?:\s*([^<\s]*)\s*)?", xml): u, lm = m.group(1), m.group(2) or "" if "/archives/" in u: rows.append((u, lm)) added = store.add_index(rows) if not quiet: out(" 索引:共 %d 条,新增 %d 条" % (store.index_total(), added)) return added def ensure_index(net, store, refresh=False, quiet=False): if refresh or store.index_total() == 0: sync_index(net, store, quiet=quiet) # ───────────────────────────────────────────────────────────── # 采集 & 下图 # ───────────────────────────────────────────────────────────── def crawl_one(net, store, url, with_img=False, img_limit=0, threads=4, img_timeout=None, quiet=False): """抓一篇。返回 (ok, 消息, 文章dict或None)""" try: raw = net.text(url) except Exception as e: store.mark_fetched(url, ok=False) return False, "抓取失败: %s" % e, None try: d = parse_article(raw, url) except Exception as e: store.mark_fetched(url, ok=False) return False, "解析失败: %s" % e, None if not d["title"] and not d["content_text"]: store.mark_fetched(url, ok=False) return False, "空页面(可能被拦)", None store.save_post(d) store.mark_fetched(url, ok=True) if with_img and d["images"]: ok_n, fail_n = download_images(net, store, url, d["images"], limit=img_limit, threads=threads, timeout=img_timeout, quiet=quiet) store.save_post(d) # 刷新 img_ok msg = "OK %s [%d图 成%d/败%d]" % (d["title"][:34], len(d["images"]), ok_n, fail_n) else: msg = "OK %s [%d图]" % (d["title"][:34], len(d["images"])) return True, msg, d def download_images(net, store, post_url, img_urls, limit=0, threads=4, timeout=None, quiet=False): urls = img_urls[:limit] if limit and limit > 0 else list(img_urls) urls = [u for u in urls if not store.image_known(post_url, u)] if not urls: return 0, 0 ok_n = [0] fail_n = [0] lock = None def worker(u): try: had = store.existing_image(u) if had: store.save_image(post_url, u, had, os.path.getsize(had), True) return True base = store.image_path_for(u) tmp = base + ".tmp" n, ctype = net.download(u, tmp, referer=DEFAULT_REFERER, timeout=timeout) final = store.image_path_for(u, ext_of(u, ctype)) if os.path.exists(final): os.remove(tmp) else: os.replace(tmp, final) store.save_image(post_url, u, final, os.path.getsize(final), True) return True except Exception as e: store.save_image(post_url, u, "", 0, False, str(e)[:180]) if not quiet: out(" ! 图失败 %s (%s)" % (u[-42:], str(e)[:60])) return False if threads and threads > 1: with concurrent.futures.ThreadPoolExecutor(max_workers=threads) as ex: for r in ex.map(worker, urls): if r: ok_n[0] += 1 else: fail_n[0] += 1 else: for u in urls: if worker(u): ok_n[0] += 1 else: fail_n[0] += 1 return ok_n[0], fail_n[0] def urls_from_ids(ids_text): res = [] for piece in re.split(r"[,\s]+", ids_text or ""): piece = piece.strip() if not piece: continue if piece.startswith("http"): res.append(piece) elif piece.isdigit(): res.append("%s/archives/%s/" % (BASE, piece)) else: m = re.search(r"/archives/(\d+)", piece) if m: res.append("%s/archives/%s/" % (BASE, m.group(1))) return res # ───────────────────────────────────────────────────────────── # 导出 # ───────────────────────────────────────────────────────────── def export_json(store, path, limit=0): rows = store.all_posts(limit) data = [] for r in rows: d = dict(r) d["tags"] = [t for t in (r["tags"] or "").split(", ") if t] d["images"] = [dict(i) for i in store.images_of(r["url"])] data.append(d) with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) return len(data) def export_csv(store, path, limit=0): rows = store.all_posts(limit) cols = ["aid", "url", "title", "category", "author", "pubdate", "tags", "img_total", "img_ok", "content_len", "cover"] with open(path, "w", encoding="utf-8-sig", newline="") as f: w = csv.DictWriter(f, fieldnames=cols, extrasaction="ignore") w.writeheader() for r in rows: w.writerow({k: r[k] for k in cols}) return len(rows) def md_for_post(store, r): tags = [t for t in (r["tags"] or "").split(", ") if t] imgs = store.images_of(r["url"]) lines = ["# " + (r["title"] or "(无标题)"), ""] lines.append("- 链接: " + r["url"]) lines.append("- 分类: %s 作者: %s 日期: %s" % (r["category"] or "-", r["author"] or "-", (r["pubdate"] or "-")[:10])) if tags: lines.append("- 标签: " + "、".join(tags)) lines.append("") lines.append("---") lines.append("") if r["content_text"]: lines.append(r["content_text"]) lines.append("") if imgs: lines.append("## 图片 (%d)" % len(imgs)) lines.append("") for i in imgs: lines.append("- [%s](%s)" % ("已下载" if i["ok"] else "未下载", i["url"])) return "\n".join(lines) def export_md(store, outdir, limit=0, one_file=""): rows = store.all_posts(limit) if one_file: parts = [] for r in rows: parts.append(md_for_post(store, r)) parts.append("\n\n" + "=" * 60 + "\n") with open(one_file, "w", encoding="utf-8") as f: f.write("\n".join(parts)) return len(rows), one_file os.makedirs(outdir, exist_ok=True) n = 0 for r in rows: name = (r["aid"] or sha1(r["url"])[:10]) title = re.sub(r'[\\/:*?"<>|\n\r\t]+', "_", (r["title"] or "")[:40]).strip("_ ") fn = os.path.join(outdir, "%s_%s.md" % (name, title)) with open(fn, "w", encoding="utf-8") as f: f.write(md_for_post(store, r)) n += 1 return n, outdir # ───────────────────────────────────────────────────────────── # 通知 # ───────────────────────────────────────────────────────────── def ios_notify(title, msg): """iOS 本地通知:Pyto 和 Pythonista 各一套 API,谁在就用谁。""" try: import notification # Pyto notification.schedule(msg, title=title) return True except Exception: pass try: import notifications # Pythonista notifications.schedule(msg, title=title) return True except Exception: pass return False def tg_send(token, chat, text): if not token or not chat: return False, "未配置 tg_token / tg_chat" try: body = urllib.parse.urlencode({ "chat_id": chat, "text": text, "disable_web_page_preview": "true" }).encode() req = urllib.request.Request("https://api.telegram.org/bot%s/sendMessage" % token, data=body, headers={"Content-Type": "application/x-www-form-urlencoded"}) with urllib.request.urlopen(req, timeout=20) as r: j = json.loads(r.read().decode("utf-8", "replace")) return (True, "ok") if j.get("ok") else (False, str(j)[:200]) except Exception as e: return False, str(e)[:200] def webhook_send(url, payload): if not url: return False, "未配置 webhook" try: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") req = urllib.request.Request(url, data=body, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=20) as r: return True, str(r.status) except Exception as e: return False, str(e)[:200] # ───────────────────────────────────────────────────────────── # 配置 # ───────────────────────────────────────────────────────────── def cfg_path(datadir): return os.path.join(datadir, "hguaw_config.json") def load_cfg(datadir): p = cfg_path(datadir) if os.path.exists(p): try: with open(p, "r", encoding="utf-8") as f: return json.load(f) except Exception: return {} return {} def save_cfg(datadir, cfg): with open(cfg_path(datadir), "w", encoding="utf-8") as f: json.dump(cfg, f, ensure_ascii=False, indent=2) # ───────────────────────────────────────────────────────────── # 命令实现 # ───────────────────────────────────────────────────────────── def cmd_info(a, store, net, cfg): hr("环境自检") out("脚本版本 : hguaw_ios.py %s" % VERSION) out("运行环境 : %s%s" % (ios_kind(), " ← iOS" if IS_IOS else "")) out("Python : %s" % sys.version.split()[0]) out("工作目录 : %s" % os.getcwd()) out("数据目录 : %s" % store.dir) out(" 可写 : %s" % ("是" if os.access(store.dir, os.W_OK) else "否")) out(" 数据库 : %s" % store.db_path) out(" 图片目录 : %s" % store.img_dir) try: st = os.statvfs(store.dir) out(" 剩余空间 : %s" % human_size(st.f_bavail * st.f_frsize)) except Exception: pass out("标准库 : sqlite3 ✓ ssl ✓ urllib ✓ → 零第三方依赖") out("配置文件 : %s %s" % (cfg_path(store.dir), "(已存在)" if cfg else "(未创建)")) if cfg: safe = dict(cfg) for k in ("tg_token",): if safe.get(k): safe[k] = safe[k][:6] + "***" out("配置内容 : %s" % json.dumps(safe, ensure_ascii=False)) hr("数据库") s = stats_dict(store) out("文章 %d 篇 | 索引 %d 条 | 图片 %d 张(%s)" % (s["posts"], s["index"], s["img_ok"], human_size(s["img_size"]))) return 0 def cmd_sync(a, store, net, cfg): hr("同步索引") n = sync_index(net, store, quiet=a.quiet) out("完成:%d 条待抓" % len(store.pending())) return 0 def cmd_crawl(a, store, net, cfg): cfg = dict(cfg) limit = a.limit if a.limit is not None else cfg.get("limit", 20) with_img = a.with_img or cfg.get("with_img", False) if a.no_img: with_img = False threads = a.threads or cfg.get("threads", 4) img_limit = a.img_limit if a.img_limit is not None else cfg.get("img_limit", 0) ensure_index(net, store, refresh=a.refresh_index, quiet=a.quiet) # 指定 URL / ID if a.ids or a.urls_file: targets = urls_from_ids(a.ids) if a.urls_file and os.path.exists(a.urls_file): with open(a.urls_file, "r", encoding="utf-8", errors="replace") as f: for line in f: targets += urls_from_ids(line) targets = list(dict.fromkeys(targets)) if a.limit and a.limit > 0: targets = targets[:a.limit] else: targets = [u for u, _ in store.pending(limit=limit, since=a.since, force=a.force, order=a.order)] if not targets: out("没有待抓的文章 —— 索引里都是已抓过的。") out("想重抓加 --force;想换范围加 --since 2026-10-01;想指定篇目加 --ids 1001409") return 0 hr("采集") out("目标 %d 篇%s" % (len(targets), " (含下图)" if with_img else "")) ok = fail = 0 t0 = time.time() new_posts = [] for i, u in enumerate(targets, 1): okk, msg, d = crawl_one(net, store, u, with_img=with_img, img_limit=img_limit, threads=threads, quiet=a.quiet) if okk: ok += 1 if d: new_posts.append(d) out("[%d/%d] %s" % (i, len(targets), msg)) else: fail += 1 out("[%d/%d] 失败 %s → %s" % (i, len(targets), u, msg)) if a.pause and i < len(targets): time.sleep(a.pause) hr("结果") out("成功 %d | 失败 %d | 耗时 %.1fs" % (ok, fail, time.time() - t0)) s = stats_dict(store) out("库内共 %d 篇 | 图片 %d 张(%s)" % (s["posts"], s["img_ok"], human_size(s["img_size"]))) # 通知 if new_posts and (a.notify or cfg.get("notify")): title = "黄瓜网更新 %d 篇" % len(new_posts) lines = [title, ""] for d in new_posts[:8]: lines.append("· %s" % d["title"][:44]) lines.append(" %s" % d["url"]) if len(new_posts) > 8: lines.append("… 另有 %d 篇" % (len(new_posts) - 8)) text = "\n".join(lines) ios_notify("黄瓜网采集", title) tok = a.tg_token or cfg.get("tg_token") chat = a.tg_chat or cfg.get("tg_chat") if tok and chat: okk, m = tg_send(tok, chat, text) out("TG 推送: %s" % ("成功" if okk else "失败 " + m)) wh = a.webhook or cfg.get("webhook") if wh: okk, m = webhook_send(wh, {"count": len(new_posts), "posts": [{"title": d["title"], "url": d["url"]} for d in new_posts]}) out("Webhook: %s" % ("成功" if okk else "失败 " + m)) return 0 def cmd_imgs(a, store, net, cfg): threads = a.threads or cfg.get("threads", 4) urls = store.posts_missing_images(limit=a.limit or 0) if not urls: out("没有缺图的文章,都齐了。") return 0 hr("补图") out("需要补图的文章:%d 篇" % len(urls)) total_ok = total_fail = 0 for i, u in enumerate(urls, 1): r = store.get_post(u) imgs = [x["url"] for x in store.images_of(u) if not x["ok"]] if not imgs: # 库内没记录 → 重新解析一次拿图 try: d = parse_article(net.text(u), u) imgs = d["images"] store.save_post(d) except Exception as e: out("[%d/%d] 解析失败 %s" % (i, len(urls), e)) continue o, f = download_images(net, store, u, imgs, limit=a.img_limit or 0, threads=threads, quiet=a.quiet) total_ok += o total_fail += f out("[%d/%d] %s 成%d/败%d" % (i, len(urls), (r["title"] or u)[:30], o, f)) hr("结果") out("补图完成:成功 %d 张 | 失败 %d 张" % (total_ok, total_fail)) return 0 def cmd_export(a, store, net, cfg): fmt = (a.fmt or "json").lower() hr("导出") if fmt == "json": p = a.out or os.path.join(store.dir, "hguaw_%s.json" % today()) n = export_json(store, p, limit=a.limit or 0) elif fmt == "csv": p = a.out or os.path.join(store.dir, "hguaw_%s.csv" % today()) n = export_csv(store, p, limit=a.limit or 0) elif fmt == "md": p = a.out or store.md_dir n, p = export_md(store, store.md_dir, limit=a.limit or 0, one_file=a.one_file) else: out("不认识的格式:%s(可选 json / csv / md)" % fmt) return 1 sz = human_size(os.path.getsize(p)) if os.path.isfile(p) else "多文件" out("%s 导出完成:%d 条 → %s" % (fmt.upper(), n, p)) out("大小:%s" % sz) return 0 def stats_dict(store): c = store.conn d = {} d["posts"] = c.execute("SELECT COUNT(*) FROM posts").fetchone()[0] d["index"] = store.index_total() d["pending"] = len(store.pending()) d["img_total"] = c.execute("SELECT COUNT(*) FROM images").fetchone()[0] d["img_ok"] = c.execute("SELECT COUNT(*) FROM images WHERE ok=1").fetchone()[0] d["img_size"] = c.execute("SELECT IFNULL(SUM(size),0) FROM images WHERE ok=1").fetchone()[0] r = c.execute("SELECT MIN(pubdate),MAX(pubdate) FROM posts WHERE pubdate<>''").fetchone() d["date_min"], d["date_max"] = (r[0] or "")[:10], (r[1] or "")[:10] d["chars"] = c.execute("SELECT IFNULL(SUM(content_len),0) FROM posts").fetchone()[0] return d def cmd_stats(a, store, net, cfg): s = stats_dict(store) hr("统计") out("文章 : %d 篇" % s["posts"]) out("索引 : %d 条(未抓 %d)" % (s["index"], s["pending"])) out("正文字数 : %s 字" % format(s["chars"], ",")) out("图片 : %d 张已下载 / %d 条记录,共 %s" % (s["img_ok"], s["img_total"], human_size(s["img_size"]))) out("日期范围 : %s ~ %s" % (s["date_min"] or "-", s["date_max"] or "-")) hr("分类 TOP") for r in store.conn.execute( "SELECT category, COUNT(*) n FROM posts WHERE category<>'' " "GROUP BY category ORDER BY n DESC LIMIT 12"): out(" %-14s %d" % (r["category"][:14], r["n"])) hr("最近入库") for r in store.conn.execute("SELECT title,pubdate FROM posts ORDER BY pubdate DESC LIMIT 8"): out(" %s %s" % ((r["pubdate"] or "")[:10], (r["title"] or "")[:40])) return 0 def cmd_search(a, store, net, cfg): kw = " ".join(a.keyword or []).strip() if not kw: out("给个关键词,例:search 裸贷") return 1 like = "%" + kw + "%" rows = list(store.conn.execute( """SELECT url,title,category,pubdate,tags,content_text FROM posts WHERE title LIKE ? OR tags LIKE ? OR content_text LIKE ? ORDER BY pubdate DESC LIMIT ?""", (like, like, like, a.limit or 20))) hr("搜索:%s" % kw) if not rows: out("没搜到。") return 0 out("命中 %d 篇(显示前 %d)" % (len(rows), len(rows))) for r in rows: txt = r["content_text"] or "" pos = txt.find(kw) snippet = "" if pos >= 0: snippet = re.sub(r"\s+", " ", txt[max(0, pos - 40):pos + 80]) out("") out("· %s" % (r["title"] or "")[:52]) out(" %s | %s" % ((r["pubdate"] or "")[:10], r["category"] or "-")) out(" %s" % r["url"]) if snippet: out(" …%s…" % snippet) if a.export: export_json(store, a.export, limit=0) out("\n已导出全部到 %s" % a.export) return 0 def cmd_clean(a, store, net, cfg): c = store.conn hr("清理" + ("(预演,不实际删除)" if a.dry_run else "")) # 1. 重复标题(同标题保留 pubdate 最新的一条) dup = list(c.execute(""" SELECT title, COUNT(*) n, GROUP_CONCAT(url) urls FROM posts WHERE title<>'' GROUP BY title HAVING n>1 ORDER BY n DESC""")) out("重复标题:%d 组" % len(dup)) for d in dup[:10]: out(" ×%d %s" % (d["n"], (d["title"] or "")[:40])) # 2. 孤儿图片记录(文章已不存在) orphan = c.execute("""SELECT COUNT(*) FROM images i WHERE NOT EXISTS(SELECT 1 FROM posts p WHERE p.url=i.post_url)""").fetchone()[0] out("孤儿图片记录:%d 条" % orphan) # 3. 空记录(没标题也没正文) empty = c.execute("SELECT COUNT(*) FROM posts WHERE IFNULL(title,'')='' " "AND IFNULL(content_text,'')=''").fetchone()[0] out("空记录:%d 条" % empty) # 4. 下载失败残留的 .part / 空文件 junk = 0 for root, _dirs, files in os.walk(store.img_dir): for fn in files: p = os.path.join(root, fn) try: if fn.endswith(".part") or fn.endswith(".tmp_dl") or os.path.getsize(p) == 0: junk += 1 except Exception: pass out("垃圾文件(.part/空):%d 个" % junk) if a.dry_run: out("\n预演结束。去掉 --dry-run 就真删。") return 0 removed = 0 for d in dup: urls = d["urls"].split(",") rows = list(c.execute("SELECT url,pubdate FROM posts WHERE title=? ORDER BY pubdate DESC", (d["title"],))) for r in rows[1:]: c.execute("DELETE FROM posts WHERE url=?", (r["url"],)) removed += 1 c.execute("""DELETE FROM images WHERE NOT EXISTS( SELECT 1 FROM posts p WHERE p.url=images.post_url)""") c.execute("DELETE FROM posts WHERE IFNULL(title,'')='' AND IFNULL(content_text,'')=''") n_junk = 0 for root, _dirs, files in os.walk(store.img_dir): for fn in files: p = os.path.join(root, fn) try: if fn.endswith(".part") or fn.endswith(".tmp_dl") or os.path.getsize(p) == 0: os.remove(p) n_junk += 1 except Exception: pass c.commit() if a.vacuum: c.execute("VACUUM") c.commit() out("已 VACUUM 压缩数据库") out("清理完成:删重复 %d 篇 | 垃圾文件 %d 个" % (removed, n_junk)) return 0 def cmd_watch(a, store, net, cfg): interval = a.interval or cfg.get("interval", 600) limit = a.limit if a.limit is not None else cfg.get("limit", 10) with_img = (not a.no_img) and (a.with_img or cfg.get("with_img", True)) threads = a.threads or cfg.get("threads", 4) rounds = a.rounds or 0 hr("监控") out("间隔 %d 秒 | 每轮最多 %d 篇 | %s | %s" % (interval, limit, "带下图" if with_img else "只抓数据", "跑 %d 轮就停" % rounds if rounds else "一直跑,Ctrl-C 停")) if IS_IOS: out("提示:iOS 上要先让这个脚本在前台跑着(切后台会被系统挂起)") r = 0 while True: r += 1 out("\n[%s] 第 %d 轮" % (now_str(), r)) try: sync_index(net, store, quiet=True) targets = [u for u, _ in store.pending(limit=limit)] if not targets: out(" 无新文") else: out(" 发现 %d 篇新文" % len(targets)) new_posts = [] for u in targets: okk, msg, d = crawl_one(net, store, u, with_img=with_img, threads=threads, quiet=True) out(" %s" % msg) if okk and d: new_posts.append(d) if new_posts: title = "黄瓜网更新 %d 篇" % len(new_posts) lines = [title, ""] for d in new_posts[:8]: lines.append("· %s\n %s" % (d["title"][:44], d["url"])) text = "\n".join(lines) if a.notify or cfg.get("notify"): ios_notify("黄瓜网采集", title) tok = a.tg_token or cfg.get("tg_token") chat = a.tg_chat or cfg.get("tg_chat") if tok and chat: okk, m = tg_send(tok, chat, text) out(" TG 推送:%s" % ("成功" if okk else "失败 " + m)) wh = a.webhook or cfg.get("webhook") if wh: okk, m = webhook_send(wh, {"count": len(new_posts), "posts": [{"title": d["title"], "url": d["url"]} for d in new_posts]}) out(" Webhook:%s" % ("成功" if okk else "失败 " + m)) except KeyboardInterrupt: raise except Exception as e: out(" 本轮出错:%s" % e) if rounds and r >= rounds: out("\n已跑满 %d 轮,结束。" % rounds) break try: out(" 下次检查 %d 秒后…" % interval) time.sleep(interval) except KeyboardInterrupt: raise return 0 def cmd_cfg(a, store, net, cfg): argv = sys.argv[1:] def given(flag): return any(x == flag or x.startswith(flag + "=") for x in argv) changed = False for k, flag in (("tg_token", "--tg-token"), ("tg_chat", "--tg-chat"), ("webhook", "--webhook"), ("interval", "--interval"), ("threads", "--threads"), ("limit", "--limit"), ("img_limit", "--img-limit")): if not given(flag): continue v = getattr(a, k, None) if v is not None: cfg[k] = v changed = True if a.notify: cfg["notify"] = True changed = True if a.with_img: cfg["with_img"] = True changed = True if a.no_img: cfg["with_img"] = False changed = True if changed: save_cfg(store.dir, cfg) out("已保存配置 → %s" % cfg_path(store.dir)) else: out("(没改动,只是看一眼)") hr("当前配置") safe = dict(cfg) if safe.get("tg_token"): safe["tg_token"] = safe["tg_token"][:6] + "***" out(json.dumps(safe, ensure_ascii=False, indent=2) if safe else "(空)") out("") out("改配置示例: python3 hguaw_ios.py cfg --tg-token --tg-chat <你的id>") return 0 # ───────────────────────────────────────────────────────────── # CLI # ───────────────────────────────────────────────────────────── COMMANDS = ["info", "sync", "crawl", "imgs", "export", "stats", "search", "clean", "watch", "cfg"] def add_globals(p, suppress=False): """给解析器挂全局参数。suppress=True 时用 SUPPRESS 默认值, 这样放在子命令后的同名参数不会覆盖前面的值。""" d = argparse.SUPPRESS if suppress else None def dd(fallback): return d if suppress else fallback p.add_argument("--dir", dest="datadir", default=dd(None), metavar="PATH", help="数据目录(默认 iOS 文档目录下的 hguaw_data)") p.add_argument("--ua", dest="ua", default=dd(None), help="User-Agent(传 ios 用移动端 UA)") p.add_argument("--timeout", dest="timeout", type=int, default=dd(30), help="超时秒数,默认 30") p.add_argument("--retries", dest="retries", type=int, default=dd(2), help="失败重试次数,默认 2") p.add_argument("--threads", dest="threads", type=int, default=dd(4), help="下图线程数,默认 4") p.add_argument("--delay", dest="delay", type=float, default=dd(0.0), help="每次请求间隔秒,默认 0") p.add_argument("--insecure", dest="insecure", action="store_true", default=dd(False), help="跳过证书校验(iSH 等环境证书不全时用)") p.add_argument("--quiet", dest="quiet", action="store_true", default=dd(False), help="少输出") def build_parser(): common = argparse.ArgumentParser(add_help=False) add_globals(common, suppress=True) p = argparse.ArgumentParser( prog="hguaw_ios.py", description="黄瓜网采集工具 · iOS 版(零依赖)", epilog="例:hguaw_ios.py crawl -n 20 --with-img | hguaw_ios.py watch --interval 600", formatter_class=argparse.RawDescriptionHelpFormatter, ) add_globals(p, suppress=False) p.add_argument("-v", "--version", action="version", version="hguaw_ios.py " + VERSION) sub = p.add_subparsers(dest="cmd", metavar="命令") sp = sub.add_parser("info", parents=[common], help="环境自检") sp.set_defaults(func=cmd_info) sp = sub.add_parser("sync", parents=[common], help="刷新文章索引") sp.set_defaults(func=cmd_sync) sp = sub.add_parser("crawl", parents=[common], help="采集文章") sp.add_argument("-n", "--limit", type=int, default=None, help="最多抓几篇(0=不限)") sp.add_argument("--since", default="", metavar="YYYY-MM-DD", help="只抓该日期之后更新的") sp.add_argument("--ids", default="", help="指定文章 ID 或 URL,逗号分隔") sp.add_argument("--urls-file", default="", help="从文件读 URL/ID,一行一个") sp.add_argument("--force", action="store_true", help="已抓过的也重抓") sp.add_argument("--order", choices=["new", "old"], default="new", help="抓取顺序") sp.add_argument("--with-img", action="store_true", help="同时下载配图") sp.add_argument("--no-img", action="store_true", help="强制不下图") sp.add_argument("--img-limit", type=int, default=None, help="每篇最多下几张图(0=全部)") sp.add_argument("--refresh-index", action="store_true", help="抓之前先刷新索引") sp.add_argument("--pause", type=float, default=0, help="每篇之间停几秒") sp.add_argument("--notify", action="store_true", help="完成后发系统通知") sp.add_argument("--tg-token", dest="tg_token", default=None, help="TG Bot token") sp.add_argument("--tg-chat", dest="tg_chat", default=None, help="TG 接收 chat id") sp.add_argument("--webhook", default=None, help="抓完 POST 到这个地址") sp.set_defaults(func=cmd_crawl) sp = sub.add_parser("imgs", parents=[common], help="给缺图的文章补图") sp.add_argument("-n", "--limit", type=int, default=None, help="最多几篇") sp.add_argument("--img-limit", type=int, default=None, help="每篇最多几张") sp.set_defaults(func=cmd_imgs) sp = sub.add_parser("export", parents=[common], help="导出 json / csv / md") sp.add_argument("--fmt", default="json", choices=["json", "csv", "md"]) sp.add_argument("-o", "--out", default=None, help="输出路径") sp.add_argument("-n", "--limit", type=int, default=None, help="只导前 N 篇") sp.add_argument("--one-file", dest="one_file", default="", help="md 全并成一个文件") sp.set_defaults(func=cmd_export) sp = sub.add_parser("stats", parents=[common], help="看采集进度") sp.set_defaults(func=cmd_stats) sp = sub.add_parser("search", parents=[common], help="全文搜索") sp.add_argument("keyword", nargs="*", help="关键词") sp.add_argument("-n", "--limit", type=int, default=20, help="显示条数") sp.add_argument("--export", default="", help="同时导出 json 到指定文件") sp.set_defaults(func=cmd_search) sp = sub.add_parser("clean", parents=[common], help="查重 / 清孤儿 / 清垃圾") sp.add_argument("--dry-run", action="store_true", help="只看不动手") sp.add_argument("--vacuum", action="store_true", help="顺带压缩数据库") sp.set_defaults(func=cmd_clean) sp = sub.add_parser("watch", parents=[common], help="定时监控,有新文就抓") sp.add_argument("--interval", type=int, default=None, help="间隔秒数,默认 600") sp.add_argument("-n", "--limit", type=int, default=None, help="每轮最多几篇") sp.add_argument("--rounds", type=int, default=0, help="跑几轮就停(0=一直跑)") sp.add_argument("--with-img", action="store_true", help="带下图") sp.add_argument("--no-img", action="store_true", help="只抓数据") sp.add_argument("--notify", action="store_true", help="发系统通知") sp.add_argument("--tg-token", dest="tg_token", default=None) sp.add_argument("--tg-chat", dest="tg_chat", default=None) sp.add_argument("--webhook", default=None) sp.set_defaults(func=cmd_watch) sp = sub.add_parser("cfg", parents=[common], help="写/看配置(推送、线程等)") sp.add_argument("--tg-token", dest="tg_token", default=None) sp.add_argument("--tg-chat", dest="tg_chat", default=None) sp.add_argument("--webhook", default=None) sp.add_argument("--interval", type=int, default=None) sp.add_argument("--limit", type=int, default=None) sp.add_argument("--img-limit", type=int, default=None) sp.add_argument("--notify", action="store_true") sp.add_argument("--with-img", action="store_true") sp.add_argument("--no-img", action="store_true") sp.set_defaults(func=cmd_cfg) return p def main(argv=None): argv = list(sys.argv[1:] if argv is None else argv) if argv and argv[0] in ("source", "app", "spider"): return spider_cli(argv[1:]) parser = build_parser() a = parser.parse_args(argv) if not getattr(a, "cmd", None): parser.print_help() hr("iOS 上怎么用") out("1) 先自检: python3 hguaw_ios.py info") out("2) 建索引: python3 hguaw_ios.py sync") out("3) 抓 20 篇: python3 hguaw_ios.py crawl -n 20 --with-img") out("4) 看进度: python3 hguaw_ios.py stats") out("") out("手机上建议:抓的篇数别一次开太大,10~50 篇一批,跑完再接着跑。") return 0 # 全局参数补默认(子命令里没给就用顶层默认,顶层用 SUPPRESS 时不会互相覆盖) for k, v in (("ua", None), ("timeout", 30), ("retries", 2), ("threads", 4), ("delay", 0.0), ("insecure", False), ("quiet", False)): if not hasattr(a, k): setattr(a, k, v) datadir = getattr(a, "datadir", None) or default_datadir() datadir = os.path.abspath(os.path.expanduser(datadir)) try: os.makedirs(datadir, exist_ok=True) except Exception as e: out("数据目录建不了:%s (%s)" % (datadir, e)) return 2 ua = a.ua or DEFAULT_UA if ua == "ios": ua = UA_IOS net = Net(ua=ua, timeout=a.timeout, retries=a.retries, insecure=a.insecure, delay=a.delay, quiet=a.quiet) store = Store(datadir) cfg = load_cfg(datadir) try: return a.func(a, store, net, cfg) except KeyboardInterrupt: out("\n已中断。进度都存好了,下次接着跑。") return 130 except urllib.error.HTTPError as e: out("网络错误:HTTP %s %s" % (e.code, e.reason)) return 1 except urllib.error.URLError as e: out("网络不通:%s" % e.reason) out("如果是证书问题,加 --insecure 再试") return 1 finally: store.close() # ───────────────────────────────────────────────────────────── # App 接口层 —— 给 iOS 上「Python 脚本源」类 App 用的标准入口 # ───────────────────────────────────────────────────────────── # 那类 App 会把脚本当模块导入,再找模块里的 Spider 类,依次调用 # homeContent / categoryContent / detailContent / searchContent / # playerContent 取数据。下面按这个约定实现,并兼容一批常见别名。 # 命令行那一套(sync / crawl / stats ...)完全不受影响。 SPIDER_RETURN_JSON_STRING = False # 有的 App 只收字符串不收 dict:改成 True SPIDER_UA = UA_IOS # 手机上跑就用移动端 UA NAV_CATEGORIES = [ ("recent", "今日鲜瓜"), ("rebang", "黄瓜热榜"), ("aiduanju", "AI短剧"), ("baoliao", "吃瓜爆料"), ("hot", "热门爆料"), ("wanghong", "网红吃瓜"), ("mingxing", "明星吃瓜"), ("campus", "校园话题"), ("haiwai", "海外吃瓜"), ("guanchang", "官场秘闻"), ("shehui", "社会奇闻"), ("yingshi", "影视专区"), ("chengren", "成人影视"), ("juqing", "国产剧情"), ("lunli", "伦理道德"), ("wanghuang", "网黄专区"), ("whjx", "网黄精选"), ("tanhua", "探花精选"), ("aimingxing", "AI明星"), ("simi", "私密精选"), ("tese", "特色专区"), ("pincha", "黄瓜品茶"), ("cabian", "擦边专区"), ("taose", "桃色图集"), ("dasai", "每日大赛"), ("tougao", "瓜友投稿"), ] VIDEO_RE = re.compile( r"""https?://[^\s"'<>\\]+?\.(?:m3u8|mp4|m4v|flv)(?:\?[^\s"'<>\\]*)?""", re.I) MEDIA_RE = re.compile( r"\.(?:m3u8|mp4|m4v|flv|mp3|m4a|jpg|jpeg|png|webp|gif|bmp)(?:\?|$)", re.I) class JSONResult(dict): """既是 dict 又能直接 str 成 JSON —— 两种 App 都能吃。""" def __str__(self): return json.dumps(self, ensure_ascii=False) def __bytes__(self): return str(self).encode("utf-8") def _lr(name): return re.sub(r"[^a-z0-9\u4e00-\u9fff]", "", (name or "").lower()) _NAME_MAP = {} for _grp, _names in ( ("home", ("home", "homecontent", "gethome", "gethomecontent", "homepage", "index", "first", "首页", "主页")), ("homevod", ("homevideo", "homevideocontent", "homevod", "gethomevideo", "gethomevod", "homevideos", "首页视频")), ("category", ("category", "categorycontent", "getcategory", "getcategorycontent", "class", "classes", "list", "getlist", "typecontent", "classcontent", "type", "gettypecontent", "typelist", "分类", "列表")), ("detail", ("detail", "detailcontent", "getdetail", "getdetailcontent", "info", "getinfo", "详情", "详情页")), ("search", ("search", "searchcontent", "getsearch", "getsearchcontent", "find", "搜索")), ("play", ("play", "player", "playercontent", "playcontent", "getplay", "getplayer", "getplayurl", "playurl", "播放")), ("init", ("init", "initialize", "setup", "start", "load", "构造", "初始化")), ): for _n in _names: _NAME_MAP[_lr(_n)] = _grp class Spider: """黄瓜网 —— App 侧标准入口(兼容 TVBox 系 python 源约定)。""" name = "黄瓜网" title = "黄瓜网" version = VERSION api = BASE site = BASE host = BASE home_url = BASE + "/" def __init__(self, *a, **kw): self.net = Net(ua=SPIDER_UA, referer=DEFAULT_REFERER, timeout=20, retries=1, quiet=True) self._cache = {} self.extend = kw.get("extend", "") or (a[0] if a else "") # ── 取页面(带 3 分钟内存缓存,翻页来回切不重复请求)── def _text(self, url): now = time.time() hit = self._cache.get(url) if hit and now - hit[0] < 180: return hit[1] raw = self.net.text(url, referer=DEFAULT_REFERER) if len(self._cache) > 60: self._cache.clear() self._cache[url] = (now, raw) return raw def _wrap(self, obj): if SPIDER_RETURN_JSON_STRING: return json.dumps(obj, ensure_ascii=False) return JSONResult(obj) # ── 列表页解析 ── def _parse_list(self, raw, limit=0): doc = Doc(raw) out, seen = [], set() for art in doc.find_all("article", "card"): link = doc.first("a", "card-title-link", art) if link is None: link = doc.first("a", "card-cover-link", art) if link is None: continue href = abs_url(link.get("href"), BASE) if not href or href in seen: continue seen.add(href) title = "" t = doc.first("a", "card-title-link", art) if t is not None: title = doc.text(t) img = doc.first("img", None, art) if not title and img is not None: title = img.get("alt") or "" if not title: title = href pic = "" if img is not None: pic = img.get("data-original-src") or img.get("src") or "" if not pic: src = doc.first("source", None, art) if src is not None: ss = (src.get("srcset") or "").split(",")[0].strip().split(" ") if ss: pic = ss[0] pic = abs_url(pic) badge = doc.first("span", "card-badge", art) remark = doc.text(badge) if badge is not None else "" dt = doc.first("time", "card-date", art) date = "" if dt is not None: date = (dt.get("datetime") or doc.text(dt) or "").strip() if date: remark = (remark + " · " + date[:10]).strip(" ·") out.append({ "vod_id": href.replace(BASE, "") or href, "vod_name": clean_text(title), "vod_pic": pic, "vod_remarks": remark, "vod_url": href, }) if limit and len(out) >= limit: break return out def _list(self, url, limit=0): return self._parse_list(self._text(url), limit) def _pagecount(self, url, raw=None): raw = raw if raw is not None else self._text(url) i = raw.find('class="pagination') seg = raw[i:i + 5000] if i >= 0 else raw nums = [int(x) for x in re.findall(r'href="[^"]*?/(\d+)/"', seg)] return max(nums) if nums else 1 def _page_url(self, url, pg): # 分类分页是 /category/xx/2/,搜索分页是 /search/kw/page/2/ url = url.rstrip("/") + "/" try: pg = int(pg or 1) except Exception: pg = 1 if pg <= 1: return url if "/search/" in url: return url + "page/%d/" % pg return url + "%d/" % pg # ── 标准接口 ── def init(self, *a, **kw): return True def homeContent(self, *a, **kw): classes = [{"type_id": s, "type_name": n} for s, n in NAV_CATEGORIES] vids = [] for url in (BASE + "/", BASE + "/category/recent/"): try: vids = self._list(url, limit=60) except Exception: vids = [] if vids: break return self._wrap({"class": classes, "filters": {}, "list": vids}) def homeVideoContent(self, *a, **kw): vids = [] for url in (BASE + "/", BASE + "/category/recent/"): try: vids = self._list(url, limit=60) except Exception: vids = [] if vids: break return self._wrap({"list": vids}) def categoryContent(self, tid=None, pg=1, *a, **kw): # 有的 App 会多塞参数(filter / extend),按位置兜一下 args = [x for x in a] if tid is not None and str(tid).isdigit() and not args: pass if tid is None and args: tid = args.pop(0) for extra in list(args): if isinstance(extra, (int, float, str)) and str(extra).isdigit(): pg = extra args.remove(extra) break slug = str(tid or "recent").strip().strip("/") if slug.startswith("http"): url = slug elif "/" in slug: url = BASE + "/" + slug else: url = BASE + "/category/" + slug + "/" url = self._page_url(url, pg) raw = self._text(url) lst = self._parse_list(raw) return self._wrap({"page": int(pg) if str(pg).isdigit() else 1, "pagecount": self._pagecount(url, raw), "limit": len(lst), "total": len(lst), "list": lst}) def detailContent(self, ids=None, *a, **kw): if ids is None: ids = list(a) if isinstance(ids, (str, int)): ids = [ids] out = [] for vid in list(ids)[:10]: vid = str(vid) url = vid if vid.startswith("http") else BASE + ("" if vid.startswith("/") else "/") + vid try: raw = self._text(url) d = parse_article(raw, url) except Exception: continue images = list(d.get("images") or []) vids = [] for m in VIDEO_RE.finditer(raw): u = m.group(0).replace("\\/", "/") if u not in vids: vids.append(u) if vids: play_from = "视频" play_url = "#".join( "正片$%s" % u if i == 0 else "第%d段$%s" % (i + 1, u) for i, u in enumerate(vids[:20])) elif images: play_from = "图集" play_url = "#".join("图%d$%s" % (i + 1, u) for i, u in enumerate(images[:60])) else: play_from = "原文" play_url = "打开原文$%s" % url content = (d.get("content_text") or "").strip() if len(content) > 600: content = content[:600] + "…" head = "分类:%s　作者:%s　日期:%s" % ( d.get("category") or "-", d.get("author") or "-", (d.get("pubdate") or "-")[:10]) if d.get("tags"): head += "　标签:%s" % "、".join(d["tags"][:8]) out.append({ "vod_id": vid, "vod_name": d.get("title") or url, "vod_pic": d.get("cover") or (images[0] if images else ""), "vod_content": (head + "\n\n" + content).strip(), "vod_remarks": (d.get("pubdate") or "")[:10], "vod_year": (d.get("pubdate") or "")[:4], "vod_area": d.get("category") or "", "vod_actor": d.get("author") or "", "type_name": d.get("category") or "", "vod_play_from": play_from, "vod_play_url": play_url, "vod_pic_list": images[:60], }) return self._wrap({"list": out}) def searchContent(self, key=None, quick=False, pg=1, *a, **kw): kwd = key for extra in (a if key is not None else ()): if isinstance(extra, str) and extra.strip(): kwd = extra break if not kwd: return self._wrap({"list": []}) for extra in list(a): if isinstance(extra, (int, float)) or (isinstance(extra, str) and extra.isdigit()): pg = extra break url = BASE + "/search/" + urllib.parse.quote(str(kwd).strip()) + "/" url = self._page_url(url, pg) try: raw = self._text(url) lst = self._parse_list(raw) except Exception: return self._wrap({"list": []}) return self._wrap({"page": int(pg) if str(pg).isdigit() else 1, "pagecount": self._pagecount(url, raw), "limit": len(lst), "total": len(lst), "list": lst}) def playerContent(self, flag=None, id=None, *a, **kw): target = id if id is not None else flag if isinstance(target, (list, tuple)): target = target[0] if target else "" target = str(target or "") url = target if url.startswith("/"): url = BASE + url if not MEDIA_RE.search(url): # 给的是文章地址 → 现场取里面的视频 / 首图 try: raw = self._text(url) m = VIDEO_RE.search(raw) if m: url = m.group(0).replace("\\/", "/") else: d = parse_article(raw, url) imgs = list(d.get("images") or []) if imgs: url = imgs[0] except Exception: pass return self._wrap({"parse": 0, "playUrl": "", "url": url, "header": "Referer: %s/" % BASE}) # ── App 侧一些可有可无的钩子,给它个东西回,免得判空报错 ── def getName(self, *a, **kw): return self.name def setName(self, *a, **kw): return True def setExtend(self, *a, **kw): self.extend = (a[0] if a else "") or "" return True def getExtend(self, *a, **kw): return self.extend or "" def isVideo(self, *a, **kw): return True def isVideoFormat(self, *a, **kw): url = str(a[0] if a else "") return bool(MEDIA_RE.search(url)) def manualVideoCheck(self, *a, **kw): return False def localProxy(self, *a, **kw): url = str(a[0] if a else "") if url.startswith("/"): url = BASE + url return [200, "video/MP2T", b"", {"Location": url}] if url else [404, "", b""] def refresh(self, *a, **kw): self._cache.clear() return True def destroy(self, *a, **kw): return True # ── 别名容错 ── def __getattr__(self, name): if name.startswith("__"): raise AttributeError(name) grp = _NAME_MAP.get(_lr(name)) if grp: table = {"home": "homeContent", "homevod": "homeVideoContent", "category": "categoryContent", "detail": "detailContent", "search": "searchContent", "play": "playerContent", "init": "init"} return getattr(self, table[grp]) raise AttributeError( "Spider 没有 %s 这个接口。可用:homeContent / homeVideoContent / " "categoryContent / detailContent / searchContent / playerContent" % name) _default_spider = None def _sp(): global _default_spider if _default_spider is None: _default_spider = Spider() return _default_spider # 模块级函数(有的 App 不实例化类,直接找模块里的同名函数) def init(*a, **kw): return _sp().init(*a, **kw) def homeContent(*a, **kw): return _sp().homeContent(*a, **kw) def homeVideoContent(*a, **kw): return _sp().homeVideoContent(*a, **kw) def homeVod(*a, **kw): return _sp().homeVideoContent(*a, **kw) def getName(*a, **kw): return _sp().getName() def categoryContent(*a, **kw): return _sp().categoryContent(*a, **kw) def detailContent(*a, **kw): return _sp().detailContent(*a, **kw) def searchContent(*a, **kw): return _sp().searchContent(*a, **kw) def playerContent(*a, **kw): return _sp().playerContent(*a, **kw) spider = Spider() # 有些 App 直接拿实例 _MODULE_FUNCS = {"home": homeContent, "homevod": homeVideoContent, "category": categoryContent, "detail": detailContent, "search": searchContent, "play": playerContent, "init": init} def __getattr__(name): # PEP 562,Python 3.7+ grp = _NAME_MAP.get(_lr(name)) if grp: return _MODULE_FUNCS[grp] try: # 兜底:Spider 上的方法/属性也能从模块级直接拿 return getattr(_sp(), name) except AttributeError: pass raise AttributeError( "module hguaw_ios 没有 %s。可用:Spider / homeContent / categoryContent / " "detailContent / searchContent / playerContent" % name) def spider_cli(args): """命令行自测:python3 hguaw_ios.py source [home|cat <分类>|search <词>|detail <文章>|play ]""" what = (args[0] if args else "home").lower() sp = _sp() if what in ("home", "首页"): res = sp.homeContent() elif what in ("cat", "category") and len(args) > 1: res = sp.categoryContent(args[1], args[2] if len(args) > 2 else 1) elif what in ("search", "搜") and len(args) > 1: res = sp.searchContent(args[1]) elif what in ("detail", "详情") and len(args) > 1: res = sp.detailContent([args[1]]) elif what in ("play", "播放") and len(args) > 1: res = sp.playerContent("", args[1]) else: out("用法: source [home | cat <分类> [页] | search <关键词> | detail <文章地址> | play <地址>]") return 2 out(json.dumps(res, ensure_ascii=False, indent=2)[:4000]) return 0 if __name__ == "__main__": sys.exit(main())